From d969f4832465727fa8dcbecb48ceaadff2ac0cd4 Mon Sep 17 00:00:00 2001 From: John Jones Date: Mon, 31 Jul 2017 08:16:52 -0500 Subject: [PATCH] implementation of bitswap network receive --- core/null.c | 4 + exchange/bitswap/message.c | 6 +- exchange/bitswap/network.c | 96 +++++++++++++++---- exchange/bitswap/peer_request_queue.c | 8 +- include/ipfs/exchange/bitswap/message.h | 6 +- include/ipfs/exchange/bitswap/network.h | 12 ++- .../exchange/bitswap/peer_request_queue.h | 2 +- 7 files changed, 103 insertions(+), 31 deletions(-) diff --git a/core/null.c b/core/null.c index 64aa309..cce4329 100644 --- a/core/null.c +++ b/core/null.c @@ -21,6 +21,7 @@ #include "ipfs/merkledag/merkledag.h" #include "ipfs/merkledag/node.h" #include "ipfs/util/thread_pool.h" +#include "ipfs/exchange/bitswap/network.h" #define BUF_SIZE 4096 @@ -101,6 +102,9 @@ int ipfs_null_marshal(const unsigned char* incoming, size_t incoming_size, struc // this handles 1 transaction libp2p_routing_dht_handle_message(session, connection_param->local_node->peerstore, connection_param->local_node->providerstore); libp2p_logger_log("null", LOGLEVEL_DEBUG, "kademlia message handled\n"); + } else if (protocol_compare(incoming, incoming_size, "/ipfs/bitswap/")) { + libp2p_logger_debug("null", "Attempting bitswap connection...\n"); + return ipfs_bitswap_network_handle_message(connection_param->local_node, session, incoming, incoming_size); } else { libp2p_logger_error("null", "There was a problem with this connection. It is nothing I can handle. Disconnecting.\n"); diff --git a/exchange/bitswap/message.c b/exchange/bitswap/message.c index b0342a0..d39d621 100644 --- a/exchange/bitswap/message.c +++ b/exchange/bitswap/message.c @@ -464,7 +464,7 @@ int ipfs_bitswap_message_free(struct BitswapMessage* message) { * @param message the BitswapMessage * @returns the maximum size of the protobuf'd BitswapMessage */ -size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message) { +size_t ipfs_bitswap_message_protobuf_encode_size(const struct BitswapMessage* message) { size_t total = 0; if (message != NULL) { if (message->blocks != NULL) { @@ -495,7 +495,7 @@ size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message) * @param bytes_written the total number of bytes written to the buffer * @returns true(1) on success, otherwise false(0) */ -int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written) { +int ipfs_bitswap_message_protobuf_encode(const struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written) { size_t bytes_used = 0; *bytes_written = 0; @@ -558,7 +558,7 @@ int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigne * @param output the newly allocated BitswapMessage * @returns true(1) on success, otherwise false(0) */ -int ipfs_bitswap_message_protobuf_decode(unsigned char* buffer, size_t buffer_length, struct BitswapMessage** output) { +int ipfs_bitswap_message_protobuf_decode(const uint8_t* buffer, size_t buffer_length, struct BitswapMessage** output) { size_t pos = 0; *output = NULL; diff --git a/exchange/bitswap/network.c b/exchange/bitswap/network.c index 4f45896..6b7ca3f 100644 --- a/exchange/bitswap/network.c +++ b/exchange/bitswap/network.c @@ -6,23 +6,7 @@ */ #include "ipfs/exchange/bitswap/network.h" - -/*** - * The main loop - */ - -/** - * We received a BitswapMessage from the network - */ -/* -ipfs_bitswap_network_receive_message(struct BitswapContext* context) { - -} -*/ - -/** - * We want to pop something off the queue - */ +#include "ipfs/exchange/bitswap/peer_request_queue.h" /**** * send a message to a particular peer @@ -30,7 +14,81 @@ ipfs_bitswap_network_receive_message(struct BitswapContext* context) { * @param peer the peer that is the recipient * @param message the message to send */ -int ipfs_bitswap_network_send_message(const struct BitswapContext* context, const struct Libp2pPeer* peer, const struct BitswapMessage* message) { - return 0; +int ipfs_bitswap_network_send_message(const struct BitswapContext* context, struct Libp2pPeer* peer, const struct BitswapMessage* message) { + // get a connection to the peer + if (peer->connection_type != CONNECTION_TYPE_CONNECTED) { + libp2p_peer_connect(&context->ipfsNode->identity->private_key, peer, 10); + if(peer->connection_type != CONNECTION_TYPE_CONNECTED) + return 0; + } + // protobuf the message + size_t buf_size = ipfs_bitswap_message_protobuf_encode_size(message); + uint8_t* buf = (uint8_t*) malloc(buf_size + 20); + if (buf == NULL) + return 0; + if (!ipfs_bitswap_message_protobuf_encode(message, &buf[20], buf_size, &buf_size)) { + free(buf); + return 0; + } + // tack on the protocol header + memcpy(buf, "/ipfs/bitswap/1.1.0\n", 20); + buf_size += 20; + // send it + int bytes_written = peer->sessionContext->default_stream->write(peer->sessionContext->default_stream, buf, buf_size); + if (bytes_written <= 0) { + free(buf); + return 0; + } + free(buf); + return 1; } +/*** + * Handle a raw incoming bitswap message from the network + * @param node us + * @param sessionContext the connection context + * @param bytes the message + * @param bytes_size the size of the message + * @returns true(1) on success, false(0) otherwise. + */ +int ipfs_bitswap_network_handle_message(const struct IpfsNode* node, const struct SessionContext* sessionContext, const uint8_t* bytes, size_t bytes_length) { + struct BitswapContext* bitswapContext = (struct BitswapContext*)node->exchange->exchangeContext; + // strip off the protocol header + int start = -1; + for(int i = 0; i < bytes_length; i++) { + if (bytes[i] == '\n') { + start = i+1; + break; + } + } + if (start == -1) + return 0; + // un-protobuf the message + struct BitswapMessage* message = NULL; + if (!ipfs_bitswap_message_protobuf_decode(&bytes[start], bytes_length - start, &message)) + return 0; + // process the message + // payload - what we want + for(int i = 0; i < message->payload->total; i++) { + struct Block* blk = (struct Block*)libp2p_utils_vector_get(message->payload, i); + node->exchange->HasBlock(node->exchange, blk); + } + // wantlist - what they want + if (message->wantlist != NULL && message->wantlist->entries != NULL && message->wantlist->entries->total > 0) { + // get the peer + struct Libp2pPeer* peer = libp2p_peerstore_get_peer(node->peerstore, (unsigned char*)sessionContext->remote_peer_id, strlen(sessionContext->remote_peer_id)); + struct PeerRequestEntry* queueEntry = ipfs_bitswap_peer_request_queue_find_entry(bitswapContext->peerRequestQueue, peer); + for(int i = 0; i < message->wantlist->entries->total; i++) { + struct WantlistEntry* entry = (struct WantlistEntry*) libp2p_utils_vector_get(message->wantlist->entries, i); + // turn the "block" back into a cid + struct Cid* cid = NULL; + if (!ipfs_cid_protobuf_decode(entry->block, entry->block_size, &cid)) { + ipfs_cid_free(cid); + return 0; + } + // add the cid to their queue + libp2p_utils_vector_add(queueEntry->current->cids, cid); + } + } + return 1; +} diff --git a/exchange/bitswap/peer_request_queue.c b/exchange/bitswap/peer_request_queue.c index ef4cb94..3cbc27c 100644 --- a/exchange/bitswap/peer_request_queue.c +++ b/exchange/bitswap/peer_request_queue.c @@ -95,7 +95,7 @@ int ipfs_bitswap_peer_request_queue_add(struct PeerRequestQueue* queue, struct P */ 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); + 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 @@ -116,11 +116,11 @@ int ipfs_bitswap_peer_request_queue_remove(struct PeerRequestQueue* queue, struc * @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 PeerRequest* request) { - if (request != NULL) { +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, request->peer) == 0) + if (libp2p_peer_compare(current->current->peer, peer) == 0) return current; current = current->next; } diff --git a/include/ipfs/exchange/bitswap/message.h b/include/ipfs/exchange/bitswap/message.h index 721b58f..012be47 100644 --- a/include/ipfs/exchange/bitswap/message.h +++ b/include/ipfs/exchange/bitswap/message.h @@ -180,7 +180,7 @@ int ipfs_bitswap_message_free(struct BitswapMessage* message); * @param message the BitswapMessage * @returns the maximum size of the protobuf'd BitswapMessage */ -size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message); +size_t ipfs_bitswap_message_protobuf_encode_size(const struct BitswapMessage* message); /*** * Encode a BitswapMessage into a protobuf buffer @@ -190,7 +190,7 @@ size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message) * @param bytes_written the total number of bytes written to the buffer * @returns true(1) on success, otherwise false(0) */ -int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written); +int ipfs_bitswap_message_protobuf_encode(const struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written); /*** * Decode a BitswapMessage from a protobuf @@ -199,7 +199,7 @@ int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigne * @param output the newly allocated BitswapMessage * @returns true(1) on success, otherwise false(0) */ -int ipfs_bitswap_message_protobuf_decode(unsigned char* buffer, size_t buffer_length, struct BitswapMessage** output); +int ipfs_bitswap_message_protobuf_decode(const uint8_t* buffer, size_t buffer_length, struct BitswapMessage** output); /**** * Add a vector of Cids to the bitswap message diff --git a/include/ipfs/exchange/bitswap/network.h b/include/ipfs/exchange/bitswap/network.h index 10f2013..d49b9ca 100644 --- a/include/ipfs/exchange/bitswap/network.h +++ b/include/ipfs/exchange/bitswap/network.h @@ -70,4 +70,14 @@ struct BitswapNetwork { * @param peer the peer that is the recipient * @param message the message to send */ -int ipfs_bitswap_network_send_message(const struct BitswapContext* context, const struct Libp2pPeer* peer, const struct BitswapMessage* message); +int ipfs_bitswap_network_send_message(const struct BitswapContext* context, struct Libp2pPeer* peer, const struct BitswapMessage* message); + +/*** + * Handle a raw incoming bitswap message from the network + * @param node us + * @param sessionContext the connection context + * @param bytes the message + * @param bytes_size the size of the message + * @returns true(1) on success, false(0) otherwise. + */ +int ipfs_bitswap_network_handle_message(const struct IpfsNode* node, const struct SessionContext* sessionContext, const uint8_t* bytes, size_t bytes_length); diff --git a/include/ipfs/exchange/bitswap/peer_request_queue.h b/include/ipfs/exchange/bitswap/peer_request_queue.h index 2386df4..d83c754 100644 --- a/include/ipfs/exchange/bitswap/peer_request_queue.h +++ b/include/ipfs/exchange/bitswap/peer_request_queue.h @@ -83,7 +83,7 @@ struct PeerRequest* ipfs_bitswap_peer_request_queue_pop(struct PeerRequestQueue* * @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 PeerRequest* request); +struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct Libp2pPeer* peer); /*** * Add a block to the appropriate peer's queue