c-ipfs/core/null.c

273 lines
9.7 KiB
C
Raw Normal View History

2017-02-10 01:10:21 +00:00
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
2017-02-10 01:10:21 +00:00
#include <unistd.h>
#include <pthread.h>
2017-04-04 01:54:03 +00:00
#include <netinet/in.h>
#include <arpa/inet.h>
2017-07-27 19:33:19 +00:00
#include "libp2p/conn/session.h"
#include "libp2p/net/multistream.h"
2017-02-10 01:10:21 +00:00
#include "libp2p/net/p2pnet.h"
#include "libp2p/net/protocol.h"
2017-07-27 19:33:19 +00:00
#include "libp2p/nodeio/nodeio.h"
#include "libp2p/record/message.h"
2017-07-27 19:33:19 +00:00
#include "libp2p/routing/dht_protocol.h"
#include "libp2p/secio/secio.h"
2017-04-03 18:20:35 +00:00
#include "libp2p/utils/logger.h"
2017-02-10 01:10:21 +00:00
#include "ipfs/core/daemon.h"
#include "ipfs/routing/routing.h"
#include "ipfs/core/ipfs_node.h"
2017-03-19 19:40:16 +00:00
#include "ipfs/merkledag/merkledag.h"
#include "ipfs/merkledag/node.h"
#include "ipfs/util/thread_pool.h"
#include "ipfs/exchange/bitswap/network.h"
2017-02-10 01:10:21 +00:00
#define BUF_SIZE 4096
2017-05-11 18:53:52 +00:00
// this should be set to 5 for normal operation, perhaps higher for debugging purposes
#define DEFAULT_NETWORK_TIMEOUT 5
static int null_shutting_down = 0;
/***
* Handle the incoming request from a Multistream
* @param incoming the incoming request
* @param incoming_size the size of the request in bytes
* @param session the session context
* @param connection_param the connection parameters
2017-08-03 16:16:58 +00:00
* @returns 1 to indicate it was handled, 0 to indicate that the daemon should no longer loop (future messages will be handled by another message loop), and -1 to indicate a problem
*/
int ipfs_multistream_marshal(const unsigned char* incoming, size_t incoming_size, struct SessionContext* session, struct IpfsNode* local_node) {
/* to be deleted after we get rid of nodeio*/
/*
if (protocol_compare(incoming, incoming_size, "/secio")) {
libp2p_logger_debug("null", "Attempting secure io connection...\n");
if (!libp2p_secio_handshake(session, &local_node->identity->private_key, local_node->peerstore, 1)) {
// rejecting connection
libp2p_logger_debug("null", "Secure IO connection failed\n");
2017-08-03 16:16:58 +00:00
return -1;
}
libp2p_logger_debug("null", "Secure IO connection successful.\n");
2017-08-03 16:16:58 +00:00
return 0;
} else if (protocol_compare(incoming, incoming_size, "/nodeio")) {
libp2p_logger_debug("null", "Attempting a nodeio connection.\n");
if (!libp2p_nodeio_handshake(session)) {
2017-08-03 16:16:58 +00:00
return -1;
}
// loop through file requests
int _continue = 1;
while(_continue) {
unsigned char* hash;
size_t hash_length = 0;
_continue = session->default_stream->read(session, &hash, &hash_length, DEFAULT_NETWORK_TIMEOUT);
if (hash_length < 20) {
_continue = 0;
continue;
}
else {
// try to get the Node
struct HashtableNode* node = NULL;
if (!ipfs_merkledag_get(hash, hash_length, &node, local_node->repo)) {
_continue = 0;
continue;
}
size_t results_size = ipfs_hashtable_node_protobuf_encode_size(node);
unsigned char results[results_size];
if (!ipfs_hashtable_node_protobuf_encode(node, results, results_size, &results_size)) {
_continue = 0;
continue;
}
// send it to the requestor
session->default_stream->write(session, results, results_size);
}
}
} else if (protocol_compare(incoming, incoming_size, "/ipfs/kad/")) {
libp2p_logger_log("null", LOGLEVEL_DEBUG, "Attempting kademlia connection...\n");
if (!libp2p_routing_dht_handshake(session)) {
libp2p_logger_log("null", LOGLEVEL_DEBUG, "kademlia connection handshake failed\n");
2017-08-03 16:16:58 +00:00
return -1;
}
// this handles 1 transaction
libp2p_routing_dht_handle_message(session, local_node->peerstore, 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");
2017-08-03 22:51:34 +00:00
if (!ipfs_bitswap_network_handle_message(local_node, session, incoming, incoming_size)) {
libp2p_logger_error("null", "Bitswap message handler reported a failure.\n");
2017-08-03 16:16:58 +00:00
return -1;
2017-08-03 22:51:34 +00:00
}
}
else {
libp2p_logger_error("null", "There was a problem with this connection. It is nothing I can handle. Disconnecting.\n");
2017-08-03 16:16:58 +00:00
return -1;
}
*/
return 1;
}
/**
2017-07-27 19:33:19 +00:00
* We've received a connection. Find out what they want.
*
* @param ptr a pointer to a null_connection_params struct
*/
2017-08-03 16:16:58 +00:00
void ipfs_null_connection (void *ptr) {
2017-07-27 19:33:19 +00:00
struct null_connection_params *connection_param = (struct null_connection_params*) ptr;
2017-08-03 16:16:58 +00:00
int retVal = 0;
2017-07-27 19:33:19 +00:00
struct SessionContext* session = libp2p_session_context_new();
if (session == NULL) {
2017-08-03 16:16:58 +00:00
libp2p_logger_error("null", "Unable to allocate SessionContext. Out of memory?\n");
return;
2017-07-27 19:33:19 +00:00
}
session->insecure_stream = libp2p_net_multistream_stream_new(connection_param->file_descriptor, connection_param->ip, connection_param->port);
session->default_stream = session->insecure_stream;
session->datastore = connection_param->local_node->repo->config->datastore;
session->filestore = connection_param->local_node->repo->config->filestore;
2017-08-03 16:16:58 +00:00
libp2p_logger_info("null", "Connection %d, count %d\n", connection_param->file_descriptor, *(connection_param->count));
2017-02-10 01:10:21 +00:00
2017-07-27 19:33:19 +00:00
if (libp2p_net_multistream_negotiate(session)) {
// Someone has connected and successfully negotiated multistream. Now talk to them...
2017-07-31 17:50:12 +00:00
int unsuccessful_max = 30;
int unsuccessful_counter = 0;
for(;;) {
// Wait for them to ask something...
2017-03-02 21:18:02 +00:00
unsigned char* results = NULL;
2017-04-03 22:26:33 +00:00
size_t bytes_read = 0;
2017-07-27 19:33:19 +00:00
if (null_shutting_down) {
2017-07-31 17:50:12 +00:00
libp2p_logger_debug("null", "%s null shutting down before read.\n", connection_param->local_node->identity->peer->id);
2017-07-27 19:33:19 +00:00
// this service is shutting down. Ignore the request and exit the loop
2017-04-03 22:26:33 +00:00
break;
2017-04-03 18:41:26 +00:00
}
2017-08-03 16:16:58 +00:00
// see if we have something to read
retVal = session->default_stream->peek(session);
if (retVal < 0) { // error
libp2p_logger_debug("null", "Peer returned %d. Exiting loop\n", retVal);
retVal = -1;
break;
}
if (retVal == 0) { // nothing to read
2017-07-31 11:43:15 +00:00
sleep(1);
2017-07-31 17:50:12 +00:00
unsuccessful_counter++;
2017-08-03 16:16:58 +00:00
if (unsuccessful_counter >= unsuccessful_max) {
libp2p_logger_debug("null", "We've tried %d times in the daemon loop. Exiting.\n", unsuccessful_counter);
retVal = -1;
2017-07-31 17:50:12 +00:00
break;
2017-08-03 16:16:58 +00:00
}
2017-07-27 19:33:19 +00:00
continue;
}
2017-08-03 16:16:58 +00:00
if (retVal > 0 && !session->default_stream->read(session, &results, &bytes_read, DEFAULT_NETWORK_TIMEOUT) ) {
// it said it was ready, but something happened
libp2p_logger_debug("null", "Peek said there was something there, but there was not. Exiting.\n");
retVal = -1;
break;
}
if (null_shutting_down) {
2017-07-31 17:50:12 +00:00
libp2p_logger_debug("null", "%s null shutting down after read.\n", connection_param->local_node->identity->peer->id);
// this service is shutting down. Ignore the request and exit the loop
2017-08-03 16:16:58 +00:00
retVal = -1;
break;
}
// We actually got something. Process the request...
2017-07-31 17:50:12 +00:00
unsuccessful_counter = 0;
2017-04-03 22:26:33 +00:00
libp2p_logger_debug("null", "Read %lu bytes from a stream tranaction\n", bytes_read);
retVal = libp2p_protocol_marshal(results, bytes_read, session, connection_param->local_node->protocol_handlers);
free(results);
2017-08-03 16:16:58 +00:00
if (retVal == -1) {
libp2p_logger_debug("null", "ipfs_null_marshal returned false\n");
break;
2017-08-03 16:16:58 +00:00
} else if (retVal == 0) {
// clean up, but let someone else handle this from now on
libp2p_logger_debug("null", "ipfs_null_marshal returns 0. The daemon will no longer handle this.\n");
break;
} else {
libp2p_logger_debug("null", "ipfs_null_marshal returned 1. Looping again.");
}
}
2017-04-03 17:42:35 +00:00
} else {
2017-04-03 18:20:35 +00:00
libp2p_logger_log("null", LOGLEVEL_DEBUG, "Multistream negotiation failed\n");
}
2017-08-03 16:16:58 +00:00
(*(connection_param->count))--; // update counter.
if (connection_param->ip != NULL)
free(connection_param->ip);
free (connection_param);
if (retVal != 0) {
libp2p_logger_debug("null", "%s Freeing session context.\n", connection_param->local_node->identity->peer->id);
2017-08-03 19:35:06 +00:00
//libp2p_session_context_free(session);
2017-08-03 16:16:58 +00:00
}
return;
2017-02-10 01:10:21 +00:00
}
2017-07-27 19:33:19 +00:00
/***
* Called by the daemon to listen for connections
* @param ptr a pointer to an IpfsNodeListenParams struct
* @returns nothing useful.
*/
2017-07-17 18:05:56 +00:00
void* ipfs_null_listen (void *ptr)
2017-02-10 01:10:21 +00:00
{
int socketfd, s, count = 0;
threadpool thpool = thpool_init(25);
struct IpfsNodeListenParams *listen_param;
2017-02-10 01:10:21 +00:00
struct null_connection_params *connection_param;
listen_param = (struct IpfsNodeListenParams*) ptr;
2017-02-10 01:10:21 +00:00
if ((socketfd = socket_listen(socket_tcp4(), &(listen_param->ipv4), &(listen_param->port))) <= 0) {
2017-04-27 20:52:20 +00:00
libp2p_logger_error("null", "Failed to init null router. Address: %d, Port: %d\n", listen_param->ipv4, listen_param->port);
2017-05-11 18:53:52 +00:00
return (void*) 2;
2017-02-10 01:10:21 +00:00
}
2017-05-11 18:53:52 +00:00
libp2p_logger_error("null", "Ipfs listening on %d\n", listen_param->port);
2017-02-10 01:10:21 +00:00
2017-08-03 16:16:58 +00:00
// the main loop, listening for new connections
2017-02-10 01:10:21 +00:00
for (;;) {
2017-08-03 16:16:58 +00:00
//libp2p_logger_debug("null", "%s Attempting socket read with fd %d.\n", listen_param->local_node->identity->peer->id, socketfd);
int numDescriptors = socket_read_select4(socketfd, 2);
if (null_shutting_down) {
libp2p_logger_debug("null", "%s null_listen shutting down.\n", listen_param->local_node->identity->peer->id);
break;
}
if (numDescriptors > 0) {
s = socket_accept4(socketfd, &(listen_param->ipv4), &(listen_param->port));
if (count >= CONNECTIONS) { // limit reached.
close (s);
continue;
}
count++;
connection_param = malloc (sizeof (struct null_connection_params));
if (connection_param) {
connection_param->file_descriptor = s;
connection_param->count = &count;
connection_param->local_node = listen_param->local_node;
connection_param->port = listen_param->port;
connection_param->ip = malloc(INET_ADDRSTRLEN);
if (inet_ntop(AF_INET, &(listen_param->ipv4), connection_param->ip, INET_ADDRSTRLEN) == NULL) {
free(connection_param->ip);
connection_param->ip = NULL;
connection_param->port = 0;
}
// Create pthread for ipfs_null_connection.
thpool_add_work(thpool, ipfs_null_connection, connection_param);
}
2017-08-03 16:16:58 +00:00
}
2017-02-10 01:10:21 +00:00
}
thpool_destroy(thpool);
2017-05-11 19:30:52 +00:00
close(socketfd);
2017-02-10 01:10:21 +00:00
return (void*) 2;
}
int ipfs_null_shutdown() {
null_shutting_down = 1;
return 1;
}