#include "platform.h"
#include "gnunet_util_lib.h"
-#include "gnunet_core_service.h"
-
#include "gnunet-service-mesh_connection.h"
#include "gnunet-service-mesh_peer.h"
#include "mesh_protocol_enc.h"
#include "mesh_path.h"
-#define MESH_DEBUG_CONNECTION GNUNET_NO
#define MESH_MAX_POLL_TIME GNUNET_TIME_relative_multiply (\
GNUNET_TIME_UNIT_MINUTES,\
10)
#define MESH_RETRANSMIT_MARGIN 4
-#if MESH_DEBUG_CONNECTION
-#define DEBUG_CONN(...) GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
-#else
-#define DEBUG_CONN(...)
-#endif
-
-
/**
* All the states a connection can be in.
*/
static unsigned long long max_connections;
/**
- * How many messages *in total* are we willing to queue, divide by number of
+ * How many messages *in total* are we willing to queue, divide by number of
* connections to get connection queue size.
*/
static unsigned long long max_msgs_queue;
*/
static struct GNUNET_TIME_Relative refresh_connection_time;
-/**
- * Handle to communicate with core.
- */
-static struct GNUNET_CORE_Handle *core_handle;
-
-/**
- * Initialize a Flow Control structure to the initial state.
- *
- * @param fc Flow Control structure to initialize.
- */
+#if 0 // avoid compiler warning for unused static function
static void
-fc_init (struct MeshFlowControl *fc)
+fc_debug (struct MeshFlowControl *fc)
{
- fc->next_pid = 0;
- fc->last_pid_sent = (uint32_t) -1; /* Next (expected) = 0 */
- fc->last_pid_recv = (uint32_t) -1;
- fc->last_ack_sent = (uint32_t) 0;
- fc->last_ack_recv = (uint32_t) 0;
- fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
- fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
- fc->queue_n = 0;
- fc->queue_max = (max_msgs_queue / max_connections) + 1;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " IN: %u/%u\n",
+ fc->last_pid_recv, fc->last_ack_sent);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " OUT: %u/%u\n",
+ fc->last_pid_sent, fc->last_ack_recv);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " QUEUE: %u/%u\n",
+ fc->queue_n, fc->queue_max);
}
-
-/**
- * Find a connection.
- *
- * @param cid Connection ID.
- */
-static struct MeshConnection *
-connection_get (const struct GNUNET_HashCode *cid)
-{
- return GNUNET_CONTAINER_multihashmap_get (connections, cid);
-}
-
-
-/**
- * Get first sendable message.
- *
- * @param peer The destination peer.
- *
- * @return Best current known path towards the peer, if any.
- */
-static struct MeshPeerQueue *
-peer_get_first_message (const struct MeshPeer *peer)
-{
- struct MeshPeerQueue *q;
-
- for (q = peer->queue_head; NULL != q; q = q->next)
- {
- if (queue_is_sendable (q))
- return q;
- }
-
- return NULL;
-}
-
-
-static int
-queue_is_sendable (struct MeshPeerQueue *q)
+static void
+connection_debug (struct MeshConnection *c)
{
- struct MeshFlowControl *fc;
-
- /* Is PID-independent? */
- switch (q->type)
+ if (NULL == c)
{
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- return GNUNET_YES;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*** DEBUG NULL CONNECTION ***\n");
+ return;
}
-
- /* Is PID allowed? */
- fc = q->fwd ? &q->c->fwd_fc : &q->c->bck_fc;
- if (GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
- return GNUNET_YES;
-
- return GNUNET_NO;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connection %s:%X\n",
+ peer2s (c->t->peer), GNUNET_h2s (&c->id));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " state: %u, pending msgs: %u\n",
+ c->state, c->pending_messages);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " FWD FC\n");
+ fc_debug (&c->fwd_fc);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " BCK FC\n");
+ fc_debug (&c->bck_fc);
}
-
-
+#endif
/**
- * Free a transmission that was already queued with all resources
- * associated to the request.
+ * Get string description for tunnel state.
+ *
+ * @param s Tunnel state.
*
- * @param queue Queue handler to cancel.
- * @param clear_cls Is it necessary to free associated cls?
+ * @return String representation.
*/
-static void
-queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
+static const char *
+GMC_DEBUG_state2s (enum MeshTunnelState s)
{
- struct MeshPeer *peer;
- struct MeshFlowControl *fc;
- int fwd;
-
- fwd = queue->fwd;
- peer = queue->peer;
- GNUNET_assert (NULL != queue->c);
- fc = fwd ? &queue->c->fwd_fc : &queue->c->bck_fc;
-
- if (GNUNET_YES == clear_cls)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " queue destroy type %s\n",
- GNUNET_MESH_DEBUG_M2S (queue->type));
- switch (queue->type)
- {
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
- case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO, "destroying a DESTROY message\n");
- GNUNET_break (GNUNET_YES == queue->c->destroy);
- /* fall through */
- case GNUNET_MESSAGE_TYPE_MESH_FWD:
- case GNUNET_MESSAGE_TYPE_MESH_BCK:
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " prebuilt message\n");;
- GNUNET_free_non_null (queue->cls);
- break;
-
- default:
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR, " type %s unknown!\n",
- GNUNET_MESH_DEBUG_M2S (queue->type));
- }
-
- }
- GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
-
- if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
- queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
+ switch (s)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Q_N- %p %u\n", fc, fc->queue_n);
- fc->queue_n--;
- peer->queue_n--;
- }
- if (NULL != queue->c)
- {
- queue->c->pending_messages--;
- if (NULL != queue->c->t)
- {
- queue->c->t->pending_messages--;
- }
- }
-
- GNUNET_free (queue);
-}
-
-
-
-static size_t
-queue_send (void *cls, size_t size, void *buf)
-{
- struct MeshPeer *peer = cls;
- struct MeshFlowControl *fc;
- struct MeshConnection *c;
- struct GNUNET_MessageHeader *msg;
- struct MeshPeerQueue *queue;
- struct MeshTunnel2 *t;
- struct MeshChannel *ch;
- const struct GNUNET_PeerIdentity *dst_id;
- size_t data_size;
- uint32_t pid;
- uint16_t type;
- int fwd;
-
- peer->core_transmit = NULL;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Queue send (max %u)\n", size);
-
- if (NULL == buf || 0 == size)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Buffer size 0.\n");
- return 0;
- }
-
- /* Initialize */
- queue = peer_get_first_message (peer);
- if (NULL == queue)
- {
- GNUNET_break (0); /* Core tmt_rdy should've been canceled */
- return 0;
- }
- c = queue->c;
- fwd = queue->fwd;
- fc = fwd ? &c->fwd_fc : &c->bck_fc;
-
- dst_id = GNUNET_PEER_resolve2 (peer->id);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* towards %s\n", GNUNET_i2s (dst_id));
- /* Check if buffer size is enough for the message */
- if (queue->size > size)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* not enough room, reissue\n");
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready (core_handle,
- GNUNET_NO,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- dst_id,
- queue->size,
- &queue_send,
- peer);
- return 0;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* size %u ok\n", queue->size);
-
- t = (NULL != c) ? c->t : NULL;
- type = 0;
-
- /* Fill buf */
- switch (queue->type)
- {
- case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
- case GNUNET_MESSAGE_TYPE_MESH_FWD:
- case GNUNET_MESSAGE_TYPE_MESH_BCK:
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "* raw: %s\n",
- GNUNET_MESH_DEBUG_M2S (queue->type));
- data_size = send_core_data_raw (queue->cls, size, buf);
- msg = (struct GNUNET_MessageHeader *) buf;
- type = ntohs (msg->type);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* path create\n");
- if (GMC_is_origin (c, GNUNET_YES))
- data_size = send_core_connection_create (queue->c, size, buf);
- else
- data_size = send_core_data_raw (queue->cls, size, buf);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* path ack\n");
- if (GMC_is_origin (c, GNUNET_NO) ||
- GMC_is_origin (c, GNUNET_YES))
- data_size = send_core_connection_ack (queue->c, size, buf);
- else
- data_size = send_core_data_raw (queue->cls, size, buf);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_DATA:
- case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
- case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
- /* This should be encapsulted */
- GNUNET_break (0);
- data_size = 0;
- break;
- default:
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "* type unknown: %u\n",
- queue->type);
- data_size = 0;
- }
-
- if (0 < drop_percent &&
- GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK, 101) < drop_percent)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Dropping message of type %s\n",
- GNUNET_MESH_DEBUG_M2S (queue->type));
- data_size = 0;
- }
-
- /* Free queue, but cls was freed by send_core_* */
- ch = queue->ch;
- queue_destroy (queue, GNUNET_NO);
-
- /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
- switch (type)
- {
- case GNUNET_MESSAGE_TYPE_MESH_FWD:
- case GNUNET_MESSAGE_TYPE_MESH_BCK:
- pid = ntohl ( ((struct GNUNET_MESH_Encrypted *) buf)->pid );
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* accounting pid %u\n", pid);
- fc->last_pid_sent = pid;
- send_ack (c, ch, fwd);
- break;
+ case MESH_CONNECTION_NEW:
+ return "MESH_CONNECTION_NEW";
+ case MESH_CONNECTION_SENT:
+ return "MESH_CONNECTION_SENT";
+ case MESH_CONNECTION_ACK:
+ return "MESH_CONNECTION_ACK";
+ case MESH_CONNECTION_READY:
+ return "MESH_CONNECTION_READY";
default:
- break;
- }
-
- /* If more data in queue, send next */
- queue = peer_get_first_message (peer);
- if (NULL != queue)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* more data!\n");
- if (NULL == peer->core_transmit) {
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready(core_handle,
- 0,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- dst_id,
- queue->size,
- &queue_send,
- peer);
- }
- else
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "* tmt rdy called somewhere else\n");
- }
- if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* starting poll timeout\n");
- fc->poll_task =
- GNUNET_SCHEDULER_add_delayed (fc->poll_time, &connection_poll, fc);
- }
- }
- else
- {
- if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
- {
- GNUNET_SCHEDULER_cancel (fc->poll_task);
- fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
- }
- }
- if (NULL != c)
- {
- c->pending_messages--;
- if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying connection!\n");
- GMC_destroy (c);
- }
- }
-
- if (NULL != t)
- {
- t->pending_messages--;
- if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
- {
-// GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying tunnel!\n");
- tunnel_destroy (t);
- }
+ return "MESH_CONNECTION_STATE_ERROR";
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Return %d\n", data_size);
- return data_size;
}
-static void
-queue_add (void *cls, uint16_t type, size_t size,
- struct MeshConnection *c,
- struct MeshChannel *ch,
- int fwd)
-{
- struct MeshPeerQueue *queue;
- struct MeshFlowControl *fc;
- struct MeshPeer *peer;
- int priority;
- int call_core;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "queue add %s %s (%u) on c %p, ch %p\n",
- fwd ? "FWD" : "BCK", GNUNET_MESH_DEBUG_M2S (type), size, c, ch);
- GNUNET_assert (NULL != c);
-
- fc = fwd ? &c->fwd_fc : &c->bck_fc;
- peer = fwd ? connection_get_next_hop (c) : connection_get_prev_hop (c);
-
- if (NULL == fc)
- {
- GNUNET_break (0);
- return;
- }
-
- if (NULL == peer->connections)
- {
- /* We are not connected to this peer, ignore request. */
- GNUNET_break_op (0);
- return;
- }
-
- priority = 0;
-
- if (GNUNET_MESSAGE_TYPE_MESH_POLL == type ||
- GNUNET_MESSAGE_TYPE_MESH_ACK == type)
- {
- priority = 100;
- }
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "priority %d\n", priority);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "fc %p\n", fc);
- if (fc->queue_n >= fc->queue_max && 0 == priority)
- {
- GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
- 1, GNUNET_NO);
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "queue full: %u/%u\n",
- fc->queue_n, fc->queue_max);
- return; /* Drop this message */
- }
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "last pid %u\n", fc->last_pid_sent);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", fc->last_ack_recv);
- if (GMC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
- {
- call_core = GNUNET_NO;
- if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task &&
- GNUNET_MESSAGE_TYPE_MESH_POLL != type)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "no buffer space (%u > %u): starting poll\n",
- fc->last_pid_sent + 1, fc->last_ack_recv);
- fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
- &connection_poll,
- fc);
- }
- }
- else
- call_core = GNUNET_YES;
- queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
- queue->cls = cls;
- queue->type = type;
- queue->size = size;
- queue->peer = peer;
- queue->c = c;
- queue->ch = ch;
- queue->fwd = fwd;
- if (100 <= priority)
- {
- struct MeshPeerQueue *copy;
- struct MeshPeerQueue *next;
-
- for (copy = peer->queue_head; NULL != copy; copy = next)
- {
- next = copy->next;
- if (copy->type == type && copy->c == c && copy->fwd == fwd)
- {
- /* Example: also a FWD ACK for connection XYZ */
- queue_destroy (copy, GNUNET_YES);
- }
- }
- GNUNET_CONTAINER_DLL_insert (peer->queue_head, peer->queue_tail, queue);
- }
- else
- {
- GNUNET_CONTAINER_DLL_insert_tail (peer->queue_head, peer->queue_tail, queue);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Q_N+ %p %u\n", fc, fc->queue_n);
- fc->queue_n++;
- peer->queue_n++;
- }
-
- if (NULL == peer->core_transmit && GNUNET_YES == call_core)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "calling core tmt rdy towards %s for %u bytes\n",
- peer2s (peer), size);
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready (core_handle,
- 0,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_PEER_resolve2 (peer->id),
- size,
- &queue_send,
- peer);
- }
- else
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "core tmt rdy towards %s already called\n",
- peer2s (peer));
-
- }
- c->pending_messages++;
- if (NULL != c->t)
- c->t->pending_messages++;
-}
-
-
-
-
/**
- * Sends an already built message on a connection, properly registering
- * all used resources.
+ * Initialize a Flow Control structure to the initial state.
*
- * @param message Message to send. Function makes a copy of it.
- * If message is not hop-by-hop, decrements TTL of copy.
- * @param c Connection on which this message is transmitted.
- * @param ch Channel on which this message is transmitted, or NULL.
- * @param fwd Is this a fwd message?
+ * @param fc Flow Control structure to initialize.
*/
static void
-send_prebuilt_message_connection (const struct GNUNET_MessageHeader *message,
- struct MeshConnection *c,
- struct MeshChannel *ch,
- int fwd)
-{
- void *data;
- size_t size;
- uint16_t type;
-
- size = ntohs (message->size);
- data = GNUNET_malloc (size);
- memcpy (data, message, size);
- type = ntohs (message->type);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send %s (%u) on connection %s\n",
- GNUNET_MESH_DEBUG_M2S (type), size, GNUNET_h2s (&c->id));
-
- switch (type)
- {
- struct GNUNET_MESH_Encrypted *emsg;
- struct GNUNET_MESH_ACK *amsg;
- struct GNUNET_MESH_Poll *pmsg;
- struct GNUNET_MESH_ConnectionDestroy *dmsg;
- struct GNUNET_MESH_ConnectionBroken *bmsg;
- uint32_t ttl;
-
- case GNUNET_MESSAGE_TYPE_MESH_FWD:
- case GNUNET_MESSAGE_TYPE_MESH_BCK:
- emsg = (struct GNUNET_MESH_Encrypted *) data;
- ttl = ntohl (emsg->ttl);
- if (0 == ttl)
- {
- GNUNET_break_op (0);
- return;
- }
- emsg->cid = c->id;
- emsg->ttl = htonl (ttl - 1);
- emsg->pid = htonl (fwd ? c->fwd_fc.next_pid++ : c->bck_fc.next_pid++);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " pid %u\n", ntohl (emsg->pid));
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- amsg = (struct GNUNET_MESH_ACK *) data;
- amsg->cid = c->id;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- pmsg = (struct GNUNET_MESH_Poll *) data;
- pmsg->cid = c->id;
- pmsg->pid = htonl (fwd ? c->fwd_fc.last_pid_sent : c->bck_fc.last_pid_sent);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " poll %u\n", ntohl (pmsg->pid));
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
- dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
- dmsg->cid = c->id;
- dmsg->reserved = 0;
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
- bmsg = (struct GNUNET_MESH_ConnectionBroken *) data;
- bmsg->cid = c->id;
- bmsg->reserved = 0;
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
- case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
- break;
-
- default:
- GNUNET_break (0);
- }
-
- queue_add (data,
- type,
- size,
- c,
- ch,
- fwd);
-}
-
-
-
-
-struct MeshConnection *
-GMC_new (const struct GNUNET_HashCode *cid)
-{
- struct MeshConnection *c;
-
- c = GNUNET_new (struct MeshConnection);
- c->id = *cid;
- GNUNET_CONTAINER_multihashmap_put (connections, &c->id, c,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
- fc_init (&c->fwd_fc);
- fc_init (&c->bck_fc);
- c->fwd_fc.c = c;
- c->bck_fc.c = c;
-
- return c;
-}
-
-
-static void
-GMC_destroy (struct MeshConnection *c)
+fc_init (struct MeshFlowControl *fc)
{
- struct MeshPeer *peer;
-
- if (NULL == c)
- return;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s[%X]\n",
- peer2s (c->t->peer),
- c->id);
-
- /* Cancel all traffic */
- connection_cancel_queues (c, GNUNET_YES);
- connection_cancel_queues (c, GNUNET_NO);
-
- /* Cancel maintainance task (keepalive/timeout) */
- if (GNUNET_SCHEDULER_NO_TASK != c->fwd_maintenance_task)
- GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
- if (GNUNET_SCHEDULER_NO_TASK != c->bck_maintenance_task)
- GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
-
- /* Deregister from neighbors */
- peer = connection_get_next_hop (c);
- if (NULL != peer && NULL != peer->connections)
- GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
- peer = connection_get_prev_hop (c);
- if (NULL != peer && NULL != peer->connections)
- GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
-
- /* Delete */
- GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
- GNUNET_CONTAINER_DLL_remove (c->t->connection_head, c->t->connection_tail, c);
- GNUNET_free (c);
+ fc->next_pid = 0;
+ fc->last_pid_sent = (uint32_t) -1; /* Next (expected) = 0 */
+ fc->last_pid_recv = (uint32_t) -1;
+ fc->last_ack_sent = (uint32_t) 0;
+ fc->last_ack_recv = (uint32_t) 0;
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
+ fc->queue_n = 0;
+ fc->queue_max = (max_msgs_queue / max_connections) + 1;
}
+/**
+ * Find a connection.
+ *
+ * @param cid Connection ID.
+ */
+static struct MeshConnection *
+connection_get (const struct GNUNET_HashCode *cid)
+{
+ return GNUNET_CONTAINER_multihashmap_get (connections, cid);
+}
+
/**
* Send an ACK informing the predecessor about the available buffer space.
}
+/**
+ * Sends a CONNECTION ACK message in reponse to a received CONNECTION_CREATE
+ * directed to us.
+ *
+ * @param connection Connection to confirm.
+ * @param fwd Is this a fwd ACK? (First is bck (SYNACK), second is fwd (ACK))
+ */
+static void
+send_connection_ack (struct MeshConnection *connection, int fwd)
+{
+ struct MeshTunnel2 *t;
+
+ t = connection->t;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send connection ack\n");
+ queue_add (NULL,
+ GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
+ sizeof (struct GNUNET_MESH_ConnectionACK),
+ connection,
+ NULL,
+ fwd);
+ if (MESH_TUNNEL_NEW == t->state)
+ tunnel_change_state (t, MESH_TUNNEL_WAITING);
+ if (MESH_CONNECTION_READY != connection->state)
+ connection_change_state (connection, MESH_CONNECTION_SENT);
+}
+
+
+/**
+ * Sends a CREATE CONNECTION message for a path to a peer.
+ * Changes the connection and tunnel states if necessary.
+ *
+ * @param connection Connection to create.
+ */
+static void
+send_connection_create (struct MeshConnection *connection)
+{
+ struct MeshTunnel2 *t;
+
+ t = connection->t;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send connection create\n");
+ queue_add (NULL,
+ GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE,
+ sizeof (struct GNUNET_MESH_ConnectionCreate) +
+ (connection->path->length *
+ sizeof (struct GNUNET_PeerIdentity)),
+ connection,
+ NULL,
+ GNUNET_YES);
+ if (NULL != t &&
+ (MESH_TUNNEL_SEARCHING == t->state || MESH_TUNNEL_NEW == t->state))
+ tunnel_change_state (t, MESH_TUNNEL_WAITING);
+ if (MESH_CONNECTION_NEW == connection->state)
+ connection_change_state (connection, MESH_CONNECTION_SENT);
+}
+
+
static void
connection_change_state (struct MeshConnection* c,
enum MeshConnectionState state)
connection_get_buffer (struct MeshConnection *c, int fwd)
{
struct MeshFlowControl *fc;
-
+
fc = fwd ? &c->fwd_fc : &c->bck_fc;
-
+
return (fc->queue_max - fc->queue_n);
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** Polling!\n");
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** connection [%X]\n",
GNUNET_h2s (&c->id));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** %s\n",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** %s\n",
fc == &c->fwd_fc ? "FWD" : "BCK");
msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
*
* @param c Connection.
*
- * @return Next peer in the connection.
+ * @return Next peer in the connection.
*/
static struct MeshPeer *
connection_get_next_hop (struct MeshConnection *c)
* @param c Connection.
* @param fwd Next hop?
*
- * @return Next peer in the connection.
+ * @return Next peer in the connection.
*/
static struct MeshPeer *
connection_get_hop (struct MeshConnection *c, int fwd)
+
+
/**
* Core handler for connection creation.
*
return GNUNET_OK;
}
- fwd = GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE == ntohs (message->type) ?
+ fwd = GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE == ntohs (message->type) ?
GNUNET_YES : GNUNET_NO;
/* Check if origin is as expected */
/**
- * Iterator to notify all connections of a broken link. Mark connections
- * to destroy after all traffic has been sent.
- *
- * @param cls Closure (peer disconnected).
- * @param key Current key code (tid).
- * @param value Value in the hash map (connection).
+ * Send an ACK on the appropriate connection/channel, depending on
+ * the direction and the position of the peer.
*
- * @return GNUNET_YES if we should continue to iterate,
- * GNUNET_NO if not.
+ * @param c Which connection to send the hop-by-hop ACK.
+ * @param ch Channel, if any.
+ * @param fwd Is this a fwd ACK? (will go dest->root)
*/
-int
-GMC_notify_broken (void *cls,
- const struct GNUNET_HashCode *key,
- void *value)
+static void
+send_ack (struct MeshConnection *c, struct MeshChannel *ch, int fwd)
{
- struct MeshPeer *peer = cls;
- struct MeshConnection *c = value;
- struct GNUNET_MESH_ConnectionBroken msg;
- int fwd;
-
- fwd = peer == connection_get_prev_hop (c);
+ unsigned int buffer;
- connection_cancel_queues (c, !fwd);
- if (GMC_is_terminal (c, fwd))
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "send ack %s on %p %p\n",
+ fwd ? "FWD" : "BCK", c, ch);
+ if (NULL == c || GMC_is_terminal (c, fwd))
{
- /* Local shutdown, no one to notify about this. */
- GMC_destroy (c);
- return GNUNET_YES;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " getting from all connections\n");
+ buffer = tunnel_get_buffer (NULL == c ? ch->t : c->t, fwd);
}
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " getting from one connection\n");
+ buffer = connection_get_buffer (c, fwd);
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " buffer available: %u\n", buffer);
- msg.header.size = htons (sizeof (struct GNUNET_MESH_ConnectionBroken));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN);
- msg.cid = c->id;
- msg.peer1 = my_full_id;
- msg.peer2 = *GNUNET_PEER_resolve2 (peer->id);
- send_prebuilt_message_connection (&msg.header, c, NULL, fwd);
- c->destroy = GNUNET_YES;
-
- return GNUNET_YES;
+ if ( (NULL != ch && channel_is_origin (ch, fwd)) ||
+ (NULL != c && connection_is_origin (c, fwd)) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending on channel...\n");
+ if (0 < buffer)
+ {
+ GNUNET_assert (NULL != ch);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " really sending!\n");
+ send_local_ack (ch, fwd);
+ }
+ }
+ else if (NULL == c)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending on all connections\n");
+ GNUNET_assert (NULL != ch);
+ channel_send_connections_ack (ch, buffer, fwd);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending on connection\n");
+ connection_send_ack (c, buffer, fwd);
+ }
}
+
/**
* Initialize the connections subsystem
*
return;
}
connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
-
- core_handle = GNUNET_CORE_connect (c, /* Main configuration */
- NULL, /* Closure passed to MESH functions */
- &core_init, /* Call core_init once connected */
- &core_connect, /* Handle connects */
- &core_disconnect, /* remove peers on disconnects */
- NULL, /* Don't notify about all incoming messages */
- GNUNET_NO, /* For header only in notification */
- NULL, /* Don't notify about all outbound messages */
- GNUNET_NO, /* For header-only out notification */
- core_handlers); /* Register these handlers */
- if (NULL == core_handle)
- {
- GNUNET_break (0);
- GNUNET_SCHEDULER_shutdown ();
- return;
- }
}
/**
}
+struct MeshConnection *
+GMC_new (const struct GNUNET_HashCode *cid)
+{
+ struct MeshConnection *c;
+
+ c = GNUNET_new (struct MeshConnection);
+ c->id = *cid;
+ GNUNET_CONTAINER_multihashmap_put (connections, &c->id, c,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ fc_init (&c->fwd_fc);
+ fc_init (&c->bck_fc);
+ c->fwd_fc.c = c;
+ c->bck_fc.c = c;
+
+ return c;
+}
+
+
+static void
+GMC_destroy (struct MeshConnection *c)
+{
+ struct MeshPeer *peer;
+
+ if (NULL == c)
+ return;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s[%X]\n",
+ peer2s (c->t->peer),
+ c->id);
+
+ /* Cancel all traffic */
+ connection_cancel_queues (c, GNUNET_YES);
+ connection_cancel_queues (c, GNUNET_NO);
+
+ /* Cancel maintainance task (keepalive/timeout) */
+ if (GNUNET_SCHEDULER_NO_TASK != c->fwd_maintenance_task)
+ GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
+ if (GNUNET_SCHEDULER_NO_TASK != c->bck_maintenance_task)
+ GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
+
+ /* Deregister from neighbors */
+ peer = connection_get_next_hop (c);
+ if (NULL != peer && NULL != peer->connections)
+ GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
+ peer = connection_get_prev_hop (c);
+ if (NULL != peer && NULL != peer->connections)
+ GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
+
+ /* Delete */
+ GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
+ GNUNET_CONTAINER_DLL_remove (c->t->connection_head, c->t->connection_tail, c);
+ GNUNET_free (c);
+}
+
+
+/**
+ * Notify other peers on a connection of a broken link. Mark connections
+ * to destroy after all traffic has been sent.
+ *
+ * @param c Connection on which there has been a disconnection.
+ * @param peer Peer that disconnected.
+ * @param my_full_id My ID (to send to other peers).
+ */
+void
+GMC_notify_broken (struct MeshConnection *c,
+ struct MeshPeer *peer,
+ struct GNUNET_PeerIdentity *my_full_id)
+{
+ struct MeshConnection *c = value;
+ struct GNUNET_MESH_ConnectionBroken msg;
+ int fwd;
+
+ fwd = peer == connection_get_prev_hop (c);
+
+ connection_cancel_queues (c, !fwd);
+ if (GMC_is_terminal (c, fwd))
+ {
+ /* Local shutdown, no one to notify about this. */
+ GMC_destroy (c);
+ return GNUNET_YES;
+ }
+
+ msg.header.size = htons (sizeof (struct GNUNET_MESH_ConnectionBroken));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN);
+ msg.cid = c->id;
+ msg.peer1 = *my_full_id;
+ msg.peer2 = *GNUNET_PEER_resolve2 (peer->id);
+ GMC_send_prebuilt_message (&msg.header, c, NULL, fwd);
+ c->destroy = GNUNET_YES;
+
+ return GNUNET_YES;
+}
+
+
/**
* Is this peer the first one on the connection?
*
{
return GMC_is_origin (c, !fwd);
}
+
+
+/**
+ * Count connections in a DLL.
+ */
+unsigned int
+GMC_count (const struct MeshConnection *head)
+{
+ unsigned int count;
+ struct MeshConnection *iter;
+
+ for (count = 0, iter = head; NULL != iter; iter = iter->next, count++);
+
+ return count;
+}
+
+
+/**
+ * Sends an already built message on a connection, properly registering
+ * all used resources.
+ *
+ * @param message Message to send. Function makes a copy of it.
+ * If message is not hop-by-hop, decrements TTL of copy.
+ * @param c Connection on which this message is transmitted.
+ * @param ch Channel on which this message is transmitted, or NULL.
+ * @param fwd Is this a fwd message?
+ */
+void
+GMC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
+ struct MeshConnection *c,
+ struct MeshChannel *ch,
+ int fwd)
+{
+ void *data;
+ size_t size;
+ uint16_t type;
+
+ size = ntohs (message->size);
+ data = GNUNET_malloc (size);
+ memcpy (data, message, size);
+ type = ntohs (message->type);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send %s (%u) on connection %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), size, GNUNET_h2s (&c->id));
+
+ switch (type)
+ {
+ struct GNUNET_MESH_Encrypted *emsg;
+ struct GNUNET_MESH_ACK *amsg;
+ struct GNUNET_MESH_Poll *pmsg;
+ struct GNUNET_MESH_ConnectionDestroy *dmsg;
+ struct GNUNET_MESH_ConnectionBroken *bmsg;
+ uint32_t ttl;
+
+ case GNUNET_MESSAGE_TYPE_MESH_FWD:
+ case GNUNET_MESSAGE_TYPE_MESH_BCK:
+ emsg = (struct GNUNET_MESH_Encrypted *) data;
+ ttl = ntohl (emsg->ttl);
+ if (0 == ttl)
+ {
+ GNUNET_break_op (0);
+ return;
+ }
+ emsg->cid = c->id;
+ emsg->ttl = htonl (ttl - 1);
+ emsg->pid = htonl (fwd ? c->fwd_fc.next_pid++ : c->bck_fc.next_pid++);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " pid %u\n", ntohl (emsg->pid));
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ amsg = (struct GNUNET_MESH_ACK *) data;
+ amsg->cid = c->id;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_POLL:
+ pmsg = (struct GNUNET_MESH_Poll *) data;
+ pmsg->cid = c->id;
+ pmsg->pid = htonl (fwd ? c->fwd_fc.last_pid_sent : c->bck_fc.last_pid_sent);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " poll %u\n", ntohl (pmsg->pid));
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
+ dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
+ dmsg->cid = c->id;
+ dmsg->reserved = 0;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
+ bmsg = (struct GNUNET_MESH_ConnectionBroken *) data;
+ bmsg->cid = c->id;
+ bmsg->reserved = 0;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
+ break;
+
+ default:
+ GNUNET_break (0);
+ }
+
+ GMP_queue_add (data,
+ type,
+ size,
+ c,
+ ch,
+ fwd);
+}
\ No newline at end of file
projects / oweals / gnunet.git / blobdiff