handle_message(peer, peer->task_received);
}
} else {
+ handle_pending_sends(peer);
if (peer->current_piece != -1) {
send_interested_to_peers(peer);
} else {
peer->comm_received = NULL;
peer->round = 0;
+
+ peer->pending_sends = xbt_dynar_new(sizeof(msg_comm_t), NULL);
}
/**
xbt_dict_free(&peer->peers);
xbt_dict_free(&peer->active_peers);
xbt_dynar_free(&peer->current_pieces);
+ xbt_dynar_free(&peer->pending_sends);
xbt_free(peer->pieces_count);
xbt_free(peer->bitfield);
xbt_free(peer->bitfield_blocks);
return ((memchr(bitfield, '0', sizeof(char) * FILE_PIECES) == NULL) ? 1 : 0);
}
+/**
+ * Handle pending sends and remove those which are done
+ * @param peer Peer data
+ */
+void handle_pending_sends(peer_t peer)
+{
+ int index;
+
+ while ((index = MSG_comm_testany(peer->pending_sends)) != -1) {
+ msg_comm_t comm_send = xbt_dynar_get_as(peer->pending_sends, index, msg_comm_t);
+ int status = MSG_comm_get_status(comm_send);
+ xbt_dynar_remove_at(peer->pending_sends, index, &comm_send);
+ XBT_DEBUG("Communication %p is finished with status %d, dynar size is now %lu", comm_send, status, xbt_dynar_length(peer->pending_sends));
+
+ msg_task_t task = MSG_comm_get_task(comm_send);
+ MSG_comm_destroy(comm_send);
+
+ if (status != MSG_OK) {
+ task_message_free(task);
+ }
+ }
+}
+
/**
* Handle a received message sent by another peer
* @param peer Peer data
msg_task_t task =
task_message_bitfield_new(peer->hostname, peer->mailbox, peer->id,
peer->bitfield, FILE_PIECES);
- MSG_task_dsend(task, mailbox, task_message_free);
+ //Async send and append to pending sends
+ msg_comm_t comm = MSG_task_isend(task, mailbox);
+ xbt_dynar_push(peer->pending_sends, &comm);
}
/**