/*** * A queue for requests from remote peers * NOTE: This should handle multiple threads */ #include #include "libp2p/conn/session.h" #include "ipfs/cid/cid.h" #include "ipfs/exchange/bitswap/peer_request_queue.h" #include "ipfs/exchange/bitswap/message.h" #include "ipfs/exchange/bitswap/network.h" /*** * Allocate memory for CidEntry * @returns new CidEntry struct */ struct CidEntry* ipfs_bitswap_peer_request_cid_entry_new() { struct CidEntry* entry = (struct CidEntry*) malloc(sizeof(struct CidEntry)); if (entry != NULL) { entry->cid = NULL; entry->cancel = 0; } return entry; } /** * Allocate resources for a new PeerRequest * @returns a new PeerRequest struct or NULL if there was a problem */ struct PeerRequest* ipfs_bitswap_peer_request_new() { int retVal = 0; struct PeerRequest* request = (struct PeerRequest*) malloc(sizeof(struct PeerRequest)); if (request != NULL) { request->cids_they_want = libp2p_utils_vector_new(1); if (request->cids_they_want == NULL) goto exit; request->cids_we_want = libp2p_utils_vector_new(1); if (request->cids_we_want == NULL) goto exit; request->blocks_we_want_to_send = libp2p_utils_vector_new(1); if (request->blocks_we_want_to_send == NULL) goto exit; request->peer = NULL; } retVal = 1; exit: if (retVal == 0 && request != NULL) { if (request->blocks_we_want_to_send != NULL) libp2p_utils_vector_free(request->blocks_we_want_to_send); if (request->cids_they_want != NULL) libp2p_utils_vector_free(request->cids_they_want); if (request->cids_we_want != NULL) libp2p_utils_vector_free(request->cids_we_want); free(request); request = NULL; } return request; } /** * Free resources from a PeerRequest * @param request the request to free * @returns true(1) */ int ipfs_bitswap_peer_request_free(struct PeerRequest* request) { free(request); return 1; } /** * Allocate resources for a new queue */ struct PeerRequestQueue* ipfs_bitswap_peer_request_queue_new() { struct PeerRequestQueue* queue = malloc(sizeof(struct PeerRequestQueue)); if (queue != NULL) { pthread_mutex_init(&queue->queue_mutex, NULL); queue->first = NULL; queue->last = NULL; } return queue; } /** * Free all resources related to the queue * @param queue the queue * @returns true(1) */ int ipfs_bitswap_peer_request_queue_free(struct PeerRequestQueue* queue) { pthread_mutex_lock(&queue->queue_mutex); struct PeerRequestEntry* current = queue->last; while (current != NULL) { struct PeerRequestEntry* prior = current->prior; ipfs_bitswap_peer_request_entry_free(current); current = prior; } pthread_mutex_unlock(&queue->queue_mutex); free(queue); return 1; } /** * Adds a peer request to the end of the queue * @param queue the queue * @param request the request * @returns true(1) on success, otherwise false */ int ipfs_bitswap_peer_request_queue_add(struct PeerRequestQueue* queue, struct PeerRequest* request) { if (request != NULL) { struct PeerRequestEntry* entry = ipfs_bitswap_peer_request_entry_new(); entry->current = request; pthread_mutex_lock(&queue->queue_mutex); entry->prior = queue->last; queue->last = entry; if (queue->first == NULL) { queue->first = entry; } pthread_mutex_unlock(&queue->queue_mutex); return 1; } return 0; } /** * Removes a peer request from the queue, no mather where it is * @param queue the queue * @param request the request * @returns true(1) on success, otherwise false(0) */ int ipfs_bitswap_peer_request_queue_remove(struct PeerRequestQueue* queue, struct PeerRequest* request) { if (request != NULL) { struct PeerRequestEntry* entry = ipfs_bitswap_peer_request_queue_find_entry(queue, request->peer); if (entry != NULL) { pthread_mutex_lock(&queue->queue_mutex); // remove the entry's link, and hook prior and next together entry->prior->next = entry->next; entry->prior = NULL; entry->next = NULL; ipfs_bitswap_peer_request_entry_free(entry); pthread_mutex_unlock(&queue->queue_mutex); return 1; } } return 0; } /** * Finds a PeerRequestEntry that contains the specified PeerRequest * @param queue the queue to look through * @param request what we're looking for * @returns the PeerRequestEntry or NULL if not found */ struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct Libp2pPeer* peer) { if (peer != NULL) { struct PeerRequestEntry* current = queue->first; while (current != NULL) { if (libp2p_peer_compare(current->current->peer, peer) == 0) return current; current = current->next; } } return NULL; } /*** * Determine if any of the cids in the list are waiting to be filled * @param cidEntries a Vector of CidEntry objects * @returns true(1) if we have some waiting, false(0) otherwise */ int ipfs_bitswap_peer_request_cids_waiting(struct Libp2pVector* cidEntries) { if (cidEntries == NULL) return 0; for(int i = 0; i < cidEntries->total; i++) { const struct CidEntry* entry = (const struct CidEntry*)libp2p_utils_vector_get(cidEntries, i); if (entry != NULL && !entry->cancel) return 1; } return 0; } /*** * Determine if there is something to process in this request * @param entry the entry to look at * @returns true(1) if there is something to do */ int ipfs_bitswap_peer_request_something_to_do(struct PeerRequestEntry* entry) { if (entry != NULL) { struct PeerRequest* request = entry->current; // do we have something in the queue? if (request->blocks_we_want_to_send->total > 0) return 1; if (request->cids_we_want->total > 0) return 1; if (ipfs_bitswap_peer_request_cids_waiting(request->cids_they_want)) return 1; } return 0; } /** * Pull a PeerRequest off the queue * @param queue the queue * @returns the PeerRequest that should be handled next, or NULL if the queue is empty */ struct PeerRequest* ipfs_bitswap_peer_request_queue_pop(struct PeerRequestQueue* queue) { struct PeerRequest* retVal = NULL; if (queue != NULL) { pthread_mutex_lock(&queue->queue_mutex); struct PeerRequestEntry* entry = queue->first; if (entry != NULL) { if (ipfs_bitswap_peer_request_something_to_do(entry)) { retVal = entry->current; // move to the end of the queue queue->first = queue->first->next; queue->last->next = entry; queue->last = entry; } } pthread_mutex_unlock(&queue->queue_mutex); // disable temporarily // JMJ Debugging /* if (entry != NULL) ipfs_bitswap_peer_request_entry_free(entry); */ } return retVal; } /*** * Allocate resources for a PeerRequestEntry struct * @returns the allocated struct or NULL if there was a problem */ struct PeerRequestEntry* ipfs_bitswap_peer_request_entry_new() { struct PeerRequestEntry* entry = (struct PeerRequestEntry*) malloc(sizeof(struct PeerRequestEntry)); if (entry != NULL) { entry->current = NULL; entry->next = NULL; entry->prior = NULL; } return entry; } /** * Frees resources allocated * @param entry the PeerRequestEntry to free * @returns true(1) */ int ipfs_bitswap_peer_request_entry_free(struct PeerRequestEntry* entry) { entry->next = NULL; entry->prior = NULL; ipfs_bitswap_peer_request_free(entry->current); entry->current = NULL; free(entry); return 1; } /*** * Add a block to the appropriate peer's queue * @param queue the queue * @param who the session context that identifies the peer * @param block the block * @returns true(1) on success, otherwise false(0) */ int ipfs_bitswap_peer_request_queue_fill(struct PeerRequestQueue* queue, struct Libp2pPeer* who, struct Block* block) { // find the right entry struct PeerRequest* entry = ipfs_peer_request_queue_find_peer(queue, who); if (entry != NULL) { // add to the block array libp2p_utils_vector_add(entry->blocks_we_want_to_send, block); } return 0; } /**** * Find blocks they want, and put them in the request */ int ipfs_bitswap_peer_request_get_blocks_they_want(const struct BitswapContext* context, struct PeerRequest* request) { for(int i = 0; i < request->cids_they_want->total; i++) { struct CidEntry* cidEntry = (struct CidEntry*)libp2p_utils_vector_get(request->cids_they_want, i); if (cidEntry != NULL && !cidEntry->cancel) { struct Block* block = NULL; context->ipfsNode->blockstore->Get(context->ipfsNode->blockstore->blockstoreContext, cidEntry->cid, &block); if (block != NULL) { libp2p_utils_vector_add(request->blocks_we_want_to_send, block); cidEntry->cancel = 1; } } } return 0; } /**** * Handle a PeerRequest * @param context the BitswapContext * @param request the request to process * @returns true(1) on succes, otherwise false(0) */ int ipfs_bitswap_peer_request_process_entry(const struct BitswapContext* context, struct PeerRequest* request) { // determine if we have enough information to continue if (request == NULL) return 0; if (request->peer == NULL) return 0; if (!request->peer->is_local) { if (request->peer->connection_type != CONNECTION_TYPE_CONNECTED) if (request->peer->addr_head == NULL || request->peer->addr_head->item == NULL) return 0; } // determine if we're connected int connected = request->peer->is_local || request->peer->connection_type == CONNECTION_TYPE_CONNECTED; int need_to_connect = request->cids_we_want->total != 0 || ipfs_bitswap_peer_request_cids_waiting(request->cids_they_want) || request->blocks_we_want_to_send->total != 0; // determine if we need to connect if (need_to_connect) { if (!connected) { // connect connected = libp2p_peer_connect(&context->ipfsNode->identity->private_key, request->peer, context->ipfsNode->peerstore, 0); } if (connected) { // build a message struct BitswapMessage* msg = ipfs_bitswap_message_new(); // see if we can fulfill any of their requests. If so, fill in msg->payload ipfs_bitswap_peer_request_get_blocks_they_want(context, request); ipfs_bitswap_message_add_blocks(msg, request->blocks_we_want_to_send); // add requests that we would like ipfs_bitswap_message_add_wantlist_items(msg, request->cids_we_want); // send message if (ipfs_bitswap_network_send_message(context, request->peer, msg)) return 1; } } return 0; } /*** * Find a PeerRequest related to a peer. If one is not found, it is created. * * @param peer_request_queue the queue to look through * @param peer the peer to look for * @returns a PeerRequestEntry or NULL on error */ struct PeerRequest* ipfs_peer_request_queue_find_peer(struct PeerRequestQueue* queue, struct Libp2pPeer* peer) { struct PeerRequestEntry* entry = queue->first; while (entry != NULL) { if (libp2p_peer_compare(entry->current->peer, peer) == 0) { return entry->current; } } entry = ipfs_bitswap_peer_request_entry_new(); entry->current = ipfs_bitswap_peer_request_new(); entry->current->peer = peer; entry->prior = queue->last; queue->last = entry; return entry->current; }