#include "gnunet_statistics_service.h"
+#include "mesh_path.h"
+#include "mesh_protocol.h"
+#include "mesh.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"
-#include "mesh_protocol_enc.h"
-#include "mesh_path.h"
#define LOG(level, ...) GNUNET_log_from (level,"mesh-con",__VA_ARGS__)
* How frequently to poll for ACKs.
*/
struct GNUNET_TIME_Relative poll_time;
+
+ /**
+ * Queued poll message, to cancel if not necessary anymore (got ACK).
+ */
+ struct MeshConnectionQueue *poll_msg;
+
+ /**
+ * Queued poll message, to cancel if not necessary anymore (got ACK).
+ */
+ struct MeshConnectionQueue *ack_msg;
};
/**
int destroy;
};
+/**
+ * Handle for messages queued but not yet sent.
+ */
+struct MeshConnectionQueue
+{
+ /**
+ * Peer queue handle, to cancel if necessary.
+ */
+ struct MeshPeerQueue *q;
+
+ /**
+ * Continuation to call once sent.
+ */
+ GMC_sent cont;
+
+ /**
+ * Closure for @c cont.
+ */
+ void *cont_cls;
+};
+
/******************************************************************************/
/******************************* GLOBALS ***********************************/
/******************************************************************************/
*/
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 struct GNUNET_TIME_Relative refresh_connection_time;
+/**
+ * How often to send path create / ACKs.
+ */
+static struct GNUNET_TIME_Relative create_connection_time;
+
/******************************************************************************/
/******************************** STATIC ***********************************/
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s:%X\n",
- peer2s (c->t->peer), GNUNET_h2s (&c->id));
+ peer2s (c->t->peer), GMC_2s (c));
LOG (GNUNET_ERROR_TYPE_DEBUG, " state: %u, pending msgs: %u\n",
c->state, c->pending_messages);
LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD FC\n");
return "MESH_CONNECTION_ACK";
case MESH_CONNECTION_READY:
return "MESH_CONNECTION_READY";
+ case MESH_CONNECTION_DESTROYED:
+ return "MESH_CONNECTION_DESTROYED";
default:
return "MESH_CONNECTION_STATE_ERROR";
}
}
+static void
+connection_change_state (struct MeshConnection* c,
+ enum MeshConnectionState state)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s state was %s\n",
+ GMC_2s (c), GMC_state2s (c->state));
+ if (MESH_CONNECTION_DESTROYED == c->state)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "state not changing anymore\n");
+ return;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connection %s state is now %s\n",
+ GMC_2s (c), GMC_state2s (state));
+ c->state = state;
+}
+
+
+/**
+ * Callback called when a queued ACK message is sent.
+ *
+ * @param cls Closure (FC).
+ * @param c Connection this message was on.
+ * @param q Queue handler this call invalidates.
+ * @param type Type of message sent.
+ * @param fwd Was this a FWD going message?
+ * @param size Size of the message.
+ */
+static void
+ack_sent (void *cls,
+ struct MeshConnection *c,
+ struct MeshConnectionQueue *q,
+ uint16_t type, int fwd, size_t size)
+{
+ struct MeshFlowControl *fc = cls;
+
+ fc->ack_msg = NULL;
+}
+
+
/**
- * Send an ACK informing the predecessor about the available buffer space.
+ * Send an ACK on the connection, informing the predecessor about
+ * the available buffer space. Should not be called in case the peer
+ * is origin (no predecessor) in the @c fwd direction.
*
* 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)
+ * @param fwd Is this FWD ACK? (Going dest -> root)
+ * @param force Don't optimize out.
*/
static void
-connection_send_ack (struct MeshConnection *c, unsigned int buffer, int fwd)
+send_ack (struct MeshConnection *c, unsigned int buffer, int fwd, int force)
{
struct MeshFlowControl *next_fc;
struct MeshFlowControl *prev_fc;
uint32_t ack;
int delta;
+ /* If origin, there is no connection to send ACKs. Wrong function! */
+ if (GMC_is_origin (c, fwd))
+ {
+ GNUNET_break (0);
+ return;
+ }
+
next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"connection send %s ack on %s\n",
- fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
+ fwd ? "FWD" : "BCK", GMC_2s (c));
- /* Check if we need to transmit the ACK */
- if (prev_fc->last_ack_sent - prev_fc->last_pid_recv > 3)
+ /* 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_NO == force)
{
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);
+ " last pid recv: %u, last ack sent: %u\n",
+ prev_fc->last_pid_recv, prev_fc->last_ack_sent);
return;
}
/* 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;
+ 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)
+ " 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 && GNUNET_NO == force)
{
LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
return;
}
+ /* Check if message is already in queue */
+ if (NULL != prev_fc->ack_msg)
+ {
+ if (GMC_is_pid_bigger (ack, prev_fc->last_ack_sent))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " canceling old ACK\n");
+ GMC_cancel (prev_fc->ack_msg);
+ /* GMC_cancel triggers ack_sent(), which clears fc->ack_msg */
+ }
+ else
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " same ACK already in queue\n");
+ return;
+ }
+ }
+
prev_fc->last_ack_sent = ack;
/* Build ACK message and send on connection */
msg.ack = htonl (ack);
msg.cid = c->id;
- send_prebuilt_message_connection (&msg.header, c, NULL, !fwd);
+ prev_fc->ack_msg = GMC_send_prebuilt_message (&msg.header, c, !fwd,
+ &ack_sent, prev_fc);
}
/**
- * Sends a CONNECTION ACK message in reponse to a received CONNECTION_CREATE
- * directed to us.
+ * Callback called when a queued message is sent.
*
- * @param connection Connection to confirm.
- * @param fwd Is this a fwd ACK? (First is bck (SYNACK), second is fwd (ACK))
+ * Calculates the average time and connection packet tracking.
+ *
+ * @param cls Closure (ConnectionQueue Handle).
+ * @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 void
-send_connection_ack (struct MeshConnection *connection, int fwd)
+message_sent (void *cls,
+ struct MeshConnection *c, uint16_t type,
+ int fwd, size_t size,
+ struct GNUNET_TIME_Relative wait)
{
- struct MeshTunnel3 *t;
+ struct MeshConnectionPerformance *p;
+ struct MeshFlowControl *fc;
+ struct MeshConnectionQueue *q = cls;
+ double usecsperbyte;
- t = connection->t;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Send connection ack\n");
- GMP_queue_add (NULL,
- GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
- sizeof (struct GNUNET_MESH_ConnectionACK),
- connection, NULL, fwd,
- &message_sent, sizeof (struct GNUNET_MESH_ConnectionACK));
- if (MESH_TUNNEL_NEW == t->state)
- GMT_change_state (t, MESH_TUNNEL_WAITING);
- if (MESH_CONNECTION_READY != connection->state)
- GMC_change_state (connection, MESH_CONNECTION_SENT);
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "! sent %s %s\n",
+ fwd ? "FWD" : "BCK",
+ GNUNET_MESH_DEBUG_M2S (type));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! C_P- %p %u\n", c, c->pending_messages);
+ if (NULL != q)
+ {
+ if (NULL != q->cont)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! calling cont\n");
+ q->cont (q->cont_cls, c, q, type, fwd, size);
+ }
+ GNUNET_free (q);
+ }
+ c->pending_messages--;
+ if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! destroying connection!\n");
+ GMC_destroy (c);
+ return;
+ }
+ /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
+ switch (type)
+ {
+ case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
+ fc->last_pid_sent++;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! Q_N- %p %u\n", fc, fc->queue_n);
+ fc->queue_n--;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "! accounting pid %u\n",
+ fc->last_pid_sent);
+ GMC_send_ack (c, fwd, GNUNET_NO);
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_POLL:
+ fc->poll_msg = NULL;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ fc->ack_msg = NULL;
+ break;
+
+ default:
+ break;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! message sent!\n");
+
+ if (NULL == c->perf)
+ return; /* Only endpoints are interested in timing. */
+
+ 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;
}
/**
- * Sends a CREATE CONNECTION message for a path to a peer.
- * Changes the connection and tunnel states if necessary.
+ * Get the previous hop in a connection
*
- * @param connection Connection to create.
+ * @param c Connection.
+ *
+ * @return Previous peer in the connection.
+ */
+static struct MeshPeer *
+get_prev_hop (const struct MeshConnection *c)
+{
+ GNUNET_PEER_Id id;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Get prev hop, own pos %u\n", c->own_pos);
+ if (0 == c->own_pos || c->path->length < 2)
+ id = c->path->peers[0];
+ else
+ id = c->path->peers[c->own_pos - 1];
+
+ return GMP_get_short (id);
+}
+
+
+/**
+ * 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);
+}
+
+
+/**
+ * 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.
+ */
+static 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;
+}
+
+
+/**
+ * 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 Should we send it FWD? (root->dest)
+ * (First (~SYNACK) goes BCK, second (~ACK) goes FWD)
*/
static void
-send_connection_create (struct MeshConnection *connection)
+send_connection_ack (struct MeshConnection *connection, int fwd)
{
struct MeshTunnel3 *t;
t = connection->t;
- 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)
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Send connection %s ACK\n",
+ !fwd ? "FWD" : "BCK");
+ GMP_queue_add (get_hop (connection, fwd), NULL,
+ GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
+ sizeof (struct GNUNET_MESH_ConnectionACK),
+ connection, fwd, &message_sent, NULL);
+ connection->pending_messages++;
+ 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);
}
+/**
+ * Send a notification that a connection is broken.
+ *
+ * @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_change_state (struct MeshConnection* c,
- enum MeshConnectionState state)
+send_broken (struct MeshConnection *c,
+ const struct GNUNET_PeerIdentity *id1,
+ const struct GNUNET_PeerIdentity *id2,
+ int fwd)
{
- 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;
+ struct GNUNET_MESH_ConnectionBroken msg;
+
+ 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, fwd, NULL, NULL);
}
*
* @param c Connection to keep alive..
* @param fwd Is this a FWD keepalive? (owner -> dest).
+ *
+ * FIXME use only one type, register in GMC_send_prebuilt_message()
*/
static void
connection_keepalive (struct MeshConnection *c, int fwd)
GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE;
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "sending %s keepalive for connection %s[%d]\n",
- fwd ? "FWD" : "BCK",
- peer2s (c->t->peer),
- c->id);
+ "sending %s keepalive for connection %s]\n",
+ fwd ? "FWD" : "BCK", GMC_2s (c));
msg = (struct GNUNET_MESH_ConnectionKeepAlive *) cbuf;
msg->header.size = htons (size);
msg->header.type = htons (type);
msg->cid = c->id;
- send_prebuilt_message_connection (&msg->header, c, NULL, fwd);
+ GMC_send_prebuilt_message (&msg->header, c, fwd, NULL, NULL);
}
* 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.
+ * @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)
- send_connection_create (c);
+ GMC_send_create (c);
else
send_connection_ack (c, GNUNET_NO);
}
static void
connection_maintain (struct MeshConnection *c, int fwd)
{
- if (MESH_TUNNEL_SEARCHING == c->t->state)
+ if (MESH_TUNNEL3_SEARCHING == GMT_get_state (c->t))
{
/* TODO DHT GET with RO_BART */
return;
{
case MESH_CONNECTION_NEW:
GNUNET_break (0);
+ /* fall-through */
case MESH_CONNECTION_SENT:
connection_recreate (c, fwd);
break;
connection_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct MeshConnection *c = cls;
+ struct GNUNET_TIME_Relative delay;
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,
+ delay = c->state == MESH_CONNECTION_READY ?
+ refresh_connection_time : create_connection_time;
+ c->fwd_maintenance_task = GNUNET_SCHEDULER_add_delayed (delay,
&connection_fwd_keepalive,
c);
}
connection_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct MeshConnection *c = cls;
+ struct GNUNET_TIME_Relative delay;
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,
+ delay = c->state == MESH_CONNECTION_READY ?
+ refresh_connection_time : create_connection_time;
+ c->bck_maintenance_task = GNUNET_SCHEDULER_add_delayed (delay,
&connection_bck_keepalive,
c);
}
-
-/**
- * Get the first transmittable message for a connection.
- *
- * @param c Connection.
- * @param fwd Is this FWD?
- *
- * @return First transmittable message.
- */
-static struct MeshPeerQueue *
-connection_get_first_message (struct MeshConnection *c, int fwd)
-{
- struct MeshPeerQueue *q;
- struct MeshPeer *p;
-
- p = connection_get_hop (c, fwd);
-
- for (q = p->queue_head; NULL != q; q = q->next)
- {
- if (q->c != c)
- continue;
- if (queue_is_sendable (q))
- return q;
- }
-
- return NULL;
-}
-
-
/**
* @brief Re-initiate traffic on this connection if necessary.
*
connection_unlock_queue (struct MeshConnection *c, int fwd)
{
struct MeshPeer *peer;
- struct MeshPeerQueue *q;
- size_t size;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"connection_unlock_queue %s on %s\n",
- fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
+ fwd ? "FWD" : "BCK", GMC_2s (c));
if (GMC_is_terminal (c, fwd))
{
return;
}
- peer = connection_get_hop (c, fwd);
-
- if (NULL != peer->core_transmit)
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG, " already unlocked!\n");
- return; /* Already unlocked */
- }
-
- q = connection_get_first_message (c, fwd);
- if (NULL == q)
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG, " queue empty!\n");
- return; /* Nothing to transmit */
- }
-
- 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);
+ 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.
+ * If the connection is scheduled for destruction and no more messages are left,
+ * the connection will be destroyed by the continuation call.
+ *
+ * @param c Connection which to cancel. Might be destroyed during this call.
* @param fwd Cancel fwd traffic?
*/
static void
connection_cancel_queues (struct MeshConnection *c, int fwd)
{
- struct MeshPeerQueue *q;
- struct MeshPeerQueue *next;
struct MeshFlowControl *fc;
struct MeshPeer *peer;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " *** Cancel %s queues for connection %s\n",
+ fwd ? "FWD" : "BCK", GMC_2s (c));
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)
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
{
- next = q->next;
- if (q->c == c)
- {
- 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;
- }
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** Cancel POLL in ccq for fc %p\n", fc);
}
+ peer = get_hop (c, fwd);
+ GMP_queue_cancel (peer, c);
}
* @param tc TaskContext.
*/
static void
-connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
+ * Callback called when a queued POLL message is sent.
+ *
+ * @param cls Closure (FC).
+ * @param c Connection this message was on.
+ * @param q Queue handler this call invalidates.
+ * @param type Type of message sent.
+ * @param fwd Was this a FWD going message?
+ * @param size Size of the message.
+ */
+static void
+poll_sent (void *cls,
+ struct MeshConnection *c,
+ struct MeshConnectionQueue *q,
+ uint16_t type, int fwd, size_t size)
{
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))
+ if (2 == c->destroy)
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL canceled on shutdown\n");
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);
- send_prebuilt_message_connection (&msg.header, c, NULL, fc == &c->fwd_fc);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " *** POLL sent for , scheduling new one!\n");
+ fc->poll_msg = NULL;
fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
&connection_poll, fc);
-}
-
-
-
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " task %u\n", fc->poll_task);
-/**
- * Get the previous hop in a connection
- *
- * @param c Connection.
- *
- * @return Previous peer in the connection.
- */
-static struct MeshPeer *
-connection_get_prev_hop (struct MeshConnection *c)
-{
- GNUNET_PEER_Id id;
-
- if (0 == c->own_pos || c->path->length < 2)
- id = c->path->peers[0];
- else
- id = c->path->peers[c->own_pos - 1];
-
- return peer_get_short (id);
}
-
/**
- * Get the next hop in a connection
- *
- * @param c Connection.
+ * Function called if a connection has been stalled for a while,
+ * possibly due to a missed ACK. Poll the neighbor about its ACK status.
*
- * @return Next peer in the connection.
+ * @param cls Closure (poll ctx).
+ * @param tc TaskContext.
*/
-static struct MeshPeer *
-connection_get_next_hop (struct MeshConnection *c)
+static void
+connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
- 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];
+ struct MeshFlowControl *fc = cls;
+ struct GNUNET_MESH_Poll msg;
+ struct MeshConnection *c;
- return peer_get_short (id);
-}
+ 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 [%s]\n", GMC_2s (c));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** %s\n",
+ fc == &c->fwd_fc ? "FWD" : "BCK");
-/**
- * Get the hop in a connection.
- *
- * @param c Connection.
- * @param fwd Next hop?
- *
- * @return Next peer in the connection.
- */
-static struct MeshPeer *
-connection_get_hop (struct MeshConnection *c, int fwd)
-{
- if (fwd)
- return connection_get_next_hop (c);
- return connection_get_prev_hop (c);
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
+ msg.header.size = htons (sizeof (msg));
+ msg.pid = htonl (fc->last_pid_sent);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** last pid sent: %u!\n", fc->last_pid_sent);
+ fc->poll_msg = GMC_send_prebuilt_message (&msg.header, c, fc == &c->fwd_fc,
+ &poll_sent, fc);
}
-
-
/**
* Timeout function due to lack of keepalive/traffic from the owner.
* Destroys connection if called.
return;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Connection %s[%X] FWD timed out. Destroying.\n",
- peer2s (c->t->peer),
+ GMT_2s (c->t),
c->id);
if (GMC_is_origin (c, GNUNET_YES)) /* If local, leave. */
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Connection %s[%X] FWD timed out. Destroying.\n",
- peer2s (c->t->peer),
- c->id);
+ GMT_2s (c->t), c->id);
if (GMC_is_origin (c, GNUNET_NO)) /* If local, leave. */
return;
if (GNUNET_SCHEDULER_NO_TASK != *ti)
GNUNET_SCHEDULER_cancel (*ti);
- if (GMC_is_origin (c, fwd)) /* Endpoint */
+ if (GMC_is_origin (c, fwd)) /* Startpoint */
{
f = fwd ? &connection_fwd_keepalive : &connection_bck_keepalive;
*ti = GNUNET_SCHEDULER_add_delayed (refresh_connection_time, f, c);
}
- else /* Relay */
+ else /* Relay, endpoint. */
{
struct GNUNET_TIME_Relative delay;
* Add the connection to the list of both neighbors.
*
* @param c Connection.
+ *
+ * @return #GNUNET_OK if everything went fine
+ * #GNUNET_SYSERR if the was an error and @c c is malformed.
*/
-static void
+static int
register_neighbors (struct MeshConnection *c)
{
- struct MeshPeer *peer;
+ struct MeshPeer *next_peer;
+ struct MeshPeer *prev_peer;
- peer = connection_get_next_hop (c);
- if (GNUNET_NO == GMP_is_neighbor (peer))
- {
- GMC_destroy (c);
- return NULL;
- }
- GMP_add_connection (peer, c);
- peer = connection_get_prev_hop (c);
- if (GNUNET_NO == GMP_is_neighbor (peer))
- {
- GMC_destroy (c);
- return NULL;
- }
- GMP_add_connection (peer, c);
+ next_peer = get_next_hop (c);
+ prev_peer = get_prev_hop (c);
+
+ if (GNUNET_NO == GMP_is_neighbor (next_peer)
+ || GNUNET_NO == GMP_is_neighbor (prev_peer))
+ return GNUNET_SYSERR;
+
+ GMP_add_connection (next_peer, c);
+ GMP_add_connection (prev_peer, c);
+
+ return GNUNET_OK;
}
{
struct MeshPeer *peer;
- peer = connection_get_next_hop (c);
+ peer = get_next_hop (c);
GMP_remove_connection (peer, c);
- peer = connection_get_prev_hop (c);
+ peer = get_prev_hop (c);
GMP_remove_connection (peer, c);
}
/**
- * Callback called when a queued message is sent.
+ * Bind the connection to the peer and the tunnel to that peer.
*
- * Calculates the average time
+ * If the peer has no tunnel, create one. Update tunnel and connection
+ * data structres to reflect new status.
*
- * @param cls Closure.
- * @param c Connection this message was on.
- * @param wait Time spent waiting for core (only the time for THIS message)
+ * @param c Connection.
+ * @param peer Peer.
*/
-static void
-message_sent (void *cls,
- struct MeshConnection *c,
- struct GNUNET_TIME_Relative wait)
+static void
+add_to_peer (struct MeshConnection *c, struct MeshPeer *peer)
{
- struct MeshConnectionPerformance *p;
- size_t size = (size_t) cls;
- double usecsperbyte;
-
- if (NULL == c->perf)
- return; /* Only endpoints are interested in this. */
-
- 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;
+ GMP_add_tunnel (peer);
+ c->t = GMP_get_tunnel (peer);
+ GMT_add_connection (c->t, c);
}
-
/******************************************************************************/
/******************************** API ***********************************/
/******************************************************************************/
c = connection_get (cid);
if (NULL == c)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " Creating connection\n");
- c = connection_new (cid);
- if (NULL == c)
- return GNUNET_OK;
- connection_reset_timeout (c, GNUNET_YES);
-
/* Create path */
LOG (GNUNET_ERROR_TYPE_DEBUG, " Creating path...\n");
path = path_new (size);
LOG (GNUNET_ERROR_TYPE_DEBUG, " ... adding %s\n",
GNUNET_i2s (&id[i]));
path->peers[i] = GNUNET_PEER_intern (&id[i]);
- if (path->peers[i] == my_short_id)
+ 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);
- GMC_destroy (c);
return GNUNET_OK;
}
LOG (GNUNET_ERROR_TYPE_DEBUG, " Own position: %u\n", own_pos);
- path_add_to_peers (path, GNUNET_NO);
- c->path = path_duplicate (path);
- c->own_pos = 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)
+ {
+ path_destroy (path);
+ return GNUNET_OK;
+ }
+ connection_reset_timeout (c, GNUNET_YES);
}
else
{
- path = NULL;
+ path = path_duplicate (c->path);
}
if (MESH_CONNECTION_NEW == c->state)
connection_change_state (c, MESH_CONNECTION_SENT);
/* Remember peers */
- dest_peer = peer_get (&id[size - 1]);
- orig_peer = peer_get (&id[0]);
+ dest_peer = GMP_get (&id[size - 1]);
+ orig_peer = GMP_get (&id[0]);
/* Is it a connection to us? */
if (c->own_pos == size - 1)
{
LOG (GNUNET_ERROR_TYPE_DEBUG, " It's for us!\n");
- peer_add_path_to_origin (orig_peer, path, GNUNET_YES);
+ GMP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_YES);
- if (NULL == orig_peer->tunnel)
- {
- orig_peer->tunnel = tunnel_new ();
- orig_peer->tunnel->peer = orig_peer;
- }
- GMT_add_connection (orig_peer->tunnel, c);
- if (MESH_TUNNEL_NEW == GMT_get_state (c->t))
- GMT_change_state (c->t, MESH_TUNNEL_WAITING);
+ 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);
+ c->bck_maintenance_task =
+ GNUNET_SCHEDULER_add_delayed (create_connection_time,
+ &connection_bck_keepalive, c);
}
else
{
/* It's for somebody else! Retransmit. */
LOG (GNUNET_ERROR_TYPE_DEBUG, " Retransmitting.\n");
- peer_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
- peer_add_path_to_origin (orig_peer, path, GNUNET_NO);
- send_prebuilt_message_connection (message, c, NULL, GNUNET_YES);
+ GMP_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
+ GMP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_NO);
+ GMC_send_prebuilt_message (message, c, GNUNET_YES, NULL, NULL);
}
+ path_destroy (path);
return GNUNET_OK;
}
struct MeshConnection *c;
struct MeshPeerPath *p;
struct MeshPeer *pi;
+ enum MeshConnectionState oldstate;
int fwd;
LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
return GNUNET_OK;
}
-
- LOG (GNUNET_ERROR_TYPE_DEBUG, " via peer %s\n",
- GNUNET_i2s (peer));
- pi = peer_get (peer);
- if (connection_get_next_hop (c) == pi)
+ oldstate = c->state;
+ 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)
+ if (MESH_CONNECTION_SENT == oldstate)
connection_change_state (c, MESH_CONNECTION_ACK);
}
- else if (connection_get_prev_hop (c) == pi)
+ else if (get_prev_hop (c) == pi)
{
LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK\n");
fwd = GNUNET_YES;
GNUNET_break_op (0);
return GNUNET_OK;
}
+
connection_reset_timeout (c, fwd);
/* Add path to peers? */
p = c->path;
if (NULL != p)
{
- path_add_to_peers (p, GNUNET_YES);
+ GMP_add_path_to_all (p, GNUNET_YES);
}
else
{
}
/* Message for us as creator? */
- if (connection_is_origin (c, GNUNET_YES))
+ if (GMC_is_origin (c, GNUNET_YES))
{
+ if (GNUNET_NO != fwd)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
LOG (GNUNET_ERROR_TYPE_DEBUG, " Connection (SYN)ACK for us!\n");
+
+ /* If just created, cancel the short timeout and start a long one */
+ if (MESH_CONNECTION_SENT == oldstate)
+ connection_reset_timeout (c, GNUNET_YES);
+
+ /* Change connection and tunnel state */
connection_change_state (c, MESH_CONNECTION_READY);
- if (MESH_TUNNEL_READY != c->t->state)
- tunnel_change_state (c->t, MESH_TUNNEL_READY);
+ if (MESH_TUNNEL3_WAITING == GMT_get_state (c->t))
+ GMT_change_state (c->t, MESH_TUNNEL3_READY);
+
+ /* Send ACK (~TCP ACK)*/
send_connection_ack (c, GNUNET_YES);
- tunnel_send_queued_data (c->t, GNUNET_YES);
- if (3 <= tunnel_count_connections (c->t) && NULL != c->t->peer->dhtget)
- {
- GMP_stop_search (c->t->peer);
- }
return GNUNET_OK;
}
/* Message for us as destination? */
if (GMC_is_terminal (c, GNUNET_YES))
{
+ if (GNUNET_YES != fwd)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
LOG (GNUNET_ERROR_TYPE_DEBUG, " Connection ACK for us!\n");
- GMC_change_state (c, MESH_CONNECTION_READY);
- GMT_change_state (c->t, MESH_TUNNEL_READY);
- GMT_send_queued_data (c->t, GNUNET_NO);
+
+ /* If just created, cancel the short timeout and start a long one */
+ if (MESH_CONNECTION_ACK == oldstate)
+ connection_reset_timeout (c, GNUNET_NO);
+
+ /* Change tunnel state */
+ if (MESH_TUNNEL3_WAITING == GMT_get_state (c->t))
+ GMT_change_state (c->t, MESH_TUNNEL3_READY);
+
return GNUNET_OK;
}
LOG (GNUNET_ERROR_TYPE_DEBUG, " not for us, retransmitting...\n");
- GMC_send_prebuilt_message (message, c, NULL, fwd);
+ GMC_send_prebuilt_message (message, c, fwd, NULL, NULL);
return GNUNET_OK;
}
* Core handler for notifications of broken paths
*
* @param cls Closure (unused).
- * @param peer Peer identity of sending neighbor.
+ * @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 *peer,
- const struct GNUNET_MessageHeader *message)
+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 (peer));
+ "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));
GNUNET_break_op (0);
return GNUNET_OK;
}
- tunnel_notify_connection_broken (c->t, GNUNET_PEER_search (&msg->peer1),
- GNUNET_PEER_search (&msg->peer2));
+
+ 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, fwd, NULL, NULL);
+ c->destroy = GNUNET_YES;
+ }
+
return GNUNET_OK;
}
{
struct GNUNET_MESH_ConnectionDestroy *msg;
struct MeshConnection *c;
- GNUNET_PEER_Id id;
int fwd;
msg = (struct GNUNET_MESH_ConnectionDestroy *) message;
*/
GNUNET_STATISTICS_update (stats, "# control on unknown tunnel",
1, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " connection unknown: already destroyed?\n");
return GNUNET_OK;
}
- id = GNUNET_PEER_search (peer);
- if (id == GMP_get_short_id (connection_get_prev_hop (c)))
- fwd = GNUNET_YES;
- else if (id == GMP_get_short_id (connection_get_next_hop (c)))
- fwd = GNUNET_NO;
- else
+ fwd = is_fwd (c, peer);
+ if (GNUNET_SYSERR == fwd)
{
GNUNET_break_op (0);
return GNUNET_OK;
}
- GMC_send_prebuilt_message (message, c, NULL, fwd);
+ GMC_send_prebuilt_message (message, c, fwd, NULL, NULL);
c->destroy = GNUNET_YES;
+ c->state = MESH_CONNECTION_DESTROYED;
return GNUNET_OK;
}
* Generic handler for mesh network encrypted traffic.
*
* @param peer Peer identity this notification is about.
- * @param message Encrypted message.
- * @param fwd Is this FWD traffic? GNUNET_YES : GNUNET_NO;
+ * @param msg 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,
- int fwd)
+ const struct GNUNET_MESH_Encrypted *msg)
{
struct MeshConnection *c;
- struct MeshTunnel3 *t;
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);
}
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));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "got a %s message (#%u) from %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), ntohl (msg->pid), 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");
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "WARNING connection %s unknown\n",
+ GNUNET_h2s (&msg->cid));
return GNUNET_OK;
}
- t = c->t;
- fc = fwd ? &c->bck_fc : &c->fwd_fc;
/* Check if origin is as expected */
- neighbor = connection_get_hop (c, !fwd);
- if (GNUNET_PEER_search (peer) != GMP_get_short_id (neighbor))
+ 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))
+ {
+ fwd = GNUNET_NO;
+ }
+ else
+ {
+ /* Unexpected peer sending traffic on a connection. */
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
+ }
+
+ /* Check PID */
+ fc = fwd ? &c->bck_fc : &c->fwd_fc;
+ pid = ntohl (msg->pid);
+ if (GMC_is_pid_bigger (pid, fc->last_ack_sent))
+ {
+ 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 (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 || MESH_CONNECTION_ACK == 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))
+ {
+ /* 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_break (0);
+ return GNUNET_OK;
+ }
+ fc->last_pid_recv = pid;
+ GMT_handle_encrypted (c->t, msg);
+ GMC_send_ack (c, fwd, GNUNET_NO);
+ 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, fwd, GNUNET_NO);
+ return GNUNET_OK;
+ }
+
+ GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
+ GMC_send_prebuilt_message (&msg->header, c, fwd, NULL, NULL);
+
+ return GNUNET_OK;
+}
+
+/**
+ * Generic handler for mesh network encrypted traffic.
+ *
+ * @param peer Peer identity this notification is about.
+ * @param msg Encrypted message.
+ *
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_mesh_kx (const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MESH_KX *msg)
+{
+ struct MeshConnection *c;
+ struct MeshPeer *neighbor;
+ GNUNET_PEER_Id peer_id;
+ size_t size;
+ uint16_t type;
+ 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 PID */
- pid = ntohl (msg->pid);
- if (GMC_is_pid_bigger (pid, fc->last_ack_sent))
+ /* 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))
{
- 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;
+ fwd = GNUNET_YES;
}
- if (GNUNET_NO == GMC_is_pid_bigger (pid, fc->last_pid_recv))
+ else
{
- 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;
+ neighbor = get_next_hop (c);
+ if (peer_id == GMP_get_short_id (neighbor))
+ {
+ fwd = GNUNET_NO;
+ }
+ else
+ {
+ /* Unexpected peer sending traffic on a connection. */
+ GNUNET_break_op (0);
+ return GNUNET_OK;
+ }
}
- if (MESH_CONNECTION_SENT == c->state)
+
+ /* Count as connection confirmation. */
+ if (MESH_CONNECTION_SENT == c->state || MESH_CONNECTION_ACK == c->state)
connection_change_state (c, MESH_CONNECTION_READY);
connection_reset_timeout (c, fwd);
- fc->last_pid_recv = pid;
+ if (NULL != c->t)
+ {
+ if (MESH_TUNNEL3_WAITING == GMT_get_state (c->t))
+ GMT_change_state (c->t, MESH_TUNNEL3_READY);
+ }
/* Is this message for us? */
if (GMC_is_terminal (c, fwd))
{
- size_t dsize = size - sizeof (struct GNUNET_MESH_Encrypted);
- char cbuf[dsize];
- struct GNUNET_MessageHeader *msgh;
- unsigned int off;
-
- /* TODO signature verification */
LOG (GNUNET_ERROR_TYPE_DEBUG, " message for us!\n");
GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
-
- fc->last_pid_recv = pid;
- tunnel_decrypt (t, cbuf, &msg[1], dsize, msg->iv, fwd);
- off = 0;
- while (off < dsize)
+ if (NULL == c->t)
{
- msgh = (struct GNUNET_MessageHeader *) &cbuf[off];
- handle_decrypted (t, msgh, fwd);
- off += ntohs (msgh->size);
+ GNUNET_break (0);
+ return GNUNET_OK;
}
- send_ack (c, NULL, fwd);
+ GMT_handle_kx (c->t, &msg[1].header);
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");
- send_ack (c, NULL, fwd);
- return GNUNET_OK;
- }
GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
-
- send_prebuilt_message_connection (&msg->header, c, NULL, fwd);
+ GMC_send_prebuilt_message (&msg->header, c, fwd, NULL, NULL);
return GNUNET_OK;
}
/**
- * Core handler for mesh network traffic going orig->dest.
+ * Core handler for encrypted mesh network traffic (channel mgmt, data).
*
* @param cls Closure (unused).
* @param message Message received.
* GNUNET_SYSERR to close it (signal serious error)
*/
int
-GMC_handle_fwd (void *cls, const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+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,
- GNUNET_YES);
+ (struct GNUNET_MESH_Encrypted *)message);
}
+
/**
- * Core handler for mesh network traffic going dest->orig.
+ * Core handler for key exchange traffic (ephemeral key, ping, pong).
*
* @param cls Closure (unused).
* @param message Message received.
* GNUNET_SYSERR to close it (signal serious error)
*/
int
-GMC_handle_bck (void *cls, const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+GMC_handle_kx (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
{
- return handle_mesh_encrypted (peer,
- (struct GNUNET_MESH_Encrypted *)message,
- GNUNET_NO);
+ return handle_mesh_kx (peer,
+ (struct GNUNET_MESH_KX *) message);
}
/* Is this a forward or backward ACK? */
id = GNUNET_PEER_search (peer);
- if (GMP_get_short_id (connection_get_next_hop (c)) == id)
+ 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 (connection_get_prev_hop (c)) == id)
+ else if (GMP_get_short_id (get_prev_hop (c)) == id)
{
LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
fc = &c->bck_fc;
int fwd;
LOG (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Got a POLL packet from %s!\n",
- GNUNET_i2s (peer));
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got a POLL packet from %s!\n",
+ GNUNET_i2s (peer));
msg = (struct GNUNET_MESH_Poll *) message;
* this way of discerining FWD/BCK should not be a problem.
*/
id = GNUNET_PEER_search (peer);
- if (GMP_get_short_id (connection_get_next_hop (c)) == id)
+ if (GMP_get_short_id (get_next_hop (c)) == id)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " FWD FC\n");
fc = &c->fwd_fc;
}
- else if (GMP_get_short_id (connection_get_prev_hop (c)) == id)
+ else if (GMP_get_short_id (get_prev_hop (c)) == id)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " BCK FC\n");
fc = &c->bck_fc;
}
else
}
pid = ntohl (msg->pid);
- LOG (GNUNET_ERROR_TYPE_DEBUG, " PID %u, OLD %u\n",
- pid, fc->last_pid_recv);
+ 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;
- send_ack (c, NULL, fwd);
+ fwd = fc == &c->bck_fc;
+ GMC_send_ack (c, fwd, GNUNET_YES);
return GNUNET_OK;
}
GNUNET_YES : GNUNET_NO;
/* Check if origin is as expected */
- neighbor = connection_get_hop (c, fwd);
+ neighbor = get_hop (c, fwd);
if (GNUNET_PEER_search (peer) != GMP_get_short_id (neighbor))
{
GNUNET_break_op (0);
return GNUNET_OK;
GNUNET_STATISTICS_update (stats, "# keepalives forwarded", 1, GNUNET_NO);
- GMC_send_prebuilt_message (message, c, NULL, fwd);
+ GMC_send_prebuilt_message (message, c, fwd, NULL, NULL);
return GNUNET_OK;
}
* 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)
+ * @param fwd Is this a fwd ACK? (will go dest->root).
+ * @param force Send the ACK even if suboptimal (e.g. requested by POLL).
*/
void
-GMC_send_ack (struct MeshConnection *c, struct MeshChannel *ch, int fwd)
+GMC_send_ack (struct MeshConnection *c, int fwd, int force)
{
unsigned int buffer;
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))
+ "GMC send %s ACK on %s\n",
+ fwd ? "FWD" : "BCK", GMC_2s (c));
+
+ if (NULL == c)
+ {
+ GNUNET_break (0);
+ return;
+ }
+
+ if (GNUNET_NO != c->destroy)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " being destroyed, why bother...\n");
+ return;
+ }
+
+ /* Get available buffer space */
+ if (GMC_is_terminal (c, fwd))
{
- 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);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " getting from all channels\n");
+ buffer = GMT_get_channels_buffer (c->t);
}
else
{
buffer = GMC_get_buffer (c, fwd);
}
LOG (GNUNET_ERROR_TYPE_DEBUG, " buffer available: %u\n", buffer);
+ if (0 == buffer && GNUNET_NO == force)
+ return;
- if ( (NULL != ch && GMCH_is_origin (ch, fwd)) ||
- (NULL != c && GMC_is_origin (c, fwd)) )
+ /* Send available buffer space */
+ if (GMC_is_origin (c, fwd))
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on channel...\n");
- if (0 < buffer)
- {
- GNUNET_assert (NULL != ch);
- LOG (GNUNET_ERROR_TYPE_DEBUG, " really sending!\n");
- send_local_ack (ch, fwd);
- }
- }
- else if (NULL == c)
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on all connections\n");
- GNUNET_assert (NULL != ch);
- channel_send_connections_ack (ch, buffer, fwd);
+ GNUNET_assert (NULL != c->t);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on channels...\n");
+ GMT_unchoke_channels (c->t);
}
else
{
LOG (GNUNET_ERROR_TYPE_DEBUG, " sending on connection\n");
- connection_send_ack (c, buffer, fwd);
+ send_ack (c, buffer, fwd, force);
}
}
void
GMC_init (const struct GNUNET_CONFIGURATION_Handle *c)
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "init\n");
if (GNUNET_OK !=
GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_MSGS_QUEUE",
&max_msgs_queue))
{
- LOG_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
"MESH", "MAX_MSGS_QUEUE", "MISSING");
GNUNET_SCHEDULER_shutdown ();
return;
GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_CONNECTIONS",
&max_connections))
{
- LOG_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
"MESH", "MAX_CONNECTIONS", "MISSING");
GNUNET_SCHEDULER_shutdown ();
return;
GNUNET_CONFIGURATION_get_value_time (c, "MESH", "REFRESH_CONNECTION_TIME",
&refresh_connection_time))
{
- LOG_config_invalid (GNUNET_ERROR_TYPE_ERROR,
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
"MESH", "REFRESH_CONNECTION_TIME", "MISSING");
GNUNET_SCHEDULER_shutdown ();
return;
}
+ create_connection_time = GNUNET_TIME_UNIT_SECONDS;
connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
}
+
+/**
+ * Destroy each connection on shutdown.
+ *
+ * @param cls Closure (unused).
+ * @param key Current key code (CID, unused).
+ * @param value Value in the hash map (connection)
+ *
+ * @return #GNUNET_YES, because we should continue to iterate,
+ */
+static int
+shutdown_iterator (void *cls,
+ const struct GNUNET_HashCode *key,
+ void *value)
+{
+ struct MeshConnection *c = value;
+
+ GMC_destroy (c);
+ return GNUNET_YES;
+}
+
+
/**
* Shut down the connections subsystem.
*/
void
GMC_shutdown (void)
{
+ GNUNET_CONTAINER_multihashmap_iterate (connections, &shutdown_iterator, NULL);
+ GNUNET_CONTAINER_multihashmap_destroy (connections);
+ connections = NULL;
}
if (0 == own_pos)
{
c->fwd_maintenance_task =
- GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
+ GNUNET_SCHEDULER_add_delayed (create_connection_time,
&connection_fwd_keepalive, c);
}
- register_neighbors (c);
+ if (GNUNET_OK != register_neighbors (c))
+ {
+ GMC_destroy (c);
+ return NULL;
+ }
+
return c;
}
void
GMC_destroy (struct MeshConnection *c)
{
- struct MeshPeer *peer;
-
if (NULL == c)
return;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s\n",
- GNUNET_h2s (&c->id));
+ if (2 == c->destroy) /* cancel queues -> GMP_queue_cancel -> q_destroy -> */
+ return; /* -> message_sent -> GMC_destroy. Don't loop. */
+ c->destroy = 2;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s\n", GMC_2s (c));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " fc's f: %p, b: %p\n",
+ &c->fwd_fc, &c->bck_fc);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " fc tasks f: %u, b: %u\n",
+ c->fwd_fc.poll_task, c->bck_fc.poll_task);
/* Cancel all traffic */
connection_cancel_queues (c, GNUNET_YES);
connection_cancel_queues (c, GNUNET_NO);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " fc tasks f: %u, b: %u\n",
+ c->fwd_fc.poll_task, c->bck_fc.poll_task);
+
/* 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);
+ if (GNUNET_SCHEDULER_NO_TASK != c->fwd_fc.poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (c->fwd_fc.poll_task);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL FWD canceled\n");
+ }
+ if (GNUNET_SCHEDULER_NO_TASK != c->bck_fc.poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (c->bck_fc.poll_task);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL BCK canceled\n");
+ }
+ if (NULL != c->fwd_fc.poll_msg)
+ {
+ GMC_cancel (c->fwd_fc.poll_msg);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL msg FWD canceled\n");
+ }
+ if (NULL != c->bck_fc.poll_msg)
+ {
+ GMC_cancel (c->bck_fc.poll_msg);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL msg BCK canceled\n");
+ }
/* Unregister from neighbors */
unregister_neighbors (c);
/* Delete */
GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
- GMT_remove_connection (c->t, c);
+ if (NULL != c->t)
+ GMT_remove_connection (c->t, c);
+
+ if (GNUNET_NO == GMC_is_origin (c, GNUNET_YES))
+ path_destroy (c->path);
+
+ GNUNET_break (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (connections, &c->id, c));
+
GNUNET_free (c);
}
}
+/**
+ * Get the connection path.
+ *
+ * @param c Connection to get the path from.
+ *
+ * @return path used by the connection.
+ */
+const struct MeshPeerPath *
+GMC_get_path (const struct MeshConnection *c)
+{
+ return c->path;
+}
+
+
/**
* Get the connection state.
*
return (fc->queue_max - fc->queue_n);
}
+/**
+ * Get how many messages have we allowed to send to us from a direction.
+ *
+ * @param c Connection.
+ * @param fwd Are we asking about traffic from FWD (BCK messages)?
+ *
+ * @return last_ack_sent - last_pid_recv
+ */
+unsigned int
+GMC_get_allowed (struct MeshConnection *c, int fwd)
+{
+ struct MeshFlowControl *fc;
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ if (GMC_is_pid_bigger(fc->last_pid_recv, fc->last_ack_sent))
+ {
+ return 0;
+ }
+ return (fc->last_ack_sent - fc->last_pid_recv);
+}
+
/**
* Get messages queued in a connection.
*
}
+/**
+ * Allow the connection to advertise a buffer of the given size.
+ *
+ * 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).
+ */
+void
+GMC_allow (struct MeshConnection *c, unsigned int buffer, int fwd)
+{
+ send_ack (c, buffer, fwd, GNUNET_NO);
+}
+
+
/**
* 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 MeshPeer *peer)
{
- struct GNUNET_MESH_ConnectionBroken msg;
int fwd;
- fwd = peer == connection_get_prev_hop (c);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " notify broken on %s due to %s disconnect\n",
+ GMC_2s (c), GMP_2s (peer));
- connection_cancel_queues (c, !fwd);
- if (GMC_is_terminal (c, fwd))
+ fwd = peer == get_prev_hop (c);
+
+ if (GNUNET_YES == GMC_is_terminal (c, fwd))
{
/* Local shutdown, no one to notify about this. */
GMC_destroy (c);
return;
}
+ if (GNUNET_NO == c->destroy)
+ send_broken (c, &my_full_id, GMP_get_id (peer), fwd);
- 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 = *GMP_get_id (peer);
- GMC_send_prebuilt_message (&msg.header, c, NULL, 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;
+ c->state = MESH_CONNECTION_DESTROYED;
+
+ /**
+ * Cancel all queues, if no message is left, connection will be destroyed.
+ */
+ connection_cancel_queues (c, !fwd);
return;
}
* @param c Connection.
* @param fwd Is this about fwd traffic?
*
- * @return GNUNET_YES if origin, GNUNET_NO if relay/terminal.
+ * @return #GNUNET_YES if origin, #GNUNET_NO if relay/terminal.
*/
int
GMC_is_origin (struct MeshConnection *c, int fwd)
* @param fwd Is this about fwd traffic?
* Note that the ROOT is the terminal for BCK traffic!
*
- * @return GNUNET_YES if terminal, GNUNET_NO if relay/origin.
+ * @return #GNUNET_YES if terminal, #GNUNET_NO if relay/origin.
*/
int
GMC_is_terminal (struct MeshConnection *c, int fwd)
}
+/**
+ * See if we are allowed to send by the next hop in the given direction.
+ *
+ * @param c Connection.
+ * @param fwd Is this about fwd traffic?
+ *
+ * @return #GNUNET_YES in case it's OK to send.
+ */
+int
+GMC_is_sendable (struct MeshConnection *c, int fwd)
+{
+ struct MeshFlowControl *fc;
+
+ 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;
+}
+
/**
* 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?
+ * @param cont Continuation called once message is sent. Can be NULL.
+ * @param cont_cls Closure for @c cont.
+ *
+ * @return Handle to cancel the message before it's sent.
+ * NULL on error or if @c cont is NULL.
+ * Invalid on @c cont call.
*/
-void
+struct MeshConnectionQueue *
GMC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
- struct MeshConnection *c,
- struct MeshChannel *ch,
- int fwd)
+ struct MeshConnection *c, int fwd,
+ GMC_sent cont, void *cont_cls)
{
+ struct MeshFlowControl *fc;
+ struct MeshConnectionQueue *q;
void *data;
size_t size;
uint16_t type;
+ int droppable;
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));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Send %s (%u bytes) on connection %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), size, GMC_2s (c));
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ droppable = GNUNET_YES;
switch (type)
{
struct GNUNET_MESH_Encrypted *emsg;
+ struct GNUNET_MESH_KX *kmsg;
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:
+ case GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED:
emsg = (struct GNUNET_MESH_Encrypted *) data;
ttl = ntohl (emsg->ttl);
if (0 == ttl)
{
GNUNET_break_op (0);
- return;
+ GNUNET_free (data);
+ return NULL;
}
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));
+ emsg->pid = htonl (fc->next_pid++);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " Q_N+ %p %u\n", fc, fc->queue_n);
+ fc->queue_n++;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "pid %u\n", ntohl (emsg->pid));
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid sent %u\n", fc->last_pid_sent);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " ack recv %u\n", fc->last_ack_recv);
+ if (GMC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
+ {
+ GMC_start_poll (c, fwd);
+ }
+ break;
+
+ case GNUNET_MESSAGE_TYPE_MESH_KX:
+ kmsg = (struct GNUNET_MESH_KX *) data;
+ kmsg->cid = c->id;
break;
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;
- case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
+ case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
dmsg->cid = c->id;
dmsg->reserved = 0;
GNUNET_break (0);
}
- GMP_queue_add (data, type, size, c, ch, fwd, &message_sent, (void *) size);
+ if (fc->queue_n > fc->queue_max && droppable)
+ {
+ 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);
+ if (GNUNET_MESSAGE_TYPE_MESH_ENCRYPTED == type)
+ {
+ fc->queue_n--;
+ fc->next_pid--;
+ }
+ GNUNET_free (data);
+ return NULL; /* Drop this message */
+ }
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " C_P+ %p %u\n", c, c->pending_messages);
+ c->pending_messages++;
+
+ if (NULL == cont)
+ {
+ (void) GMP_queue_add (get_hop (c, fwd), data, type, size, c, fwd,
+ &message_sent, NULL);
+ return NULL;
+ }
+
+ q = GNUNET_new (struct MeshConnectionQueue);
+ q->q = GMP_queue_add (get_hop (c, fwd), data, type, size, c, fwd,
+ &message_sent, q);
+ if (NULL == q->q)
+ {
+ GNUNET_break (0);
+ GNUNET_free (data);
+ GNUNET_free (q);
+ return NULL;
+ }
+ q->cont = cont;
+ q->cont_cls = cont_cls;
+ return q;
+}
+
+
+/**
+ * Cancel a previously sent message while it's in the queue.
+ *
+ * ONLY can be called before the continuation given to the send function
+ * is called. Once the continuation is called, the message is no longer in the
+ * queue.
+ *
+ * @param q Handle to the queue.
+ */
+void
+GMC_cancel (struct MeshConnectionQueue *q)
+{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "! GMC cancel message\n");
+
+ /* queue destroy calls message_sent, which calls q->cont and frees q */
+ GMP_queue_destroy (q->q, GNUNET_YES);
+}
+
+
+/**
+ * Sends a CREATE CONNECTION message for a path to a peer.
+ * Changes the connection and tunnel states if necessary.
+ *
+ * @param connection Connection to create.
+ */
+void
+GMC_send_create (struct MeshConnection *connection)
+{
+ enum MeshTunnel3State state;
+ size_t size;
+
+ 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, GNUNET_YES, &message_sent, NULL);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " C_P+ %p %u (create)\n",
+ connection, connection->pending_messages);
+ connection->pending_messages++;
+ 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);
}
return;
msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY);;
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY);;
msg.cid = c->id;
LOG (GNUNET_ERROR_TYPE_DEBUG,
" sending connection destroy for connection %s\n",
- GNUNET_h2s (&c->id));
+ GMC_2s (c));
if (GNUNET_NO == GMC_is_terminal (c, GNUNET_YES))
- GMC_send_prebuilt_message (&msg.header, c, NULL, GNUNET_YES);
+ GMC_send_prebuilt_message (&msg.header, c, GNUNET_YES, NULL, NULL);
if (GNUNET_NO == GMC_is_terminal (c, GNUNET_NO))
- GMC_send_prebuilt_message (&msg.header, c, NULL, GNUNET_NO);
+ GMC_send_prebuilt_message (&msg.header, c, GNUNET_NO, NULL, NULL);
c->destroy = GNUNET_YES;
-}
\ No newline at end of file
+ c->state = MESH_CONNECTION_DESTROYED;
+}
+
+
+/**
+ * @brief Start a polling timer for the connection.
+ *
+ * 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.
+ *
+ * @param c Connection.
+ * @param fwd Should we poll in the FWD direction?
+ */
+void
+GMC_start_poll (struct MeshConnection *c, int fwd)
+{
+ struct MeshFlowControl *fc;
+
+ fc = fwd ? &c->fwd_fc : &c->bck_fc;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL %s requested\n",
+ fwd ? "FWD" : "BCK");
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task || NULL != fc->poll_msg)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** not needed (%u, %p)\n",
+ fc->poll_task, fc->poll_msg);
+ return;
+ }
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " *** POLL started on request\n");
+ fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
+ &connection_poll,
+ fc);
+}
+
+
+/**
+ * @brief Stop polling a connection for ACKs.
+ *
+ * Once we have enough ACKs for future traffic, polls are no longer necessary.
+ *
+ * @param c Connection.
+ * @param fwd Should we stop the poll in the FWD direction?
+ */
+void
+GMC_stop_poll (struct MeshConnection *c, int 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;
+ }
+}
+
+/**
+ * Get a (static) string for a connection.
+ *
+ * @param c Connection.
+ */
+const char *
+GMC_2s (struct MeshConnection *c)
+{
+ if (NULL != c->t)
+ {
+ static char buf[128];
+
+ sprintf (buf, "%s (->%s)", GNUNET_h2s (&c->id), GMT_2s (c->t));
+ return buf;
+ }
+ return GNUNET_h2s (&c->id);
+}
projects / oweals / gnunet.git / blobdiff