*/
#include "platform.h"
-#include "gnunet_core_service.h"
-#include "gnunet-service-mesh_connection.h"
-
-
-
-
-/**
- * All the states a connection can be in.
- */
-enum MeshConnectionState
-{
- /**
- * Uninitialized status, should never appear in operation.
- */
- MESH_CONNECTION_NEW,
-
- /**
- * Connection create message sent, waiting for ACK.
- */
- MESH_CONNECTION_SENT,
-
- /**
- * Connection ACK sent, waiting for ACK.
- */
- MESH_CONNECTION_ACK,
-
- /**
- * Connection confirmed, ready to carry traffic.
- */
- MESH_CONNECTION_READY,
-};
-
-
-
-/**
- * Struct containing info about a queued transmission to this peer
- */
-struct MeshPeerQueue
-{
- /**
- * DLL next
- */
- struct MeshPeerQueue *next;
-
- /**
- * DLL previous
- */
- struct MeshPeerQueue *prev;
-
- /**
- * Peer this transmission is directed to.
- */
- struct MeshPeer *peer;
+#include "gnunet_util_lib.h"
- /**
- * Connection this message belongs to.
- */
- struct MeshConnection *c;
+#include "gnunet_statistics_service.h"
- /**
- * Is FWD in c?
- */
- int fwd;
+#include "mesh_path.h"
+#include "mesh_protocol_enc.h"
+#include "mesh_enc.h"
+#include "gnunet-service-mesh_connection.h"
+#include "gnunet-service-mesh_peer.h"
+#include "gnunet-service-mesh_tunnel.h"
+#include "gnunet-service-mesh_channel.h"
- /**
- * Channel this message belongs to, if known.
- */
- struct MeshChannel *ch;
- /**
- * Pointer to info stucture used as cls.
- */
- void *cls;
+#define LOG(level, ...) GNUNET_log_from (level,"mesh-con",__VA_ARGS__)
- /**
- * Type of message
- */
- uint16_t type;
+#define MESH_MAX_POLL_TIME GNUNET_TIME_relative_multiply (\
+ GNUNET_TIME_UNIT_MINUTES,\
+ 10)
+#define AVG_MSGS 32
- /**
- * Size of the message
- */
- size_t size;
-};
+/******************************************************************************/
+/******************************** STRUCTS **********************************/
+/******************************************************************************/
/**
* Struct to encapsulate all the Flow Control information to a peer to which
struct GNUNET_TIME_Relative poll_time;
};
-
/**
- * Struct containing all information regarding a connection to a peer.
+ * Keep a record of the last messages sent on this connection.
*/
-struct MeshConnection
+struct MeshConnectionPerformance
{
/**
- * DLL
+ * Circular buffer for storing measurements.
+ */
+ double usecsperbyte[AVG_MSGS];
+
+ /**
+ * Running average of @c usecsperbyte.
+ */
+ double avg;
+
+ /**
+ * How many values of @c usecsperbyte are valid.
+ */
+ uint16_t size;
+
+ /**
+ * Index of the next "free" position in @c usecsperbyte.
*/
- struct MeshConnection *next;
- struct MeshConnection *prev;
+ uint16_t idx;
+};
+
+/**
+ * Struct containing all information regarding a connection to a peer.
+ */
+struct MeshConnection
+{
/**
* Tunnel this connection is part of.
*/
- struct MeshTunnel2 *t;
+ struct MeshTunnel3 *t;
/**
* Flow control information for traffic fwd.
*/
struct MeshFlowControl bck_fc;
+ /**
+ * Measure connection performance on the endpoint.
+ */
+ struct MeshConnectionPerformance *perf;
+
/**
* ID of the connection.
*/
int destroy;
};
+/******************************************************************************/
+/******************************* GLOBALS ***********************************/
+/******************************************************************************/
+/**
+ * Global handle to the statistics service.
+ */
+extern struct GNUNET_STATISTICS_Handle *stats;
+/**
+ * Local peer own ID (memory efficient handle).
+ */
+extern GNUNET_PEER_Id myid;
-
+/**
+ * Local peer own ID (full value).
+ */
+extern struct GNUNET_PeerIdentity my_full_id;
/**
* Connections known, indexed by cid (MeshConnection).
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;
+/******************************************************************************/
+/******************************** STATIC ***********************************/
+/******************************************************************************/
+
+#if 0 // avoid compiler warning for unused static function
+static void
+fc_debug (struct MeshFlowControl *fc)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " IN: %u/%u\n",
+ fc->last_pid_recv, fc->last_ack_sent);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " OUT: %u/%u\n",
+ fc->last_pid_sent, fc->last_ack_recv);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " QUEUE: %u/%u\n",
+ fc->queue_n, fc->queue_max);
+}
+
+static void
+connection_debug (struct MeshConnection *c)
+{
+ if (NULL == c)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "*** DEBUG NULL CONNECTION ***\n");
+ return;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s:%X\n",
+ peer2s (c->t->peer), GNUNET_h2s (&c->id));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " state: %u, pending msgs: %u\n",
+ c->state, c->pending_messages);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD FC\n");
+ fc_debug (&c->fwd_fc);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK FC\n");
+ fc_debug (&c->bck_fc);
+}
+#endif
+
+/**
+ * Get string description for tunnel state.
+ *
+ * @param s Tunnel state.
+ *
+ * @return String representation.
+ */
+static const char *
+GMC_state2s (enum MeshConnectionState s)
+{
+ switch (s)
+ {
+ 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:
+ return "MESH_CONNECTION_STATE_ERROR";
+ }
+}
/**
* Initialize a Flow Control structure to the initial state.
- *
+ *
* @param fc Flow Control structure to initialize.
*/
static void
}
+static void
+connection_change_state (struct MeshConnection* c,
+ enum MeshConnectionState state)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s state was %s\n",
+ GNUNET_h2s (&c->id), GMC_state2s (c->state));
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s state is now %s\n",
+ GNUNET_h2s (&c->id), GMC_state2s (state));
+ c->state = state;
+}
+
+
/**
- * Get first sendable message.
+ * Callback called when a queued message is sent.
*
- * @param peer The destination peer.
+ * Calculates the average time and connection packet tracking.
*
- * @return Best current known path towards the peer, if any.
+ * @param cls Closure.
+ * @param c Connection this message was on.
+ * @param type Type of message sent.
+ * @param fwd Was this a FWD going message?
+ * @param size Size of the message.
+ * @param wait Time spent waiting for core (only the time for THIS message)
*/
-static struct MeshPeerQueue *
-peer_get_first_message (const struct MeshPeer *peer)
+static void
+message_sent (void *cls,
+ struct MeshConnection *c, uint16_t type,
+ int fwd, size_t size,
+ struct GNUNET_TIME_Relative wait)
{
- struct MeshPeerQueue *q;
+ struct MeshConnectionPerformance *p;
+ struct MeshFlowControl *fc;
+ double usecsperbyte;
- for (q = peer->queue_head; NULL != q; q = q->next)
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! Q_N- %p %u\n", fc, fc->queue_n);
+ fc->queue_n--;
+ c->pending_messages--;
+ if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! destroying connection!\n");
+ GMC_destroy (c);
+ }
+ /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
+ switch (type)
{
- if (queue_is_sendable (q))
- return q;
+ case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
+ fc->last_pid_sent++;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! accounting pid %u\n", fc->last_pid_sent);
+// send_ack (c, ch, fwd);
+ break;
+ default:
+ break;
}
- return NULL;
+ if (NULL == c->perf)
+ return; /* Only endpoints are interested in timing. */
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! message sent!\n");
+ p = c->perf;
+ usecsperbyte = ((double) wait.rel_value_us) / size;
+ if (p->size == AVG_MSGS)
+ {
+ /* Array is full. Substract oldest value, add new one and store. */
+ p->avg -= (p->usecsperbyte[p->idx] / AVG_MSGS);
+ p->usecsperbyte[p->idx] = usecsperbyte;
+ p->avg += (p->usecsperbyte[p->idx] / AVG_MSGS);
+ }
+ else
+ {
+ /* Array not yet full. Add current value to avg and store. */
+ p->usecsperbyte[p->idx] = usecsperbyte;
+ p->avg *= p->size;
+ p->avg += p->usecsperbyte[p->idx];
+ p->size++;
+ p->avg /= p->size;
+ }
+ p->idx = (p->idx + 1) % AVG_MSGS;
+
+// if (NULL != c->t)
+// {
+// c->t->pending_messages--;
+// if (GNUNET_YES == c->t->destroy && 0 == t->pending_messages)
+// {
+// LOG (GNUNET_ERROR_TYPE_DEBUG, "* destroying tunnel!\n");
+// GMT_destroy (c->t);
+// }
+// }
}
-static int
-queue_is_sendable (struct MeshPeerQueue *q)
+/**
+ * Get the previous hop in a connection
+ *
+ * @param c Connection.
+ *
+ * @return Previous peer in the connection.
+ */
+static struct MeshPeer *
+get_prev_hop (const struct MeshConnection *c)
{
- struct MeshFlowControl *fc;
+ GNUNET_PEER_Id id;
- /* Is PID-independent? */
- switch (q->type)
- {
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- return GNUNET_YES;
- }
+ if (0 == c->own_pos || c->path->length < 2)
+ id = c->path->peers[0];
+ else
+ id = c->path->peers[c->own_pos - 1];
- /* 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 GMP_get_short (id);
+}
- return GNUNET_NO;
+
+/**
+ * Get the next hop in a connection
+ *
+ * @param c Connection.
+ *
+ * @return Next peer in the connection.
+ */
+static struct MeshPeer *
+get_next_hop (const struct MeshConnection *c)
+{
+ GNUNET_PEER_Id id;
+
+ if ((c->path->length - 1) == c->own_pos || c->path->length < 2)
+ id = c->path->peers[c->path->length - 1];
+ else
+ id = c->path->peers[c->own_pos + 1];
+
+ return GMP_get_short (id);
}
+/**
+ * Get the hop in a connection.
+ *
+ * @param c Connection.
+ * @param fwd Next hop?
+ *
+ * @return Next peer in the connection.
+ */
+static struct MeshPeer *
+get_hop (struct MeshConnection *c, int fwd)
+{
+ if (fwd)
+ return get_next_hop (c);
+ return get_prev_hop (c);
+}
+
/**
- * Free a transmission that was already queued with all resources
- * associated to the request.
+ * Send an ACK informing the predecessor about the available buffer space.
+ *
+ * Note that for fwd ack, the FWD mean forward *traffic* (root->dest),
+ * the ACK itself goes "back" (dest->root).
*
- * @param queue Queue handler to cancel.
- * @param clear_cls Is it necessary to free associated cls?
+ * @param c Connection on which to send the ACK.
+ * @param buffer How much space free to advertise?
+ * @param fwd Is this FWD ACK? (Going dest->owner)
*/
static void
-queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
+connection_send_ack (struct MeshConnection *c, unsigned int buffer, int fwd)
{
- 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;
+ struct MeshFlowControl *next_fc;
+ struct MeshFlowControl *prev_fc;
+ struct GNUNET_MESH_ACK msg;
+ uint32_t ack;
+ int delta;
- 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));
- }
+ next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
- }
- GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "connection send %s ack on %s\n",
+ fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
- if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
- queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
+ /* Check if we need to transmit the ACK */
+ delta = prev_fc->last_ack_sent - prev_fc->last_pid_recv;
+ if (3 < delta && buffer < delta)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Q_N- %p %u\n", fc, fc->queue_n);
- fc->queue_n--;
- peer->queue_n--;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer > 3\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " last pid recv: %u, last ack sent: %u\n",
+ prev_fc->last_pid_recv, prev_fc->last_ack_sent);
+ return;
}
- if (NULL != queue->c)
+
+ /* Ok, ACK might be necessary, what PID to ACK? */
+ ack = prev_fc->last_pid_recv + buffer;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " last pid %u, last ack %u, qmax %u, q %u\n",
+ prev_fc->last_pid_recv, prev_fc->last_ack_sent,
+ next_fc->queue_max, next_fc->queue_n);
+ if (ack == prev_fc->last_ack_sent)
{
- queue->c->pending_messages--;
- if (NULL != queue->c->t)
- {
- queue->c->t->pending_messages--;
- }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
+ return;
}
- GNUNET_free (queue);
+ prev_fc->last_ack_sent = ack;
+
+ /* Build ACK message and send on connection */
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
+ msg.ack = htonl (ack);
+ msg.cid = c->id;
+
+ GMC_send_prebuilt_message (&msg.header, c, NULL, !fwd);
}
+/**
+ * Sends a CONNECTION ACK message in reponse to a received CONNECTION_CREATE
+ * or a first CONNECTION_ACK 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 MeshTunnel3 *t;
+
+ t = connection->t;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Send connection ack\n");
+ GMP_queue_add (get_hop (connection, fwd), NULL,
+ GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
+ sizeof (struct GNUNET_MESH_ConnectionACK),
+ connection, NULL, fwd,
+ &message_sent, NULL);
+ if (MESH_TUNNEL3_NEW == GMT_get_state (t))
+ GMT_change_state (t, MESH_TUNNEL3_WAITING);
+ if (MESH_CONNECTION_READY != connection->state)
+ connection_change_state (connection, MESH_CONNECTION_SENT);
+}
+
-static size_t
-queue_send (void *cls, size_t size, void *buf)
+/**
+ * Send keepalive packets for a connection.
+ *
+ * @param c Connection to keep alive..
+ * @param fwd Is this a FWD keepalive? (owner -> dest).
+ */
+static void
+connection_keepalive (struct MeshConnection *c, int fwd)
{
- 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;
+ struct GNUNET_MESH_ConnectionKeepAlive *msg;
+ size_t size = sizeof (struct GNUNET_MESH_ConnectionKeepAlive);
+ char cbuf[size];
uint16_t type;
- int fwd;
- peer->core_transmit = NULL;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Queue send (max %u)\n", size);
+ type = fwd ? GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE :
+ GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE;
- if (NULL == buf || 0 == size)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Buffer size 0.\n");
- return 0;
- }
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "sending %s keepalive for connection %s[%d]\n",
+ fwd ? "FWD" : "BCK", GMT_2s (c->t), c->id);
- /* 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;
+ msg = (struct GNUNET_MESH_ConnectionKeepAlive *) cbuf;
+ msg->header.size = htons (size);
+ msg->header.type = htons (type);
+ msg->cid = c->id;
- 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);
+ GMC_send_prebuilt_message (&msg->header, c, NULL, fwd);
+}
- 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 */
+/**
+ * Send CONNECTION_{CREATE/ACK} packets for a connection.
+ *
+ * @param c Connection for which to send the message.
+ * @param fwd If GNUNET_YES, send CREATE, otherwise send ACK.
+ */
+static void
+connection_recreate (struct MeshConnection *c, int fwd)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "sending connection recreate\n");
+ if (fwd)
+ GMC_send_create (c);
+ else
+ send_connection_ack (c, GNUNET_NO);
+}
+
+
+/**
+ * Generic connection timer management.
+ * Depending on the role of the peer in the connection will send the
+ * appropriate message (build or keepalive)
+ *
+ * @param c Conncetion to maintain.
+ * @param fwd Is FWD?
+ */
+static void
+connection_maintain (struct MeshConnection *c, int fwd)
+{
+ if (MESH_TUNNEL3_SEARCHING == GMT_get_state (c->t))
+ {
+ /* TODO DHT GET with RO_BART */
+ return;
+ }
+ switch (c->state)
+ {
+ case MESH_CONNECTION_NEW:
GNUNET_break (0);
- data_size = 0;
+ case MESH_CONNECTION_SENT:
+ connection_recreate (c, fwd);
+ break;
+ case MESH_CONNECTION_READY:
+ connection_keepalive (c, fwd);
break;
default:
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "* type unknown: %u\n",
- queue->type);
- data_size = 0;
+ break;
+ }
+}
+
+
+static void
+connection_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshConnection *c = cls;
+
+ c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+
+ connection_maintain (c, GNUNET_YES);
+ c->fwd_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
+ &connection_fwd_keepalive,
+ c);
+}
+
+
+static void
+connection_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshConnection *c = cls;
+
+ c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+
+ connection_maintain (c, GNUNET_NO);
+ c->bck_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
+ &connection_bck_keepalive,
+ c);
+}
+
+
+/**
+ * @brief Re-initiate traffic on this connection if necessary.
+ *
+ * Check if there is traffic queued towards this peer
+ * and the core transmit handle is NULL (traffic was stalled).
+ * If so, call core tmt rdy.
+ *
+ * @param c Connection on which initiate traffic.
+ * @param fwd Is this about fwd traffic?
+ */
+static void
+connection_unlock_queue (struct MeshConnection *c, int fwd)
+{
+ struct MeshPeer *peer;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "connection_unlock_queue %s on %s\n",
+ fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
+
+ if (GMC_is_terminal (c, fwd))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " is terminal!\n");
+ return;
+ }
+
+ peer = get_hop (c, fwd);
+ GMP_queue_unlock (peer, c);
+}
+
+
+/**
+ * Cancel all transmissions that belong to a certain connection.
+ *
+ * @param c Connection which to cancel.
+ * @param fwd Cancel fwd traffic?
+ */
+static void
+connection_cancel_queues (struct MeshConnection *c, int fwd)
+{
+
+ struct MeshFlowControl *fc;
+ struct MeshPeer *peer;
+
+ if (NULL == c)
+ {
+ GNUNET_break (0);
+ return;
+ }
+
+ peer = get_hop (c, fwd);
+ GMP_queue_cancel (peer, c);
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+}
+
+
+/**
+ * Function called if a connection has been stalled for a while,
+ * possibly due to a missed ACK. Poll the neighbor about its ACK status.
+ *
+ * @param cls Closure (poll ctx).
+ * @param tc TaskContext.
+ */
+static void
+connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshFlowControl *fc = cls;
+ struct GNUNET_MESH_Poll msg;
+ struct MeshConnection *c;
+
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ {
+ return;
+ }
+
+ c = fc->c;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** Polling!\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** connection [%X]\n",
+ GNUNET_h2s (&c->id));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** %s\n",
+ fc == &c->fwd_fc ? "FWD" : "BCK");
+
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
+ msg.header.size = htons (sizeof (msg));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** pid (%u)!\n", fc->last_pid_sent);
+ GMC_send_prebuilt_message (&msg.header, c, NULL, fc == &c->fwd_fc);
+ fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
+ fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
+ &connection_poll, fc);
+}
+
+
+/**
+ * Timeout function due to lack of keepalive/traffic from the owner.
+ * Destroys connection if called.
+ *
+ * @param cls Closure (connection to destroy).
+ * @param tc TaskContext.
+ */
+static void
+connection_fwd_timeout (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshConnection *c = cls;
+
+ c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s[%X] FWD timed out. Destroying.\n",
+ GMT_2s (c->t),
+ c->id);
+
+ if (GMC_is_origin (c, GNUNET_YES)) /* If local, leave. */
+ return;
+
+ GMC_destroy (c);
+}
+
+
+/**
+ * Timeout function due to lack of keepalive/traffic from the destination.
+ * Destroys connection if called.
+ *
+ * @param cls Closure (connection to destroy).
+ * @param tc TaskContext
+ */
+static void
+connection_bck_timeout (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshConnection *c = cls;
+
+ c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s[%X] FWD timed out. Destroying.\n",
+ GMT_2s (c->t), c->id);
+
+ if (GMC_is_origin (c, GNUNET_NO)) /* If local, leave. */
+ return;
+
+ GMC_destroy (c);
+}
+
+
+/**
+ * Resets the connection timeout task, some other message has done the
+ * task's job.
+ * - For the first peer on the direction this means to send
+ * a keepalive or a path confirmation message (either create or ACK).
+ * - For all other peers, this means to destroy the connection,
+ * due to lack of activity.
+ * Starts the tiemout if no timeout was running (connection just created).
+ *
+ * @param c Connection whose timeout to reset.
+ * @param fwd Is this forward?
+ *
+ * TODO use heap to improve efficiency of scheduler.
+ */
+static void
+connection_reset_timeout (struct MeshConnection *c, int fwd)
+{
+ GNUNET_SCHEDULER_TaskIdentifier *ti;
+ GNUNET_SCHEDULER_Task f;
+
+ ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
+
+ if (GNUNET_SCHEDULER_NO_TASK != *ti)
+ GNUNET_SCHEDULER_cancel (*ti);
+
+ if (GMC_is_origin (c, fwd)) /* Endpoint */
+ {
+ f = fwd ? &connection_fwd_keepalive : &connection_bck_keepalive;
+ *ti = GNUNET_SCHEDULER_add_delayed (refresh_connection_time, f, c);
+ }
+ else /* Relay */
+ {
+ struct GNUNET_TIME_Relative delay;
+
+ delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
+ f = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
+ *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
+ }
+}
+
+
+/**
+ * Add the connection to the list of both neighbors.
+ *
+ * @param c Connection.
+ */
+static void
+register_neighbors (struct MeshConnection *c)
+{
+ struct MeshPeer *peer;
+
+ peer = get_next_hop (c);
+ if (GNUNET_NO == GMP_is_neighbor (peer))
+ {
+ GMC_destroy (c);
+ return;
+ }
+ GMP_add_connection (peer, c);
+ peer = get_prev_hop (c);
+ if (GNUNET_NO == GMP_is_neighbor (peer))
+ {
+ GMC_destroy (c);
+ return;
+ }
+ GMP_add_connection (peer, c);
+}
+
+
+/**
+ * Remove the connection from the list of both neighbors.
+ *
+ * @param c Connection.
+ */
+static void
+unregister_neighbors (struct MeshConnection *c)
+{
+ struct MeshPeer *peer;
+
+ peer = get_next_hop (c);
+ GMP_remove_connection (peer, c);
+
+ peer = get_prev_hop (c);
+ GMP_remove_connection (peer, c);
+
+}
+
+
+/**
+ * Bind the connection to the peer and the tunnel to that peer.
+ *
+ * If the peer has no tunnel, create one. Update tunnel and connection
+ * data structres to reflect new status.
+ *
+ * @param c Connection.
+ * @param peer Peer.
+ */
+static void
+add_to_peer (struct MeshConnection *c, struct MeshPeer *peer)
+{
+ GMP_add_tunnel (peer);
+ c->t = GMP_get_tunnel (peer);
+ GMT_add_connection (c->t, c);
+}
+
+/******************************************************************************/
+/******************************** API ***********************************/
+/******************************************************************************/
+
+/**
+ * Core handler for connection creation.
+ *
+ * @param cls Closure (unused).
+ * @param peer Sender (neighbor).
+ * @param message Message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_create (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_ConnectionCreate *msg;
+ struct GNUNET_PeerIdentity *id;
+ struct GNUNET_HashCode *cid;
+ struct MeshPeerPath *path;
+ struct MeshPeer *dest_peer;
+ struct MeshPeer *orig_peer;
+ struct MeshConnection *c;
+ unsigned int own_pos;
+ uint16_t size;
+ uint16_t i;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a connection create msg\n");
+
+ /* Check size */
+ size = ntohs (message->size);
+ if (size < sizeof (struct GNUNET_MESH_ConnectionCreate))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
}
- if (0 < drop_percent &&
- GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK, 101) < drop_percent)
+ /* Calculate hops */
+ size -= sizeof (struct GNUNET_MESH_ConnectionCreate);
+ if (size % sizeof (struct GNUNET_PeerIdentity))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ size /= sizeof (struct GNUNET_PeerIdentity);
+ if (1 > size)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " path has %u hops.\n", size);
+
+ /* Get parameters */
+ msg = (struct GNUNET_MESH_ConnectionCreate *) message;
+ cid = &msg->cid;
+ id = (struct GNUNET_PeerIdentity *) &msg[1];
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " connection %s (%s).\n",
+ GNUNET_h2s (cid), GNUNET_i2s (id));
+
+ /* Create connection */
+ c = connection_get (cid);
+ if (NULL == c)
+ {
+ /* Create path */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Creating path...\n");
+ path = path_new (size);
+ own_pos = 0;
+ for (i = 0; i < size; i++)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ... adding %s\n",
+ GNUNET_i2s (&id[i]));
+ path->peers[i] = GNUNET_PEER_intern (&id[i]);
+ if (path->peers[i] == myid)
+ own_pos = i;
+ }
+ if (own_pos == 0 && path->peers[own_pos] != myid)
+ {
+ /* create path: self not found in path through self */
+ GNUNET_break_op (0);
+ path_destroy (path);
+ return GNUNET_OK;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Own position: %u\n", own_pos);
+ GMP_add_path_to_all (path, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Creating connection\n");
+ c = GMC_new (cid, NULL, path_duplicate (path), own_pos);
+ if (NULL == c)
+ return GNUNET_OK;
+ connection_reset_timeout (c, GNUNET_YES);
+ }
+ else
{
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Dropping message of type %s\n",
- GNUNET_MESH_DEBUG_M2S (queue->type));
- data_size = 0;
+ path = NULL;
}
+ if (MESH_CONNECTION_NEW == c->state)
+ connection_change_state (c, MESH_CONNECTION_SENT);
- /* Free queue, but cls was freed by send_core_* */
- ch = queue->ch;
- queue_destroy (queue, GNUNET_NO);
+ /* Remember peers */
+ dest_peer = GMP_get (&id[size - 1]);
+ orig_peer = GMP_get (&id[0]);
- /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
- switch (type)
+ /* Is it a connection to us? */
+ if (c->own_pos == size - 1)
{
- 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;
- default:
- break;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " It's for us!\n");
+ GMP_add_path_to_origin (orig_peer, path, GNUNET_YES);
+
+ add_to_peer (c, orig_peer);
+ if (MESH_TUNNEL3_NEW == GMT_get_state (c->t))
+ GMT_change_state (c->t, MESH_TUNNEL3_WAITING);
+
+ send_connection_ack (c, GNUNET_NO);
+ if (MESH_CONNECTION_SENT == c->state)
+ connection_change_state (c, MESH_CONNECTION_ACK);
+
+ /* Keep tunnel alive in direction dest->owner*/
+ connection_reset_timeout (c, GNUNET_NO);
+ }
+ else
+ {
+ /* It's for somebody else! Retransmit. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Retransmitting.\n");
+ GMP_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
+ GMP_add_path_to_origin (orig_peer, path, GNUNET_NO);
+ GMC_send_prebuilt_message (message, c, NULL, GNUNET_YES);
}
+ return GNUNET_OK;
+}
- /* 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);
- }
+
+/**
+ * Core handler for path confirmations.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_confirm (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_ConnectionACK *msg;
+ struct MeshConnection *c;
+ struct MeshPeerPath *p;
+ struct MeshPeer *pi;
+ int fwd;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a connection ACK msg\n");
+ msg = (struct GNUNET_MESH_ConnectionACK *) message;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " on connection %s\n",
+ GNUNET_h2s (&msg->cid));
+ c = connection_get (&msg->cid);
+ if (NULL == c)
+ {
+ GNUNET_STATISTICS_update (stats, "# control on unknown connection",
+ 1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " don't know the connection!\n");
+ return GNUNET_OK;
+ }
+
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " via peer %s\n",
+ GNUNET_i2s (peer));
+ pi = GMP_get (peer);
+ if (get_next_hop (c) == pi)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " SYNACK\n");
+ fwd = GNUNET_NO;
+ if (MESH_CONNECTION_SENT == c->state)
+ connection_change_state (c, MESH_CONNECTION_ACK);
+ }
+ else if (get_prev_hop (c) == pi)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK\n");
+ fwd = GNUNET_YES;
+ connection_change_state (c, MESH_CONNECTION_READY);
+ }
+ else
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ connection_reset_timeout (c, fwd);
+
+ /* Add path to peers? */
+ p = c->path;
+ if (NULL != p)
+ {
+ GMP_add_path_to_all (p, GNUNET_YES);
+ }
+ else
+ {
+ GNUNET_break (0);
+ }
+
+ /* Message for us as creator? */
+ if (GMC_is_origin (c, GNUNET_YES))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Connection (SYN)ACK for us!\n");
+ connection_change_state (c, MESH_CONNECTION_READY);
+ GMT_change_state (c->t, MESH_TUNNEL3_READY);
+ send_connection_ack (c, GNUNET_YES);
+ GMT_send_queued_data (c->t, GNUNET_YES);
+ return GNUNET_OK;
+ }
+
+ /* Message for us as destination? */
+ if (GMC_is_terminal (c, GNUNET_YES))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Connection ACK for us!\n");
+ connection_change_state (c, MESH_CONNECTION_READY);
+ GMT_change_state (c->t, MESH_TUNNEL3_READY);
+ GMT_send_queued_data (c->t, GNUNET_NO);
+ return GNUNET_OK;
+ }
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " not for us, retransmitting...\n");
+ GMC_send_prebuilt_message (message, c, NULL, fwd);
+ return GNUNET_OK;
+}
+
+
+/**
+ * Is traffic coming from this sender 'FWD' traffic?
+ *
+ * @param c Connection to check.
+ * @param sender Peer identity of neighbor.
+ *
+ * @return GNUNET_YES in case the sender is the 'prev' hop and therefore
+ * the traffic is 'FWD'. GNUNET_NO for BCK. GNUNET_SYSERR for errors.
+ */
+int
+is_fwd (const struct MeshConnection *c,
+ const struct GNUNET_PeerIdentity *sender)
+{
+ GNUNET_PEER_Id id;
+
+ id = GNUNET_PEER_search (sender);
+ if (GMP_get_short_id (get_prev_hop (c)) == id)
+ return GNUNET_YES;
+
+ if (GMP_get_short_id (get_next_hop (c)) == id)
+ return GNUNET_NO;
+
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+}
+
+
+/**
+ * Core handler for notifications of broken paths
+ *
+ * @param cls Closure (unused).
+ * @param id Peer identity of sending neighbor.
+ * @param message Message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_broken (void* cls,
+ const struct GNUNET_PeerIdentity* id,
+ const struct GNUNET_MessageHeader* message)
+{
+ struct GNUNET_MESH_ConnectionBroken *msg;
+ struct MeshConnection *c;
+ int fwd;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Received a CONNECTION BROKEN msg from %s\n", GNUNET_i2s (id));
+ msg = (struct GNUNET_MESH_ConnectionBroken *) message;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " regarding %s\n",
+ GNUNET_i2s (&msg->peer1));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " regarding %s\n",
+ GNUNET_i2s (&msg->peer2));
+ c = connection_get (&msg->cid);
+ if (NULL == c)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+
+ fwd = is_fwd (c, id);
+ connection_cancel_queues (c, !fwd);
+ if (GMC_is_terminal (c, fwd))
+ {
+ if (0 < c->pending_messages)
+ c->destroy = GNUNET_YES;
else
+ GMC_destroy (c);
+ }
+ else
+ {
+ GMC_send_prebuilt_message (message, c, NULL, fwd);
+ c->destroy = GNUNET_YES;
+ }
+
+ return GNUNET_OK;
+
+}
+
+
+/**
+ * Core handler for tunnel destruction
+ *
+ * @param cls Closure (unused).
+ * @param peer Peer identity of sending neighbor.
+ * @param message Message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_destroy (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_ConnectionDestroy *msg;
+ struct MeshConnection *c;
+ int fwd;
+
+ msg = (struct GNUNET_MESH_ConnectionDestroy *) message;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got a CONNECTION DESTROY message from %s\n",
+ GNUNET_i2s (peer));
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " for connection %s\n",
+ GNUNET_h2s (&msg->cid));
+ c = connection_get (&msg->cid);
+ if (NULL == c)
+ {
+ /* Probably already got the message from another path,
+ * destroyed the tunnel and retransmitted to children.
+ * Safe to ignore.
+ */
+ GNUNET_STATISTICS_update (stats, "# control on unknown tunnel",
+ 1, GNUNET_NO);
+ return GNUNET_OK;
+ }
+ fwd = is_fwd (c, peer);
+ if (GNUNET_SYSERR == fwd)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ GMC_send_prebuilt_message (message, c, NULL, fwd);
+ c->destroy = GNUNET_YES;
+
+ return GNUNET_OK;
+}
+
+/**
+ * Generic handler for mesh network encrypted traffic.
+ *
+ * @param peer Peer identity this notification is about.
+ * @param message Encrypted message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_mesh_encrypted (const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MESH_Encrypted *msg)
+{
+ struct MeshConnection *c;
+ struct MeshPeer *neighbor;
+ struct MeshFlowControl *fc;
+ GNUNET_PEER_Id peer_id;
+ uint32_t pid;
+ uint32_t ttl;
+ uint16_t type;
+ size_t size;
+ int fwd;
+
+ /* Check size */
+ size = ntohs (msg->header.size);
+ if (size <
+ sizeof (struct GNUNET_MESH_Encrypted) +
+ sizeof (struct GNUNET_MessageHeader))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ type = ntohs (msg->header.type);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "got a %s message from %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), GNUNET_i2s (peer));
+
+ /* Check connection */
+ c = connection_get (&msg->cid);
+ if (NULL == c)
+ {
+ GNUNET_STATISTICS_update (stats, "# unknown connection", 1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "WARNING connection unknown\n");
+ return GNUNET_OK;
+ }
+
+ /* Check if origin is as expected */
+ neighbor = get_prev_hop (c);
+ peer_id = GNUNET_PEER_search (peer);
+ if (peer_id == GMP_get_short_id (neighbor))
+ {
+ fwd = GNUNET_YES;
+ }
+ else
+ {
+ neighbor = get_next_hop (c);
+ if (peer_id == GMP_get_short_id (neighbor))
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "* tmt rdy called somewhere else\n");
+ fwd = GNUNET_NO;
}
- if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
+ else
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* starting poll timeout\n");
- fc->poll_task =
- GNUNET_SCHEDULER_add_delayed (fc->poll_time, &connection_poll, fc);
+ GNUNET_break_op (0);
+ return GNUNET_OK;
}
}
- else
+ fc = fwd ? &c->bck_fc : &c->fwd_fc;
+
+ /* Check PID */
+ pid = ntohl (msg->pid);
+ if (GMC_is_pid_bigger (pid, fc->last_ack_sent))
{
- if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
- {
- GNUNET_SCHEDULER_cancel (fc->poll_task);
- fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
- }
+ GNUNET_STATISTICS_update (stats, "# unsolicited message", 1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "WARNING Received PID %u, (prev %u), ACK %u\n",
+ pid, fc->last_pid_recv, fc->last_ack_sent);
+ return GNUNET_OK;
}
- if (NULL != c)
+ if (GNUNET_NO == GMC_is_pid_bigger (pid, fc->last_pid_recv))
+ {
+ GNUNET_STATISTICS_update (stats, "# duplicate PID", 1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " Pid %u not expected (%u+), dropping!\n",
+ pid, fc->last_pid_recv + 1);
+ return GNUNET_OK;
+ }
+ if (MESH_CONNECTION_SENT == c->state)
+ connection_change_state (c, MESH_CONNECTION_READY);
+ connection_reset_timeout (c, fwd);
+ fc->last_pid_recv = pid;
+
+ /* Is this message for us? */
+ if (GMC_is_terminal (c, fwd))
{
- c->pending_messages--;
- if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
+ /* TODO signature verification */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " message for us!\n");
+ GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
+
+ if (NULL == c->t)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying connection!\n");
- GMC_destroy (c);
+ GNUNET_break (0);
+ return GNUNET_OK;
}
+ fc->last_pid_recv = pid;
+ GMT_handle_encrypted (c->t, msg, fwd);
+ GMC_send_ack (c, NULL, fwd);
+ return GNUNET_OK;
+ }
+
+ /* Message not for us: forward to next hop */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " not for us, retransmitting...\n");
+ ttl = ntohl (msg->ttl);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ttl: %u\n", ttl);
+ if (ttl == 0)
+ {
+ GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
+ GMC_send_ack (c, NULL, fwd);
+ return GNUNET_OK;
+ }
+ GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
+
+ GMC_send_prebuilt_message (&msg->header, c, NULL, fwd);
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Core handler for encrypted mesh network traffic (channel mgmt, data).
+ *
+ * @param cls Closure (unused).
+ * @param message Message received.
+ * @param peer Peer who sent the message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_encrypted (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ return handle_mesh_encrypted (peer,
+ (struct GNUNET_MESH_Encrypted *)message);
+}
+
+
+/**
+ * Core handler for mesh network traffic point-to-point acks.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_ACK *msg;
+ struct MeshConnection *c;
+ struct MeshFlowControl *fc;
+ GNUNET_PEER_Id id;
+ uint32_t ack;
+ int fwd;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK packet from %s!\n",
+ GNUNET_i2s (peer));
+ msg = (struct GNUNET_MESH_ACK *) message;
+
+ c = connection_get (&msg->cid);
+
+ if (NULL == c)
+ {
+ GNUNET_STATISTICS_update (stats, "# ack on unknown connection", 1,
+ GNUNET_NO);
+ return GNUNET_OK;
+ }
+
+ /* Is this a forward or backward ACK? */
+ id = GNUNET_PEER_search (peer);
+ if (GMP_get_short_id (get_next_hop (c)) == id)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK\n");
+ fc = &c->fwd_fc;
+ fwd = GNUNET_YES;
+ }
+ else if (GMP_get_short_id (get_prev_hop (c)) == id)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
+ fc = &c->bck_fc;
+ fwd = GNUNET_NO;
+ }
+ else
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+
+ ack = ntohl (msg->ack);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK %u (was %u)\n",
+ ack, fc->last_ack_recv);
+ if (GMC_is_pid_bigger (ack, fc->last_ack_recv))
+ fc->last_ack_recv = ack;
+
+ /* Cancel polling if the ACK is big enough. */
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task &&
+ GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Cancel poll\n");
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
+ }
+
+ connection_unlock_queue (c, fwd);
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Core handler for mesh network traffic point-to-point ack polls.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+int
+GMC_handle_poll (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_Poll *msg;
+ struct MeshConnection *c;
+ struct MeshFlowControl *fc;
+ GNUNET_PEER_Id id;
+ uint32_t pid;
+ int fwd;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Got a POLL packet from %s!\n",
+ GNUNET_i2s (peer));
+
+ msg = (struct GNUNET_MESH_Poll *) message;
+
+ c = connection_get (&msg->cid);
+
+ if (NULL == c)
+ {
+ GNUNET_STATISTICS_update (stats, "# poll on unknown connection", 1,
+ GNUNET_NO);
+ GNUNET_break_op (0);
+ return GNUNET_OK;
}
- if (NULL != t)
+ /* Is this a forward or backward ACK?
+ * Note: a poll should never be needed in a loopback case,
+ * since there is no possiblility of packet loss there, so
+ * this way of discerining FWD/BCK should not be a problem.
+ */
+ id = GNUNET_PEER_search (peer);
+ if (GMP_get_short_id (get_next_hop (c)) == id)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK\n");
+ fc = &c->fwd_fc;
+ }
+ else if (GMP_get_short_id (get_prev_hop (c)) == id)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
+ fc = &c->bck_fc;
+ }
+ else
{
- t->pending_messages--;
- if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
- {
-// GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying tunnel!\n");
- tunnel_destroy (t);
- }
+ GNUNET_break_op (0);
+ return GNUNET_OK;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Return %d\n", data_size);
- return data_size;
-}
+ pid = ntohl (msg->pid);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " PID %u, OLD %u\n",
+ pid, fc->last_pid_recv);
+ fc->last_pid_recv = pid;
+ fwd = fc == &c->fwd_fc;
+ GMC_send_ack (c, NULL, fwd);
+ return GNUNET_OK;
+}
-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);
+/**
+ * Core handler for mesh keepalives.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ *
+ * TODO: Check who we got this from, to validate route.
+ */
+int
+GMC_handle_keepalive (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MESH_ConnectionKeepAlive *msg;
+ struct MeshConnection *c;
+ struct MeshPeer *neighbor;
+ int fwd;
- fc = fwd ? &c->fwd_fc : &c->bck_fc;
- peer = fwd ? connection_get_next_hop (c) : connection_get_prev_hop (c);
+ msg = (struct GNUNET_MESH_ConnectionKeepAlive *) message;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "got a keepalive packet from %s\n",
+ GNUNET_i2s (peer));
- if (NULL == fc)
+ c = connection_get (&msg->cid);
+ if (NULL == c)
{
- GNUNET_break (0);
- return;
+ GNUNET_STATISTICS_update (stats, "# keepalive on unknown connection", 1,
+ GNUNET_NO);
+ return GNUNET_OK;
}
-
- if (NULL == peer->connections)
+
+ fwd = GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE == ntohs (message->type) ?
+ GNUNET_YES : GNUNET_NO;
+
+ /* Check if origin is as expected */
+ neighbor = get_hop (c, fwd);
+ if (GNUNET_PEER_search (peer) != GMP_get_short_id (neighbor))
{
- /* We are not connected to this peer, ignore request. */
GNUNET_break_op (0);
- return;
+ return GNUNET_OK;
}
- priority = 0;
+ connection_change_state (c, MESH_CONNECTION_READY);
+ connection_reset_timeout (c, fwd);
+
+ if (GMC_is_terminal (c, fwd))
+ return GNUNET_OK;
+
+ GNUNET_STATISTICS_update (stats, "# keepalives forwarded", 1, GNUNET_NO);
+ GMC_send_prebuilt_message (message, c, NULL, fwd);
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Send an ACK on the appropriate connection/channel, depending on
+ * the direction and the position of the peer.
+ *
+ * @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)
+ */
+void
+GMC_send_ack (struct MeshConnection *c, struct MeshChannel *ch, int fwd)
+{
+ unsigned int buffer;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "send ack %s on %p %p\n",
+ fwd ? "FWD" : "BCK", c, ch);
- if (GNUNET_MESSAGE_TYPE_MESH_POLL == type ||
- GNUNET_MESSAGE_TYPE_MESH_ACK == type)
+ /* Get available bufffer space */
+ if (NULL == c || GMC_is_terminal (c, fwd))
{
- priority = 100;
+ struct MeshTunnel3 *t;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " getting from all connections\n");
+ t = (NULL == c) ? GMCH_get_tunnel (ch) : GMC_get_tunnel (c);
+ buffer = GMT_get_buffer (t, fwd);
}
-
- 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)
+ else
{
- 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 */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " getting from one connection\n");
+ buffer = GMC_get_buffer (c, fwd);
}
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " buffer available: %u\n", buffer);
- 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))
+ /* Send available buffer space */
+ if ( (NULL != ch && GMCH_is_origin (ch, fwd)) ||
+ (NULL != c && GMC_is_origin (c, fwd)) )
{
- 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)
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on channel...\n");
+ if (0 < buffer)
{
- 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_assert (NULL != ch);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " really sending!\n");
+ GMCH_send_data_ack (ch, fwd);
}
- 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)
+ else if (NULL == c)
{
- 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);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on all connections\n");
+ GNUNET_assert (NULL != ch);
+ GMT_send_acks (GMCH_get_tunnel (ch), buffer, fwd);
}
else
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "core tmt rdy towards %s already called\n",
- peer2s (peer));
-
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on connection\n");
+ connection_send_ack (c, buffer, fwd);
}
- 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 the connections subsystem
*
- * @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 c Configuration handle.
*/
-static void
-send_prebuilt_message_connection (const struct GNUNET_MessageHeader *message,
- struct MeshConnection *c,
- struct MeshChannel *ch,
- int fwd)
+void
+GMC_init (const struct GNUNET_CONFIGURATION_Handle *c)
{
- 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)
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "init\n");
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_MSGS_QUEUE",
+ &max_msgs_queue))
{
- 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;
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "MESH", "MAX_MSGS_QUEUE", "MISSING");
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
- default:
- GNUNET_break (0);
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_CONNECTIONS",
+ &max_connections))
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "MESH", "MAX_CONNECTIONS", "MISSING");
+ GNUNET_SCHEDULER_shutdown ();
+ return;
}
- queue_add (data,
- type,
- size,
- c,
- ch,
- fwd);
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_time (c, "MESH", "REFRESH_CONNECTION_TIME",
+ &refresh_connection_time))
+ {
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ "MESH", "REFRESH_CONNECTION_TIME", "MISSING");
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
}
-
+/**
+ * Shut down the connections subsystem.
+ */
+void
+GMC_shutdown (void)
+{
+ GNUNET_CONTAINER_multihashmap_destroy (connections);
+}
struct MeshConnection *
-GMC_new (const struct GNUNET_HashCode *cid)
+GMC_new (const struct GNUNET_HashCode *cid,
+ struct MeshTunnel3 *t,
+ struct MeshPeerPath *p,
+ unsigned int own_pos)
{
struct MeshConnection *c;
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);
-}
-
-
-
-/**
- * Send an ACK informing the predecessor about the available buffer space.
- *
- * Note that for fwd ack, the FWD mean forward *traffic* (root->dest),
- * the ACK itself goes "back" (dest->root).
- *
- * @param c Connection on which to send the ACK.
- * @param buffer How much space free to advertise?
- * @param fwd Is this FWD ACK? (Going dest->owner)
- */
-static void
-connection_send_ack (struct MeshConnection *c, unsigned int buffer, int fwd)
-{
- struct MeshFlowControl *next_fc;
- struct MeshFlowControl *prev_fc;
- struct GNUNET_MESH_ACK msg;
- uint32_t ack;
- int delta;
-
- next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
- prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "connection send %s ack on %s\n",
- fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
-
- /* Check if we need to transmit the ACK */
- if (prev_fc->last_ack_sent - prev_fc->last_pid_recv > 3)
+ c->t = t;
+ if (own_pos > p->length - 1)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer > 3\n");
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " last pid recv: %u, last ack sent: %u\n",
- prev_fc->last_pid_recv, prev_fc->last_ack_sent);
- return;
+ GNUNET_break (0);
+ GMC_destroy (c);
+ return NULL;
}
+ c->own_pos = own_pos;
+ c->path = p;
- /* Ok, ACK might be necessary, what PID to ACK? */
- delta = next_fc->queue_max - next_fc->queue_n;
- ack = prev_fc->last_pid_recv + delta;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " last pid %u, last ack %u, qmax %u, q %u\n",
- prev_fc->last_pid_recv, prev_fc->last_ack_sent,
- next_fc->queue_max, next_fc->queue_n);
- if (ack == prev_fc->last_ack_sent)
+ if (0 == own_pos)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
- return;
+ c->fwd_maintenance_task =
+ GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
+ &connection_fwd_keepalive, c);
}
-
- prev_fc->last_ack_sent = ack;
-
- /* Build ACK message and send on connection */
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
- msg.ack = htonl (ack);
- msg.cid = c->id;
-
- send_prebuilt_message_connection (&msg.header, c, NULL, !fwd);
-}
-
-
-static void
-connection_change_state (struct MeshConnection* c,
- enum MeshConnectionState state)
-{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Connection %s state was %s\n",
- GNUNET_h2s (&c->id), GNUNET_MESH_DEBUG_CS2S (c->state));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Connection %s state is now %s\n",
- GNUNET_h2s (&c->id), GNUNET_MESH_DEBUG_CS2S (state));
- c->state = state;
+ register_neighbors (c);
+ return c;
}
-
-/**
- * Send keepalive packets for a connection.
- *
- * @param c Connection to keep alive..
- * @param fwd Is this a FWD keepalive? (owner -> dest).
- */
-static void
-connection_keepalive (struct MeshConnection *c, int fwd)
+void
+GMC_destroy (struct MeshConnection *c)
{
- struct GNUNET_MESH_ConnectionKeepAlive *msg;
- size_t size = sizeof (struct GNUNET_MESH_ConnectionKeepAlive);
- char cbuf[size];
- uint16_t type;
+ if (NULL == c)
+ return;
- type = fwd ? GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE :
- GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s\n",
+ GNUNET_h2s (&c->id));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "sending %s keepalive for connection %s[%d]\n",
- fwd ? "FWD" : "BCK",
- peer2s (c->t->peer),
- c->id);
+ /* Cancel all traffic */
+ connection_cancel_queues (c, GNUNET_YES);
+ connection_cancel_queues (c, GNUNET_NO);
- msg = (struct GNUNET_MESH_ConnectionKeepAlive *) cbuf;
- msg->header.size = htons (size);
- msg->header.type = htons (type);
- msg->cid = c->id;
+ /* 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);
- send_prebuilt_message_connection (&msg->header, c, NULL, fwd);
-}
+ /* Unregister from neighbors */
+ unregister_neighbors (c);
+ /* Delete */
+ GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
+ GMT_remove_connection (c->t, c);
+ GNUNET_free (c);
+}
/**
- * Send CONNECTION_{CREATE/ACK} packets for a connection.
+ * Get the connection ID.
*
- * @param c Connection for which to send the message.
- * @param fwd If GNUNET_YES, send CREATE, otherwise send ACK.
+ * @param c Connection to get the ID from.
+ *
+ * @return ID of the connection.
*/
-static void
-connection_recreate (struct MeshConnection *c, int fwd)
+const struct GNUNET_HashCode *
+GMC_get_id (const struct MeshConnection *c)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending connection recreate\n");
- if (fwd)
- send_connection_create (c);
- else
- send_connection_ack (c, GNUNET_NO);
+ return &c->id;
}
/**
- * Generic connection timer management.
- * Depending on the role of the peer in the connection will send the
- * appropriate message (build or keepalive)
+ * Get the connection path.
*
- * @param c Conncetion to maintain.
- * @param fwd Is FWD?
+ * @param c Connection to get the path from.
+ *
+ * @return path used by the connection.
*/
-static void
-connection_maintain (struct MeshConnection *c, int fwd)
-{
- if (MESH_TUNNEL_SEARCHING == c->t->state)
- {
- /* TODO DHT GET with RO_BART */
- return;
- }
- switch (c->state)
- {
- case MESH_CONNECTION_NEW:
- GNUNET_break (0);
- case MESH_CONNECTION_SENT:
- connection_recreate (c, fwd);
- break;
- case MESH_CONNECTION_READY:
- connection_keepalive (c, fwd);
- break;
- default:
- break;
- }
-}
-
-
-static void
-connection_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+const struct MeshPeerPath *
+GMC_get_path (const struct MeshConnection *c)
{
- struct MeshConnection *c = cls;
-
- c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
-
- connection_maintain (c, GNUNET_YES);
- c->fwd_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
- &connection_fwd_keepalive,
- c);
+ return c->path;
}
-static void
-connection_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+/**
+ * Get the connection state.
+ *
+ * @param c Connection to get the state from.
+ *
+ * @return state of the connection.
+ */
+enum MeshConnectionState
+GMC_get_state (const struct MeshConnection *c)
{
- struct MeshConnection *c = cls;
-
- c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
-
- connection_maintain (c, GNUNET_NO);
- c->bck_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
- &connection_bck_keepalive,
- c);
+ return c->state;
}
-
/**
- * Send a message to all peers in this connection that the connection
- * is no longer valid.
+ * Get the connection tunnel.
*
- * If some peer should not receive the message, it should be zero'ed out
- * before calling this function.
+ * @param c Connection to get the tunnel from.
*
- * @param c The connection whose peers to notify.
+ * @return tunnel of the connection.
*/
-static void
-connection_send_destroy (struct MeshConnection *c)
+struct MeshTunnel3 *
+GMC_get_tunnel (const struct MeshConnection *c)
{
- struct GNUNET_MESH_ConnectionDestroy msg;
-
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY);;
- msg.cid = c->id;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " sending connection destroy for connection %s[%X]\n",
- peer2s (c->t->peer),
- c->id);
-
- if (GNUNET_NO == GMC_is_terminal (c, GNUNET_YES))
- send_prebuilt_message_connection (&msg.header, c, NULL, GNUNET_YES);
- if (GNUNET_NO == GMC_is_terminal (c, GNUNET_NO))
- send_prebuilt_message_connection (&msg.header, c, NULL, GNUNET_NO);
- c->destroy = GNUNET_YES;
+ return c->t;
}
*
* @return Free buffer space [0 - max_msgs_queue/max_connections]
*/
-static unsigned int
-connection_get_buffer (struct MeshConnection *c, int fwd)
+unsigned int
+GMC_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);
}
-
/**
- * Get the first transmittable message for a connection.
+ * Get how many messages have we allowed to send to us from a direction..
*
* @param c Connection.
- * @param fwd Is this FWD?
+ * @param fwd Are we asking about traffic from FWD (BCK messages)?
*
- * @return First transmittable message.
+ * @return last_ack_sent - last_pid_recv
*/
-static struct MeshPeerQueue *
-connection_get_first_message (struct MeshConnection *c, int fwd)
+unsigned int
+GMC_get_allowed (struct MeshConnection *c, int fwd)
{
- struct MeshPeerQueue *q;
- struct MeshPeer *p;
-
- p = connection_get_hop (c, fwd);
+ struct MeshFlowControl *fc;
- for (q = p->queue_head; NULL != q; q = q->next)
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GMC_is_pid_bigger(fc->last_pid_recv, fc->last_ack_sent))
{
- if (q->c != c)
- continue;
- if (queue_is_sendable (q))
- return q;
+ return 0;
}
-
- return NULL;
+ return (fc->last_ack_sent - fc->last_pid_recv);
}
-
/**
- * @brief Re-initiate traffic on this connection if necessary.
+ * Get messages queued in a connection.
*
- * Check if there is traffic queued towards this peer
- * and the core transmit handle is NULL (traffic was stalled).
- * If so, call core tmt rdy.
+ * @param c Connection.
+ * @param fwd Is query about FWD traffic?
*
- * @param c Connection on which initiate traffic.
- * @param fwd Is this about fwd traffic?
+ * @return Number of messages queued.
*/
-static void
-connection_unlock_queue (struct MeshConnection *c, int fwd)
+unsigned int
+GMC_get_qn (struct MeshConnection *c, int fwd)
{
- struct MeshPeer *peer;
- struct MeshPeerQueue *q;
- size_t size;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "connection_unlock_queue %s on %s\n",
- fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
-
- if (GMC_is_terminal (c, fwd))
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " is terminal!\n");
- return;
- }
-
- peer = connection_get_hop (c, fwd);
-
- if (NULL != peer->core_transmit)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " already unlocked!\n");
- return; /* Already unlocked */
- }
+ struct MeshFlowControl *fc;
- q = connection_get_first_message (c, fwd);
- if (NULL == q)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " queue empty!\n");
- return; /* Nothing to transmit */
- }
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
- size = q->size;
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready (core_handle,
- GNUNET_NO,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_PEER_resolve2 (peer->id),
- size,
- &queue_send,
- peer);
+ return fc->queue_n;
}
/**
- * Cancel all transmissions that belong to a certain connection.
+ * Allow the connection to advertise a buffer of the given size.
*
- * @param c Connection which to cancel.
- * @param fwd Cancel fwd traffic?
+ * The connection will send an @c fwd ACK message (so: in direction !fwd)
+ * allowing up to last_pid_recv + buffer.
+ *
+ * @param c Connection.
+ * @param buffer How many more messages the connection can accept.
+ * @param fwd Is this about FWD traffic? (The ack will go dest->root).
*/
-static void
-connection_cancel_queues (struct MeshConnection *c, int fwd)
+void
+GMC_allow (struct MeshConnection *c, unsigned int buffer, int fwd)
{
- struct MeshPeerQueue *q;
- struct MeshPeerQueue *next;
- struct MeshFlowControl *fc;
- struct MeshPeer *peer;
-
- if (NULL == c)
- {
- GNUNET_break (0);
- return;
- }
- fc = fwd ? &c->fwd_fc : &c->bck_fc;
- peer = connection_get_hop (c, fwd);
-
- for (q = peer->queue_head; NULL != q; q = next)
- {
- next = q->next;
- if (q->c == c)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "connection_cancel_queue %s\n",
- GNUNET_MESH_DEBUG_M2S (q->type));
- queue_destroy (q, GNUNET_YES);
- }
- }
- if (NULL == peer->queue_head)
- {
- if (NULL != peer->core_transmit)
- {
- GNUNET_CORE_notify_transmit_ready_cancel (peer->core_transmit);
- peer->core_transmit = NULL;
- }
- if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
- {
- GNUNET_SCHEDULER_cancel (fc->poll_task);
- fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
- }
- }
+ connection_send_ack (c, buffer, fwd);
}
-
-
/**
- * Function called if a connection has been stalled for a while,
- * possibly due to a missed ACK. Poll the neighbor about its ACK status.
+ * Send a notification that a connection is broken.
*
- * @param cls Closure (poll ctx).
- * @param tc TaskContext.
+ * @param c Connection that is broken.
+ * @param id1 Peer that has disconnected.
+ * @param id2 Peer that has disconnected.
+ * @param fwd Direction towards which to send it.
*/
static void
-connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+send_broken (struct MeshConnection *c,
+ const struct GNUNET_PeerIdentity *id1,
+ const struct GNUNET_PeerIdentity *id2,
+ int fwd)
{
- struct MeshFlowControl *fc = cls;
- struct GNUNET_MESH_Poll msg;
- struct MeshConnection *c;
-
- fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- {
- return;
- }
-
- c = fc->c;
- 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",
- fc == &c->fwd_fc ? "FWD" : "BCK");
+ struct GNUNET_MESH_ConnectionBroken msg;
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
- msg.header.size = htons (sizeof (msg));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** pid (%u)!\n", fc->last_pid_sent);
- send_prebuilt_message_connection (&msg.header, c, NULL, fc == &c->fwd_fc);
- fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
- fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
- &connection_poll, fc);
+ 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 = *id1;
+ msg.peer2 = *id2;
+ GMC_send_prebuilt_message (&msg.header, c, NULL, fwd);
}
-
-
-
/**
- * Get the previous hop in a connection
- *
- * @param c Connection.
+ * Notify other peers on a connection of a broken link. Mark connections
+ * to destroy after all traffic has been sent.
*
- * @return Previous peer in the connection.
+ * @param c Connection on which there has been a disconnection.
+ * @param peer Peer that disconnected.
*/
-static struct MeshPeer *
-connection_get_prev_hop (struct MeshConnection *c)
+void
+GMC_notify_broken (struct MeshConnection *c,
+ struct MeshPeer *peer)
{
- GNUNET_PEER_Id id;
+ int fwd;
- if (0 == c->own_pos || c->path->length < 2)
- id = c->path->peers[0];
- else
- id = c->path->peers[c->own_pos - 1];
+ fwd = peer == 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;
+ }
- return peer_get_short (id);
+ send_broken (c, &my_full_id, GMP_get_id (peer), fwd);
+
+ /* Connection will have at least one pending message
+ * (the one we just scheduled), so no point in checking whether to
+ * destroy immediately. */
+ c->destroy = GNUNET_YES;
+
+ return;
}
/**
- * Get the next hop in a connection
+ * Is this peer the first one on the connection?
*
* @param c Connection.
+ * @param fwd Is this about fwd traffic?
*
- * @return Next peer in the connection.
+ * @return GNUNET_YES if origin, GNUNET_NO if relay/terminal.
*/
-static struct MeshPeer *
-connection_get_next_hop (struct MeshConnection *c)
+int
+GMC_is_origin (struct MeshConnection *c, int fwd)
{
- GNUNET_PEER_Id id;
-
- if ((c->path->length - 1) == c->own_pos || c->path->length < 2)
- id = c->path->peers[c->path->length - 1];
- else
- id = c->path->peers[c->own_pos + 1];
-
- return peer_get_short (id);
+ if (!fwd && c->path->length - 1 == c->own_pos )
+ return GNUNET_YES;
+ if (fwd && 0 == c->own_pos)
+ return GNUNET_YES;
+ return GNUNET_NO;
}
/**
- * Get the hop in a connection.
+ * Is this peer the last one on the connection?
*
* @param c Connection.
- * @param fwd Next hop?
+ * @param fwd Is this about fwd traffic?
+ * Note that the ROOT is the terminal for BCK traffic!
*
- * @return Next peer in the connection.
+ * @return GNUNET_YES if terminal, GNUNET_NO if relay/origin.
*/
-static struct MeshPeer *
-connection_get_hop (struct MeshConnection *c, int fwd)
+int
+GMC_is_terminal (struct MeshConnection *c, int fwd)
{
- if (fwd)
- return connection_get_next_hop (c);
- return connection_get_prev_hop (c);
+ return GMC_is_origin (c, !fwd);
}
-
-
/**
- * Timeout function due to lack of keepalive/traffic from the owner.
- * Destroys connection if called.
+ * See if we are allowed to send by the next hop in the given direction.
*
- * @param cls Closure (connection to destroy).
- * @param tc TaskContext.
+ * @param c Connection.
+ * @param fwd Is this about fwd traffic?
+ *
+ * @return GNUNET_YES in case it's OK.
*/
-static void
-connection_fwd_timeout (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
+int
+GMC_is_sendable (struct MeshConnection *c, int fwd)
{
- struct MeshConnection *c = cls;
-
- c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Connection %s[%X] FWD timed out. Destroying.\n",
- peer2s (c->t->peer),
- c->id);
-
- if (GMC_is_origin (c, GNUNET_YES)) /* If local, leave. */
- return;
+ struct MeshFlowControl *fc;
- GMC_destroy (c);
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
+ return GNUNET_YES;
+ return GNUNET_NO;
}
-
/**
- * Timeout function due to lack of keepalive/traffic from the destination.
- * Destroys connection if called.
+ * Sends an already built message on a connection, properly registering
+ * all used resources.
*
- * @param cls Closure (connection to destroy).
- * @param tc TaskContext
+ * @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?
*/
-static void
-connection_bck_timeout (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
+void
+GMC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
+ struct MeshConnection *c,
+ struct MeshChannel *ch,
+ int fwd)
{
- struct MeshConnection *c = cls;
+ struct MeshFlowControl *fc;
+ void *data;
+ size_t size;
+ uint16_t type;
+ int droppable;
- c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
+ size = ntohs (message->size);
+ data = GNUNET_malloc (size);
+ memcpy (data, message, size);
+ type = ntohs (message->type);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Send %s (%u) on connection %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), size, GNUNET_h2s (&c->id));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Connection %s[%X] FWD timed out. Destroying.\n",
- peer2s (c->t->peer),
- c->id);
+ droppable = GNUNET_YES;
+ 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;
- if (GMC_is_origin (c, GNUNET_NO)) /* If local, leave. */
- return;
+ case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
+ 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++);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " pid %u\n", ntohl (emsg->pid));
+ break;
- GMC_destroy (c);
-}
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ amsg = (struct GNUNET_MESH_ACK *) data;
+ amsg->cid = c->id;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
+ droppable = GNUNET_NO;
+ 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);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " poll %u\n", ntohl (pmsg->pid));
+ droppable = GNUNET_NO;
+ break;
-/**
- * Resets the connection timeout task, some other message has done the
- * task's job.
- * - For the first peer on the direction this means to send
- * a keepalive or a path confirmation message (either create or ACK).
- * - For all other peers, this means to destroy the connection,
- * due to lack of activity.
- * Starts the tiemout if no timeout was running (connection just created).
- *
- * @param c Connection whose timeout to reset.
- * @param fwd Is this forward?
- *
- * TODO use heap to improve efficiency of scheduler.
- */
-static void
-connection_reset_timeout (struct MeshConnection *c, int fwd)
-{
- GNUNET_SCHEDULER_TaskIdentifier *ti;
- GNUNET_SCHEDULER_Task f;
+ case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
+ dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
+ dmsg->cid = c->id;
+ dmsg->reserved = 0;
+ break;
- ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
+ bmsg = (struct GNUNET_MESH_ConnectionBroken *) data;
+ bmsg->cid = c->id;
+ bmsg->reserved = 0;
+ break;
- if (GNUNET_SCHEDULER_NO_TASK != *ti)
- GNUNET_SCHEDULER_cancel (*ti);
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
+ break;
- if (GMC_is_origin (c, fwd)) /* Endpoint */
- {
- f = fwd ? &connection_fwd_keepalive : &connection_bck_keepalive;
- *ti = GNUNET_SCHEDULER_add_delayed (refresh_connection_time, f, c);
+ default:
+ GNUNET_break (0);
}
- else /* Relay */
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (fc->queue_n >= fc->queue_max && droppable)
{
- struct GNUNET_TIME_Relative delay;
+ GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
+ 1, GNUNET_NO);
+ GNUNET_break (0);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "queue full: %u/%u\n",
+ fc->queue_n, fc->queue_max);
+ return; /* Drop this message */
+ }
- delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
- f = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
- *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid %u\n", fc->last_pid_sent);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", fc->last_ack_recv);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Q_N+ %p %u\n", fc, fc->queue_n);
+ if (GMC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
+ {
+ GMC_start_poll (c, fwd);
}
+ fc->queue_n++;
+ c->pending_messages++;
+
+ GMP_queue_add (get_hop (c, fwd), data, type, size, c, ch, fwd,
+ &message_sent, NULL);
}
/**
- * 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).
+ * Sends a CREATE CONNECTION message for a path to a peer.
+ * Changes the connection and tunnel states if necessary.
*
- * @return GNUNET_YES if we should continue to iterate,
- * GNUNET_NO if not.
+ * @param connection Connection to create.
*/
-int
-GMC_notify_broken (void *cls,
- const struct GNUNET_HashCode *key,
- void *value)
+void
+GMC_send_create (struct MeshConnection *connection)
{
- struct MeshPeer *peer = cls;
- 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);
- send_prebuilt_message_connection (&msg.header, c, NULL, fwd);
- c->destroy = GNUNET_YES;
+enum MeshTunnel3State state;
+ size_t size;
- return GNUNET_YES;
+ size = sizeof (struct GNUNET_MESH_ConnectionCreate);
+ size += connection->path->length * sizeof (struct GNUNET_PeerIdentity);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Send connection create\n");
+ GMP_queue_add (get_next_hop (connection), NULL,
+ GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE,
+ size, connection, NULL,
+ GNUNET_YES, &message_sent, NULL);
+ state = GMT_get_state (connection->t);
+ if (MESH_TUNNEL3_SEARCHING == state || MESH_TUNNEL3_NEW == state)
+ GMT_change_state (connection->t, MESH_TUNNEL3_WAITING);
+ if (MESH_CONNECTION_NEW == connection->state)
+ connection_change_state (connection, MESH_CONNECTION_SENT);
}
/**
- * Initialize the connections subsystem
+ * Send a message to all peers in this connection that the connection
+ * is no longer valid.
*
- * @param c Configuration handle.
+ * If some peer should not receive the message, it should be zero'ed out
+ * before calling this function.
+ *
+ * @param c The connection whose peers to notify.
*/
void
-GMC_init (struct GNUNET_CONFIGURATION_Handle *c)
+GMC_send_destroy (struct MeshConnection *c)
{
- if (GNUNET_OK !=
- GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_MSGS_QUEUE",
- &max_msgs_queue))
- {
- GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
- "MESH", "MAX_MSGS_QUEUE", "MISSING");
- GNUNET_SCHEDULER_shutdown ();
- return;
- }
+ struct GNUNET_MESH_ConnectionDestroy msg;
- if (GNUNET_OK !=
- GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_CONNECTIONS",
- &max_connections))
- {
- GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
- "MESH", "MAX_CONNECTIONS", "MISSING");
- GNUNET_SCHEDULER_shutdown ();
+ if (GNUNET_YES == c->destroy)
return;
- }
- if (GNUNET_OK !=
- GNUNET_CONFIGURATION_get_value_time (c, "MESH", "REFRESH_CONNECTION_TIME",
- &refresh_connection_time))
- {
- GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
- "MESH", "REFRESH_CONNECTION_TIME", "MISSING");
- GNUNET_SCHEDULER_shutdown ();
- return;
- }
- connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY);;
+ msg.cid = c->id;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " sending connection destroy for connection %s\n",
+ GNUNET_h2s (&c->id));
+
+ if (GNUNET_NO == GMC_is_terminal (c, GNUNET_YES))
+ GMC_send_prebuilt_message (&msg.header, c, NULL, GNUNET_YES);
+ if (GNUNET_NO == GMC_is_terminal (c, GNUNET_NO))
+ GMC_send_prebuilt_message (&msg.header, c, NULL, GNUNET_NO);
+ c->destroy = GNUNET_YES;
}
/**
- * Is this peer the first one on the connection?
+ * @brief Start a polling timer for the connection.
*
- * @param c Connection.
- * @param fwd Is this about fwd traffic?
+ * When a neighbor does not accept more traffic on the connection it could be
+ * caused by a simple congestion or by a lost ACK. Polling enables to check
+ * for the lastest ACK status for a connection.
*
- * @return GNUNET_YES if origin, GNUNET_NO if relay/terminal.
+ * @param c Connection.
+ * @param fwd Should we poll in the FWD direction?
*/
-int
-GMC_is_origin (struct MeshConnection *c, int fwd)
+void
+GMC_start_poll (struct MeshConnection *c, int fwd)
{
- if (!fwd && c->path->length - 1 == c->own_pos )
- return GNUNET_YES;
- if (fwd && 0 == c->own_pos)
- return GNUNET_YES;
- return GNUNET_NO;
+ struct MeshFlowControl *fc;
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
+ {
+ return;
+ }
+ fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
+ &connection_poll,
+ fc);
}
/**
- * Is this peer the last one on the connection?
+ * @brief Stop polling a connection for ACKs.
*
- * @param c Connection.
- * @param fwd Is this about fwd traffic?
- * Note that the ROOT is the terminal for BCK traffic!
+ * Once we have enough ACKs for future traffic, polls are no longer necessary.
*
- * @return GNUNET_YES if terminal, GNUNET_NO if relay/origin.
+ * @param c Connection.
+ * @param fwd Should we stop the poll in the FWD direction?
*/
-int
-GMC_is_terminal (struct MeshConnection *c, int fwd)
+void
+GMC_stop_poll (struct MeshConnection *c, int fwd)
{
- return GMC_is_origin (c, !fwd);
+ struct MeshFlowControl *fc;
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
}
\ No newline at end of file
projects / oweals / gnunet.git / blobdiff