* @author Bartlomiej Polot
*
* FIXME in progress:
- * - keep queues until receiving ACK
+ * - when sending in-order buffered data, wait for client ACKs
*
* TODO:
* - relay corking down to core
#define MESH_MAX_POLL_TIME GNUNET_TIME_relative_multiply (\
GNUNET_TIME_UNIT_MINUTES,\
10)
-#define MESH_RETRANSMIT_TIME GNUNET_TIME_relative_multiply (\
- GNUNET_TIME_UNIT_SECONDS,\
- 5)
+#define MESH_RETRANSMIT_TIME GNUNET_TIME_UNIT_SECONDS
+#define MESH_RETRANSMIT_MARGIN 4
#if MESH_DEBUG_CONNECTION
#define DEBUG_CONN(...) GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
/******************************************************************************/
/** FWD declaration */
-struct MeshPeerInfo;
+struct MeshPeer;
struct MeshClient;
/**
* Peer this transmission is directed to.
*/
- struct MeshPeerInfo *peer;
+ struct MeshPeer *peer;
/**
* Tunnel this message belongs to.
/**
* Struct containing all information regarding a given peer
*/
-struct MeshPeerInfo
+struct MeshPeer
{
/**
* ID of the peer
};
+/**
+ * Data needed for reliable tunnel endpoint retransmission management.
+ */
+struct MeshTunnelReliability;
+
+
/**
* Info needed to retry a message in case it gets lost.
*/
/**
* ID of the message (ACK needed to free)
*/
- uint32_t id;
+ uint32_t mid;
/**
- * When was this message issued (to calculate ACK delay) FIXME update with traffic
+ * When was this message issued (to calculate ACK delay)
*/
struct GNUNET_TIME_Absolute timestamp;
};
-/**
- * Data needed for reliable tunnel endpoint retransmission management.
- */
struct MeshTunnelReliability
{
/**
/**
* DLL of messages sent and not yet ACK'd.
*/
- struct MeshReliableMessage *head_sent;
- struct MeshReliableMessage *tail_sent;
+ struct MeshReliableMessage *head_sent;
+ struct MeshReliableMessage *tail_sent;
+
+ /**
+ * Messages pending
+ */
+ unsigned int n_sent;
+
+ /**
+ * Next MID to use.
+ */
+ uint32_t mid_sent;
/**
* DLL of messages received out of order.
*/
- struct MeshReliableMessage *head_recv;
- struct MeshReliableMessage *tail_recv;
+ struct MeshReliableMessage *head_recv;
+ struct MeshReliableMessage *tail_recv;
+
+ /**
+ * Next MID expected.
+ */
+ uint32_t mid_recv;
/**
* Task to resend/poll in case no ACK is received.
*/
- GNUNET_SCHEDULER_TaskIdentifier retry_task;
+ GNUNET_SCHEDULER_TaskIdentifier retry_task;
/**
* Counter for exponential backoff.
*/
- struct GNUNET_TIME_Relative retry_timer;
+ struct GNUNET_TIME_Relative retry_timer;
/**
- * How long does it usually take to get an ACK. FIXME update with traffic
+ * How long does it usually take to get an ACK.
*/
- struct GNUNET_TIME_Relative expected_delay;
+ struct GNUNET_TIME_Relative expected_delay;
};
* Task to keep the used paths alive at the owner,
* time tunnel out on all the other peers.
*/
- GNUNET_SCHEDULER_TaskIdentifier maintenance_task;
+ GNUNET_SCHEDULER_TaskIdentifier fwd_maintenance_task;
+
+ /**
+ * Task to keep the used paths alive at the destination,
+ * time tunnel out on all the other peers.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier bck_maintenance_task;
/**
* Path being used for the tunnel.
/**
* Struct containing information about a client of the service
- *
+ *
* TODO: add a list of 'waiting' ports
*/
struct MeshClient
}
#endif
-/* FIXME */
-unsigned int debug_fwd_ack;
-unsigned int debug_bck_ack;
-
#endif
/******************************************************************************/
static unsigned long long max_tunnels;
/**
- * How many messages *in total* are we willing to queue, divided by number of
+ * How many messages *in total* are we willing to queue, divided by number of
* tunnels to get tunnel queue size.
*/
static unsigned long long max_msgs_queue;
static struct GNUNET_CONTAINER_MultiHashMap32 *incoming_tunnels;
/**
- * Peers known, indexed by PeerIdentity (MeshPeerInfo).
+ * Peers known, indexed by PeerIdentity (MeshPeer).
*/
-static struct GNUNET_CONTAINER_MultiHashMap *peers;
+static struct GNUNET_CONTAINER_MultiPeerMap *peers;
/*
* Handle to communicate with transport
/**
* Own private key.
*/
-static struct GNUNET_CRYPTO_EccPrivateKey *my_private_key;
-
-/**
- * Own public key.
- */
-static struct GNUNET_CRYPTO_EccPublicKeyBinaryEncoded my_public_key;
+static struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
/**
* Tunnel ID for the next created tunnel (global tunnel number).
/**
- * Retrieve the MeshPeerInfo stucture associated with the peer, create one
+ * Retrieve the MeshPeer stucture associated with the peer, create one
* and insert it in the appropriate structures if the peer is not known yet.
*
* @param peer Full identity of the peer.
*
* @return Existing or newly created peer info.
*/
-static struct MeshPeerInfo *
+static struct MeshPeer *
peer_get (const struct GNUNET_PeerIdentity *peer);
/**
- * Retrieve the MeshPeerInfo stucture associated with the peer, create one
+ * Retrieve the MeshPeer stucture associated with the peer, create one
* and insert it in the appropriate structures if the peer is not known yet.
*
* @param peer Short identity of the peer.
*
* @return Existing or newly created peer info.
*/
-static struct MeshPeerInfo *
+static struct MeshPeer *
peer_get_short (const GNUNET_PEER_Id peer);
tunnel_get (const struct GNUNET_PeerIdentity *oid, MESH_TunnelNumber tid);
+/**
+ * Change the tunnel state.
+ *
+ * @param t Tunnel whose ttate to change.
+ * @param state New state.
+ */
+static void
+tunnel_change_state (struct MeshTunnel *t, enum MeshTunnelState state);
+
+
/**
* Notify a tunnel that a connection has broken that affects at least
* some of its peers.
GNUNET_PEER_Id p2);
+/**
+ * Send FWD keepalive packets for a tunnel.
+ *
+ * @param cls Closure (tunnel for which to send the keepalive).
+ * @param tc Notification context.
+ */
+static void
+tunnel_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+/**
+ * Send BCK keepalive packets for a tunnel.
+ *
+ * @param cls Closure (tunnel for which to send the keepalive).
+ * @param tc Notification context.
+ */
+static void
+tunnel_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
/**
* @brief Use the given path for the tunnel.
* Update the next and prev hops (and RCs).
* (Re)start the path refresh in case the tunnel is locally owned.
- *
+ *
* @param t Tunnel to update.
* @param p Path to use.
*/
/**
* Tunnel is empty: destroy it.
- *
+ *
* Notifies all participants (peers, cleints) about the destruction.
- *
- * @param t Tunnel to destroy.
+ *
+ * @param t Tunnel to destroy.
*/
static void
tunnel_destroy_empty (struct MeshTunnel *t);
+/**
+ * Destroy the tunnel.
+ *
+ * This function does not generate any warning traffic to clients or peers.
+ *
+ * Tasks:
+ * Remove the tunnel from peer_info's and clients' hashmaps.
+ * Cancel messages belonging to this tunnel queued to neighbors.
+ * Free any allocated resources linked to the tunnel.
+ *
+ * @param t the tunnel to destroy
+ *
+ * @return GNUNET_OK on success
+ */
+static int
+tunnel_destroy (struct MeshTunnel *t);
+
/**
* @brief Queue and pass message to core when possible.
- *
+ *
* If type is payload (UNICAST, TO_ORIGIN, MULTICAST) checks for queue status
* and accounts for it. In case the queue is full, the message is dropped and
* a break issued.
- *
- * Otherwise, message is treated as internal and allowed to go regardless of
+ *
+ * Otherwise, message is treated as internal and allowed to go regardless of
* queue status.
*
* @param cls Closure (@c type dependant). It will be used by queue_send to
*/
static void
queue_add (void *cls, uint16_t type, size_t size,
- struct MeshPeerInfo *dst, struct MeshTunnel *t);
+ struct MeshPeer *dst, struct MeshTunnel *t);
/**
* NULL when there are no transmittable messages.
*/
struct MeshPeerQueue *
-queue_get_next (const struct MeshPeerInfo *peer);
+queue_get_next (const struct MeshPeer *peer);
/**
queue_send (void *cls, size_t size, void *buf);
+/**
+ * Dummy function to separate declarations from definitions in function list.
+ */
+void
+__mesh_divider______________________________________________________________();
+
+
+static const char *
+GNUNET_MESH_DEBUG_S2S (enum MeshTunnelState s)
+{
+ static char buf[128];
+
+ switch (s)
+ {
+ /**
+ * Uninitialized status, should never appear in operation.
+ */
+ case MESH_TUNNEL_NEW: return "MESH_TUNNEL_NEW";
+
+ /**
+ * Path to the peer not known yet
+ */
+ case MESH_TUNNEL_SEARCHING: return "MESH_TUNNEL_SEARCHING";
+
+ /**
+ * Request sent, not yet answered.
+ */
+ case MESH_TUNNEL_WAITING: return "MESH_TUNNEL_WAITING";
+
+ /**
+ * Peer connected and ready to accept data
+ */
+ case MESH_TUNNEL_READY: return "MESH_TUNNEL_READY";
+
+ /**
+ * Peer connected previosly but not responding
+ */
+ case MESH_TUNNEL_RECONNECTING: return "MESH_TUNNEL_RECONNECTING";
+
+ default:
+ sprintf (buf, "%u (UNKNOWN STATE)", s);
+ return buf;
+ }
+}
+
+
/******************************************************************************/
/************************ PERIODIC FUNCTIONS ****************************/
/******************************************************************************/
announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct PBlock block;
+ struct GNUNET_HashCode phash;
if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
{
* - Adapt X to churn
*/
DEBUG_DHT ("DHT_put for ID %s started.\n", GNUNET_i2s (&my_full_id));
-
+ GNUNET_CRYPTO_hash (&my_full_id, sizeof (my_full_id), &phash);
block.id = my_full_id;
GNUNET_DHT_put (dht_handle, /* DHT handle */
- &my_full_id.hashPubKey, /* Key to use */
+ &phash,
dht_replication_level, /* Replication level */
GNUNET_DHT_RO_RECORD_ROUTE | GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, /* DHT options */
GNUNET_BLOCK_TYPE_MESH_PEER, /* Block type */
static struct MeshClient *
client_get (struct GNUNET_SERVER_Client *client)
{
- struct MeshClient *c;
-
- c = clients_head;
- while (NULL != c)
- {
- if (c->handle == client)
- return c;
- c = c->next;
- }
- return NULL;
+ return GNUNET_SERVER_client_get_user_context (client, struct MeshClient);
}
static void
client_delete_tunnel (struct MeshClient *c, struct MeshTunnel *t)
{
+ int res;
+
if (c == t->owner)
{
- GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap32_remove (c->own_tunnels,
- t->local_tid,
- t));
+ res = GNUNET_CONTAINER_multihashmap32_remove (c->own_tunnels,
+ t->local_tid, t);
+ if (GNUNET_YES != res)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client_delete_tunnel owner KO\n");
}
if (c == t->client)
{
- GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap32_remove (c->incoming_tunnels,
- t->local_tid_dest,
- t));
+ res = GNUNET_CONTAINER_multihashmap32_remove (c->incoming_tunnels,
+ t->local_tid_dest, t);
+ if (GNUNET_YES != res)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client_delete_tunnel client KO\n");
}
}
+
/**
* Notify the appropriate client that a new incoming tunnel was created.
*
* @param t Tunnel that was created.
*/
static void
-send_client_tunnel_create (struct MeshTunnel *t)
+send_local_tunnel_create (struct MeshTunnel *t)
{
struct GNUNET_MESH_TunnelMessage msg;
msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_CREATE);
msg.tunnel_id = htonl (t->local_tid_dest);
msg.port = htonl (t->port);
- msg.options = 0;
- msg.options |= GNUNET_YES == t->reliable ? GNUNET_MESH_OPTION_RELIABLE : 0;
- msg.options |= GNUNET_YES == t->nobuffer ? GNUNET_MESH_OPTION_NOBUFFER : 0;
- msg.options = htonl (msg.options);
+ msg.opt = 0;
+ msg.opt |= GNUNET_YES == t->reliable ? GNUNET_MESH_OPTION_RELIABLE : 0;
+ msg.opt |= GNUNET_YES == t->nobuffer ? GNUNET_MESH_OPTION_NOBUFFER : 0;
+ msg.opt = htonl (msg.opt);
GNUNET_PEER_resolve (t->id.oid, &msg.peer);
GNUNET_SERVER_notification_context_unicast (nc, t->client->handle,
&msg.header, GNUNET_NO);
/**
- * Notify dest client that the incoming tunnel is no longer valid.
+ * Notify a client that the incoming tunnel is no longer valid.
*
- * @param c Client to notify..
* @param t Tunnel that is destroyed.
+ * @param fwd Forward notification (owner->dest)?
*/
static void
-send_client_tunnel_destroy (struct MeshClient *c, struct MeshTunnel *t)
+send_local_tunnel_destroy (struct MeshTunnel *t, int fwd)
{
struct GNUNET_MESH_TunnelMessage msg;
+ struct MeshClient *c;
+ c = fwd ? t->client : t->owner;
if (NULL == c)
{
GNUNET_break (0);
return;
}
- if (c != t->client && c != t->owner)
- {
- GNUNET_break (0);
- return;
- }
msg.header.size = htons (sizeof (msg));
msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_DESTROY);
- msg.tunnel_id = htonl (t->local_tid_dest);
+ msg.tunnel_id = htonl (fwd ? t->local_tid_dest : t->local_tid);
msg.port = htonl (0);
- memset(&msg.peer, 0, sizeof (msg.peer));
+ memset (&msg.peer, 0, sizeof (msg.peer));
+ msg.opt = htonl (0);
GNUNET_SERVER_notification_context_unicast (nc, c->handle,
&msg.header, GNUNET_NO);
}
/**
- * Iterator over all the peers to remove the oldest not-used entry.
- *
- * @param cls Closure (unsued).
- * @param key ID of the peer.
- * @param value Peer_Info of the peer.
- *
- * FIXME implement
- */
-static int
-peer_info_timeout (void *cls,
- const struct GNUNET_HashCode *key,
- void *value)
-{
- return GNUNET_YES;
-}
-
-/**
- * Retrieve the MeshPeerInfo stucture associated with the peer, create one
- * and insert it in the appropriate structures if the peer is not known yet.
- *
- * @param peer Full identity of the peer.
- *
- * @return Existing or newly created peer info.
- */
-static struct MeshPeerInfo *
-peer_get (const struct GNUNET_PeerIdentity *peer)
-{
- struct MeshPeerInfo *peer_info;
-
- peer_info = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
- if (NULL == peer_info)
- {
- peer_info =
- (struct MeshPeerInfo *) GNUNET_malloc (sizeof (struct MeshPeerInfo));
- if (GNUNET_CONTAINER_multihashmap_size (peers) > max_peers)
- {
- GNUNET_CONTAINER_multihashmap_iterate (peers,
- &peer_info_timeout,
- NULL);
- }
- GNUNET_CONTAINER_multihashmap_put (peers, &peer->hashPubKey, peer_info,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
- peer_info->id = GNUNET_PEER_intern (peer);
- }
- peer_info->last_contact = GNUNET_TIME_absolute_get();
-
- return peer_info;
-}
-
-
-/**
- * Retrieve the MeshPeerInfo stucture associated with the peer, create one
- * and insert it in the appropriate structures if the peer is not known yet.
- *
- * @param peer Short identity of the peer.
- *
- * @return Existing or newly created peer info.
- */
-static struct MeshPeerInfo *
-peer_get_short (const GNUNET_PEER_Id peer)
-{
- struct GNUNET_PeerIdentity id;
-
- GNUNET_PEER_resolve (peer, &id);
- return peer_get (&id);
-}
-
-
-/**
- * Choose the best path towards a peer considering the tunnel properties.
- *
- * @param peer The destination peer.
- * @param t The tunnel the path is for.
+ * Build a local ACK message and send it to a local client.
*
- * @return Best current known path towards the peer, if any.
+ * @param t Tunnel on which to send the ACK.
+ * @param c Client to whom send the ACK.
+ * @param is_fwd Set to GNUNET_YES for FWD ACK (dest->owner)
*/
-static struct MeshPeerPath *
-peer_get_best_path (const struct MeshPeerInfo *peer, const struct MeshTunnel *t)
+static void
+send_local_ack (struct MeshTunnel *t,
+ struct MeshClient *c,
+ int is_fwd)
{
- struct MeshPeerPath *best_p;
- struct MeshPeerPath *p;
- unsigned int best_cost;
- unsigned int cost;
+ struct GNUNET_MESH_LocalAck msg;
- best_p = p = peer->path_head;
- best_cost = cost = p->length;
- while (NULL != p)
- {
- if ((cost = p->length) < best_cost)
- {
- best_cost = cost;
- best_p = p;
- }
- p = p->next;
- }
- return best_p;
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
+ msg.tunnel_id = htonl (is_fwd ? t->local_tid : t->local_tid_dest);
+ GNUNET_SERVER_notification_context_unicast (nc,
+ c->handle,
+ &msg.header,
+ GNUNET_NO);
}
-/**
- * Core callback to write a pre-constructed data packet to core buffer
- *
- * @param cls Closure (MeshTransmissionDescriptor with data in "data" member).
- * @param size Number of bytes available in buf.
- * @param buf Where the to write the message.
- *
- * @return number of bytes written to buf
- */
-static size_t
-send_core_data_raw (void *cls, size_t size, void *buf)
-{
- struct GNUNET_MessageHeader *msg = cls;
- size_t total_size;
-
- GNUNET_assert (NULL != msg);
- total_size = ntohs (msg->size);
-
- if (total_size > size)
- {
- GNUNET_break (0);
- return 0;
- }
- memcpy (buf, msg, total_size);
- GNUNET_free (cls);
- return total_size;
-}
/**
struct MeshTunnel *t)
{
struct GNUNET_PeerIdentity id;
- struct MeshPeerInfo *neighbor;
+ struct MeshPeer *neighbor;
struct MeshPeerPath *p;
void *data;
size_t size;
}
}
#endif
- GNUNET_break (0); // FIXME sometimes fails (testing disconnect?)
+ GNUNET_break (0);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
" no direct connection to %s\n",
GNUNET_i2s (&id));
* @param t Tunnel for which the path is created.
*/
static void
-send_create_path (struct MeshTunnel *t)
+send_path_create (struct MeshTunnel *t)
{
- struct MeshPeerInfo *neighbor;
+ struct MeshPeer *neighbor;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send create path\n");
neighbor = peer_get_short (t->next_hop);
(t->path->length * sizeof (struct GNUNET_PeerIdentity)),
neighbor,
t);
- t->state = MESH_TUNNEL_WAITING;
+ tunnel_change_state (t, MESH_TUNNEL_WAITING);
}
* @param t Tunnel which to confirm.
*/
static void
-send_path_ack (struct MeshTunnel *t)
+send_path_ack (struct MeshTunnel *t)
{
- struct MeshPeerInfo *peer;
-
- peer = peer_get_short (t->prev_hop);
+ struct MeshPeer *neighbor;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send path ack\n");
+ neighbor = peer_get_short (t->prev_hop);
queue_add (t,
GNUNET_MESSAGE_TYPE_MESH_PATH_ACK,
sizeof (struct GNUNET_MESH_PathACK),
- peer,
+ neighbor,
t);
+ tunnel_change_state (t, MESH_TUNNEL_WAITING);
}
/**
- * Try to establish a new connection to this peer in the given tunnel.
- * If the peer doesn't have any path to it yet, try to get one.
- * If the peer already has some path, send a CREATE PATH towards it.
+ * Build an ACK message and queue it to send to the given peer.
*
- * @param peer PeerInfo of the peer.
- * @param t Tunnel for which to create the path, if possible.
+ * @param t Tunnel on which to send the ACK.
+ * @param peer Peer to whom send the ACK.
+ * @param ack Value of the ACK.
*/
static void
-peer_connect (struct MeshPeerInfo *peer, struct MeshTunnel *t)
+send_ack (struct MeshTunnel *t, GNUNET_PEER_Id peer, uint32_t ack)
{
- struct MeshPeerPath *p;
+ struct GNUNET_MESH_ACK msg;
- if (NULL != peer->path_head)
- {
- p = peer_get_best_path (peer, t);
- tunnel_use_path (t, p);
- send_create_path (t);
- }
- else if (NULL == peer->dhtget)
- {
- struct GNUNET_PeerIdentity id;
+ GNUNET_PEER_resolve (t->id.oid, &msg.oid);
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
+ msg.pid = htonl (ack);
+ msg.tid = htonl (t->id.tid);
- GNUNET_PEER_resolve (peer->id, &id);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " Starting DHT GET for peer %s\n", GNUNET_i2s (&id));
- peer->dhtget = GNUNET_DHT_get_start (dht_handle, /* handle */
- GNUNET_BLOCK_TYPE_MESH_PEER, /* type */
- &id.hashPubKey, /* key to search */
- dht_replication_level, /* replication level */
- GNUNET_DHT_RO_RECORD_ROUTE |
- GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
- NULL, /* xquery */
+ send_prebuilt_message (&msg.header, peer, t);
+}
+
+
+/**
+ * Core callback to write a pre-constructed data packet to core buffer
+ *
+ * @param cls Closure (MeshTransmissionDescriptor with data in "data" member).
+ * @param size Number of bytes available in buf.
+ * @param buf Where the to write the message.
+ *
+ * @return number of bytes written to buf
+ */
+static size_t
+send_core_data_raw (void *cls, size_t size, void *buf)
+{
+ struct GNUNET_MessageHeader *msg = cls;
+ size_t total_size;
+
+ GNUNET_assert (NULL != msg);
+ total_size = ntohs (msg->size);
+
+ if (total_size > size)
+ {
+ GNUNET_break (0);
+ return 0;
+ }
+ memcpy (buf, msg, total_size);
+ GNUNET_free (cls);
+ return total_size;
+}
+
+
+/**
+ * Function to send a create path packet to a peer.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+send_core_path_create (void *cls, size_t size, void *buf)
+{
+ struct MeshTunnel *t = cls;
+ struct GNUNET_MESH_CreateTunnel *msg;
+ struct GNUNET_PeerIdentity *peer_ptr;
+ struct MeshPeerPath *p = t->path;
+ size_t size_needed;
+ uint32_t opt;
+ int i;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CREATE PATH sending...\n");
+ size_needed =
+ sizeof (struct GNUNET_MESH_CreateTunnel) +
+ p->length * sizeof (struct GNUNET_PeerIdentity);
+
+ if (size < size_needed || NULL == buf)
+ {
+ GNUNET_break (0);
+ return 0;
+ }
+ msg = (struct GNUNET_MESH_CreateTunnel *) buf;
+ msg->header.size = htons (size_needed);
+ msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE);
+ msg->tid = ntohl (t->id.tid);
+
+ opt = 0;
+ if (GNUNET_YES == t->nobuffer)
+ opt |= GNUNET_MESH_OPTION_NOBUFFER;
+ if (GNUNET_YES == t->reliable)
+ opt |= GNUNET_MESH_OPTION_RELIABLE;
+ msg->opt = htonl (opt);
+ msg->port = htonl (t->port);
+
+ peer_ptr = (struct GNUNET_PeerIdentity *) &msg[1];
+ for (i = 0; i < p->length; i++)
+ {
+ GNUNET_PEER_resolve (p->peers[i], peer_ptr++);
+ }
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "CREATE PATH (%u bytes long) sent!\n", size_needed);
+ return size_needed;
+}
+
+
+/**
+ * Creates a path ack message in buf and frees all unused resources.
+ *
+ * @param cls closure (MeshTransmissionDescriptor)
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+send_core_path_ack (void *cls, size_t size, void *buf)
+{
+ struct MeshTunnel *t = cls;
+ struct GNUNET_MESH_PathACK *msg = buf;
+
+ GNUNET_assert (NULL != t);
+ if (sizeof (struct GNUNET_MESH_PathACK) > size)
+ {
+ GNUNET_break (0);
+ return 0;
+ }
+ t->prev_fc.last_ack_sent = t->nobuffer ? 0 : t->queue_max - 1;
+ msg->header.size = htons (sizeof (struct GNUNET_MESH_PathACK));
+ msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_ACK);
+ GNUNET_PEER_resolve (t->id.oid, &msg->oid);
+ msg->tid = htonl (t->id.tid);
+ msg->peer_id = my_full_id;
+ msg->ack = htonl (t->prev_fc.last_ack_sent);
+
+ /* TODO add signature */
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "PATH ACK sent!\n");
+ return sizeof (struct GNUNET_MESH_PathACK);
+}
+
+
+/**
+ * Iterator over all the peers to remove the oldest not-used entry.
+ *
+ * @param cls Closure (unsued).
+ * @param key ID of the peer.
+ * @param value Peer_Info of the peer.
+ *
+ * FIXME implement
+ */
+static int
+peer_timeout (void *cls,
+ const struct GNUNET_PeerIdentity *key,
+ void *value)
+{
+ return GNUNET_YES;
+}
+
+
+/**
+ * Retrieve the MeshPeer stucture associated with the peer, create one
+ * and insert it in the appropriate structures if the peer is not known yet.
+ *
+ * @param peer Full identity of the peer.
+ *
+ * @return Existing or newly created peer info.
+ */
+static struct MeshPeer *
+peer_get (const struct GNUNET_PeerIdentity *peer_id)
+{
+ struct MeshPeer *peer;
+
+ peer = GNUNET_CONTAINER_multipeermap_get (peers, peer_id);
+ if (NULL == peer)
+ {
+ peer = GNUNET_new (struct MeshPeer);
+ if (GNUNET_CONTAINER_multipeermap_size (peers) > max_peers)
+ {
+ GNUNET_CONTAINER_multipeermap_iterate (peers,
+ &peer_timeout,
+ NULL);
+ }
+ GNUNET_CONTAINER_multipeermap_put (peers, peer_id, peer,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ peer->id = GNUNET_PEER_intern (peer_id);
+ }
+ peer->last_contact = GNUNET_TIME_absolute_get();
+
+ return peer;
+}
+
+
+/**
+ * Retrieve the MeshPeer stucture associated with the peer, create one
+ * and insert it in the appropriate structures if the peer is not known yet.
+ *
+ * @param peer Short identity of the peer.
+ *
+ * @return Existing or newly created peer info.
+ */
+static struct MeshPeer *
+peer_get_short (const GNUNET_PEER_Id peer)
+{
+ return peer_get (GNUNET_PEER_resolve2 (peer));
+}
+
+
+/**
+ * Select which PID to POLL for, to compensate for lost messages.
+ *
+ * @param pi Peer we want to poll.
+ * @param t Tunnel about which we want to poll.
+ *
+ * @return PID to use, either last sent or first_in_queue - 1
+ */
+static uint32_t
+peer_get_first_payload_pid (struct MeshPeer *p, struct MeshTunnel *t)
+{
+ struct MeshPeerQueue *q;
+ uint16_t type;
+
+ type = p->id == t->next_hop ? GNUNET_MESSAGE_TYPE_MESH_UNICAST :
+ GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN;
+
+ for (q = p->queue_head; NULL != q; q = q->next)
+ {
+ if (q->type == type && q->tunnel == t)
+ {
+ struct GNUNET_MESH_Data *msg = q->cls;
+
+ /* Pretend that the last one sent was the previous to this */
+ return ntohl (msg->pid) - 1;
+ }
+ }
+
+ /* No data in queue, use last sent */
+ {
+ struct MeshFlowControl *fc;
+
+ fc = p->id == t->next_hop ? &t->next_fc : &t->prev_fc;
+ return fc->last_pid_sent;
+ }
+}
+
+
+/**
+ * Choose the best path towards a peer considering the tunnel properties.
+ *
+ * @param peer The destination peer.
+ * @param t The tunnel the path is for.
+ *
+ * @return Best current known path towards the peer, if any.
+ */
+static struct MeshPeerPath *
+peer_get_best_path (const struct MeshPeer *peer, const struct MeshTunnel *t)
+{
+ struct MeshPeerPath *best_p;
+ struct MeshPeerPath *p;
+ unsigned int best_cost;
+ unsigned int cost;
+
+ best_p = p = peer->path_head;
+ best_cost = cost = p->length;
+ while (NULL != p)
+ {
+ if ((cost = p->length) < best_cost)
+ {
+ best_cost = cost;
+ best_p = p;
+ }
+ p = p->next;
+ }
+ return best_p;
+}
+
+
+/**
+ * Try to establish a new connection to this peer in the given tunnel.
+ * If the peer doesn't have any path to it yet, try to get one.
+ * If the peer already has some path, send a CREATE PATH towards it.
+ *
+ * @param peer PeerInfo of the peer.
+ * @param t Tunnel for which to create the path, if possible.
+ */
+static void
+peer_connect (struct MeshPeer *peer, struct MeshTunnel *t)
+{
+ struct MeshPeerPath *p;
+ struct GNUNET_HashCode phash;
+
+ if (NULL != peer->path_head)
+ {
+ p = peer_get_best_path (peer, t);
+ tunnel_use_path (t, p);
+ send_path_create (t);
+ }
+ else if (NULL == peer->dhtget)
+ {
+ struct GNUNET_PeerIdentity id;
+
+ GNUNET_PEER_resolve (peer->id, &id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Starting DHT GET for peer %s\n",
+ GNUNET_i2s (&id));
+ GNUNET_CRYPTO_hash (&id, sizeof (my_full_id), &phash);
+ peer->dhtget = GNUNET_DHT_get_start (dht_handle, /* handle */
+ GNUNET_BLOCK_TYPE_MESH_PEER, /* type */
+ &phash, /* key to search */
+ dht_replication_level, /* replication level */
+ GNUNET_DHT_RO_RECORD_ROUTE |
+ GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
+ NULL, /* xquery */
0, /* xquery bits */
&dht_get_id_handler, peer);
- t->state = MESH_TUNNEL_SEARCHING;
+ tunnel_change_state (t, MESH_TUNNEL_SEARCHING);
}
else
{
}
+/**
+ * @brief Re-initiate traffic to this peer 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 peer_id Short ID of peer to which initiate traffic.
+ */
+static void
+peer_unlock_queue (GNUNET_PEER_Id peer_id)
+{
+ struct MeshPeer *peer;
+ struct MeshPeerQueue *q;
+ size_t size;
+
+ peer = peer_get_short (peer_id);
+ if (NULL != peer->core_transmit)
+ return;
+
+ q = queue_get_next (peer);
+ if (NULL == q)
+ return;
+
+ 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);
+}
+
+
+/**
+ * Cancel all transmissions towards a neighbor that belong to a certain tunnel.
+ *
+ * @param neighbor Short ID of the neighbor to whom cancel the transmissions.
+ * @param t Tunnel which to cancel.
+ */
+static void
+peer_cancel_queues (GNUNET_PEER_Id neighbor, struct MeshTunnel *t)
+{
+ struct MeshPeer *peer;
+ struct MeshPeerQueue *q;
+ struct MeshPeerQueue *next;
+ struct MeshFlowControl *fc;
+
+ if (0 == neighbor)
+ return; /* Was local peer, 0'ed in tunnel_destroy_iterator */
+ peer = peer_get_short (neighbor);
+ for (q = peer->queue_head; NULL != q; q = next)
+ {
+ next = q->next;
+ if (q->tunnel == t)
+ {
+ if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == q->type ||
+ GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == q->type)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "peer_cancel_queues %s\n",
+ GNUNET_MESH_DEBUG_M2S (q->type));
+ }
+ queue_destroy (q, GNUNET_YES);
+ }
+ }
+ if (NULL == peer->queue_head && NULL != peer->core_transmit)
+ {
+ GNUNET_CORE_notify_transmit_ready_cancel (peer->core_transmit);
+ peer->core_transmit = NULL;
+ }
+ fc = neighbor == t->next_hop ? &t->next_fc : &t->prev_fc;
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+}
+
+
/**
* Destroy the peer_info and free any allocated resources linked to it
*
- * @param pi The peer_info to destroy.
+ * @param peer The peer_info to destroy.
*
* @return GNUNET_OK on success
*/
static int
-peer_info_destroy (struct MeshPeerInfo *pi)
+peer_destroy (struct MeshPeer *peer)
{
struct GNUNET_PeerIdentity id;
struct MeshPeerPath *p;
struct MeshPeerPath *nextp;
unsigned int i;
- GNUNET_PEER_resolve (pi->id, &id);
- GNUNET_PEER_change_rc (pi->id, -1);
+ GNUNET_PEER_resolve (peer->id, &id);
+ GNUNET_PEER_change_rc (peer->id, -1);
if (GNUNET_YES !=
- GNUNET_CONTAINER_multihashmap_remove (peers, &id.hashPubKey, pi))
+ GNUNET_CONTAINER_multipeermap_remove (peers, &id, peer))
{
GNUNET_break (0);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"removing peer %s, not in hashmap\n", GNUNET_i2s (&id));
}
- if (NULL != pi->dhtget)
+ if (NULL != peer->dhtget)
{
- GNUNET_DHT_get_stop (pi->dhtget);
+ GNUNET_DHT_get_stop (peer->dhtget);
}
- p = pi->path_head;
+ p = peer->path_head;
while (NULL != p)
{
nextp = p->next;
- GNUNET_CONTAINER_DLL_remove (pi->path_head, pi->path_tail, p);
+ GNUNET_CONTAINER_DLL_remove (peer->path_head, peer->path_tail, p);
path_destroy (p);
p = nextp;
}
- for (i = 0; i < pi->ntunnels; i++)
- tunnel_destroy_empty (pi->tunnels[i]);
- GNUNET_array_grow (pi->tunnels, pi->ntunnels, 0);
- GNUNET_free (pi);
+ for (i = 0; i < peer->ntunnels; i++)
+ tunnel_destroy_empty (peer->tunnels[i]);
+ GNUNET_array_grow (peer->tunnels, peer->ntunnels, 0);
+ GNUNET_free (peer);
return GNUNET_OK;
}
* TODO: optimize (see below)
*/
static void
-peer_remove_path (struct MeshPeerInfo *peer, GNUNET_PEER_Id p1,
+peer_remove_path (struct MeshPeer *peer, GNUNET_PEER_Id p1,
GNUNET_PEER_Id p2)
{
struct MeshPeerPath *p;
struct MeshPeerPath *next;
- struct MeshPeerInfo *peer_d;
+ struct MeshPeer *peer_d;
GNUNET_PEER_Id d;
unsigned int destroyed;
unsigned int i;
* @param trusted Do we trust that this path is real?
*/
void
-peer_info_add_path (struct MeshPeerInfo *peer_info, struct MeshPeerPath *path,
+peer_add_path (struct MeshPeer *peer_info, struct MeshPeerPath *path,
int trusted)
{
struct MeshPeerPath *aux;
* @param trusted Do we trust that this path is real?
*/
static void
-peer_info_add_path_to_origin (struct MeshPeerInfo *peer_info,
- struct MeshPeerPath *path, int trusted)
+peer_add_path_to_origin (struct MeshPeer *peer_info,
+ struct MeshPeerPath *path, int trusted)
{
path_invert (path);
- peer_info_add_path (peer_info, path, trusted);
+ peer_add_path (peer_info, path, trusted);
}
/**
* Add a tunnel to the list of tunnels a peer participates in.
* Update the tunnel's destination.
- *
+ *
* @param p Peer to add to.
* @param t Tunnel to add.
*/
static void
-peer_info_add_tunnel (struct MeshPeerInfo *p, struct MeshTunnel *t)
+peer_add_tunnel (struct MeshPeer *p, struct MeshTunnel *t)
{
if (0 != t->dest)
{
/**
* Remove a tunnel from the list of tunnels a peer participates in.
* Free the tunnel's destination.
- *
+ *
* @param p Peer to clean.
* @param t Tunnel to remove.
*/
static void
-peer_info_remove_tunnel (struct MeshPeerInfo *p, struct MeshTunnel *t)
+peer_remove_tunnel (struct MeshPeer *p, struct MeshTunnel *t)
{
unsigned int i;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** Polling!\n");
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
- msg.header.size = htons (sizeof (msg));
- msg.tid = htonl (t->id.tid);
- msg.pid = htonl (fc->last_pid_sent);
GNUNET_PEER_resolve (t->id.oid, &msg.oid);
if (fc == &t->prev_fc)
GNUNET_break (0);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** peer: %s!\n",
+ GNUNET_i2s(GNUNET_PEER_resolve2 (peer)));
+ if (0 == peer)
+ {
+ if (GNUNET_YES == t->destroy)
+ tunnel_destroy (t);
+ else
+ GNUNET_break (0);
+
+ return;
+ }
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
+ msg.header.size = htons (sizeof (msg));
+ msg.tid = htonl (t->id.tid);
+ msg.pid = htonl (peer_get_first_payload_pid (peer_get_short (peer), t));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** pid (%u)!\n", ntohl (msg.pid));
send_prebuilt_message (&msg.header, peer, t);
fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
for (i = 0; i < p->length && p->peers[i] != myid; i++) /* skip'em */ ;
for (i++; i < p->length; i++)
{
- struct MeshPeerInfo *aux;
+ struct MeshPeer *aux;
struct MeshPeerPath *copy;
aux = peer_get_short (p->peers[i]);
copy = path_duplicate (p);
copy->length = i + 1;
- peer_info_add_path (aux, copy, p->length < 3 ? GNUNET_NO : confirmed);
+ peer_add_path (aux, copy, p->length < 3 ? GNUNET_NO : confirmed);
}
}
-/**
- * Send keepalive packets for a peer
- *
- * @param cls Closure (tunnel for which to send the keepalive).
- * @param tc Notification context.
- */
-static void
-path_refresh (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
/**
* Search for a tunnel among the incoming tunnels
*
static struct MeshTunnel *
tunnel_get_by_local_id (struct MeshClient *c, MESH_TunnelNumber tid)
{
+ if (0 == (tid & GNUNET_MESH_LOCAL_TUNNEL_ID_CLI))
+ {
+ GNUNET_break_op (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "TID %X not a local tid\n", tid);
+ return NULL;
+ }
if (tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
{
return tunnel_get_incoming (tid);
}
else
{
- GNUNET_assert (tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_CLI);
return GNUNET_CONTAINER_multihashmap32_get (c->own_tunnels, tid);
}
}
}
+/**
+ * Change the tunnel state.
+ *
+ * @param t Tunnel whose ttate to change.
+ * @param state New state.
+ */
+static void
+tunnel_change_state (struct MeshTunnel *t, enum MeshTunnelState state)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Tunnel %s[%X] state was %s\n",
+ GNUNET_i2s (GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid,
+ GNUNET_MESH_DEBUG_S2S (t->state));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Tunnel %s[%X] state is now %s\n",
+ GNUNET_i2s (GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid,
+ GNUNET_MESH_DEBUG_S2S (state));
+ t->state = state;
+}
+
+
+
/**
* Add a client to a tunnel, initializing all needed data structures.
- *
+ *
* @param t Tunnel to which add the client.
* @param c Client which to add to the tunnel.
*/
t->path = path_duplicate (p);
if (0 == own_pos)
{
- if (GNUNET_SCHEDULER_NO_TASK != t->maintenance_task)
- GNUNET_SCHEDULER_cancel (t->maintenance_task);
- t->maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_path_time,
- &path_refresh, t);
+ if (GNUNET_SCHEDULER_NO_TASK != t->fwd_maintenance_task)
+ GNUNET_SCHEDULER_cancel (t->fwd_maintenance_task);
+ t->fwd_maintenance_task =
+ GNUNET_SCHEDULER_add_delayed (refresh_path_time,
+ &tunnel_fwd_keepalive, t);
}
}
// {
// return;
// }
-//
+//
// if (tree_get_predecessor (t->tree) != 0)
// {
// /* We are the peer still connected, notify owner of the disconnection. */
// struct GNUNET_MESH_PathBroken msg;
// struct GNUNET_PeerIdentity neighbor;
-//
+//
// msg.header.size = htons (sizeof (msg));
// msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_BROKEN);
// GNUNET_PEER_resolve (t->id.oid, &msg.oid);
}
-/**
- * Build a local ACK message and send it to a local client.
- *
- * @param t Tunnel on which to send the ACK.
- * @param c Client to whom send the ACK.
- * @param is_fwd Set to GNUNET_YES for FWD ACK (dest->owner)
- */
-static void
-send_local_ack (struct MeshTunnel *t,
- struct MeshClient *c,
- int is_fwd)
-{
- struct GNUNET_MESH_LocalAck msg;
-
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
- msg.tunnel_id = htonl (is_fwd ? t->local_tid : t->local_tid_dest);
- GNUNET_SERVER_notification_context_unicast (nc,
- c->handle,
- &msg.header,
- GNUNET_NO);
-}
-
-/**
- * Build an ACK message and queue it to send to the given peer.
- *
- * @param t Tunnel on which to send the ACK.
- * @param peer Peer to whom send the ACK.
- * @param ack Value of the ACK.
- */
-static void
-send_ack (struct MeshTunnel *t, GNUNET_PEER_Id peer, uint32_t ack)
-{
- struct GNUNET_MESH_ACK msg;
-
- GNUNET_PEER_resolve (t->id.oid, &msg.oid);
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
- msg.pid = htonl (ack);
- msg.tid = htonl (t->id.tid);
-
- send_prebuilt_message (&msg.header, peer, t);
-}
-
-
/**
* Send an end-to-end FWD ACK message for the most recent in-sequence payload.
- *
+ *
* @param t Tunnel this is about.
+ * @param fwd Is for FWD traffic? (ACK dest->owner)
*/
static void
-tunnel_send_fwd_data_ack (struct MeshTunnel *t)
+tunnel_send_data_ack (struct MeshTunnel *t, int fwd)
{
struct GNUNET_MESH_DataACK msg;
+ struct MeshTunnelReliability *rel;
+ struct MeshReliableMessage *copy;
+ GNUNET_PEER_Id hop;
+ uint64_t mask;
+ unsigned int delta;
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK);
- msg.header.size = htons (sizeof (msg));
- msg.tid = htonl (t->id.tid);
- GNUNET_PEER_resolve (t->id.oid, &msg.oid);
- msg.pid = htonl (t->prev_fc.last_pid_recv);
- msg.futures = 0; // FIXME set bits of other newer messages received
-
- send_prebuilt_message (&msg.header, t->prev_hop, t);
-}
-
-
-/**
- * Send an end-to-end BCK ACK message for the most recent in-sequence payload.
- *
- * @param t Tunnel this is about.
- */
-static void
-tunnel_send_bck_data_ack (struct MeshTunnel *t)
-{
- struct GNUNET_MESH_DataACK msg;
+ rel = fwd ? t->bck_rel : t->fwd_rel;
+ hop = fwd ? t->prev_hop : t->next_hop;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "send_data_ack for %u\n",
+ rel->mid_recv - 1);
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK);
+ if (GNUNET_NO == t->reliable)
+ {
+ GNUNET_break_op (0);
+ return;
+ }
+ msg.header.type = htons (fwd ? GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK :
+ GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK);
msg.header.size = htons (sizeof (msg));
msg.tid = htonl (t->id.tid);
GNUNET_PEER_resolve (t->id.oid, &msg.oid);
- msg.pid = htonl (t->next_fc.last_pid_recv);
- msg.futures = 0; // FIXME set bits of other newer messages received
+ msg.mid = htonl (rel->mid_recv - 1);
+ msg.futures = 0;
+ for (copy = rel->head_recv; NULL != copy; copy = copy->next)
+ {
+ delta = copy->mid - rel->mid_recv;
+ if (63 < delta)
+ break;
+ mask = 0x1LL << delta;
+ msg.futures |= mask;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " setting bit for %u (delta %u) (%llX) -> %llX\n",
+ copy->mid, delta, mask, msg.futures);
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " final futures %llX\n", msg.futures);
- send_prebuilt_message (&msg.header, t->next_hop, t);
+ send_prebuilt_message (&msg.header, hop, t);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_data_ack END\n");
}
* If buffering is on, send when sent to children and buffer space is free.
* Note that although the name is fwd_ack, the FWD mean forward *traffic*,
* the ACK itself goes "back" (towards root).
- *
- * @param t Tunnel on which to send the ACK.
- * @param type Type of message that triggered the ACK transmission.
- */
-static void
-tunnel_send_fwd_ack (struct MeshTunnel *t, uint16_t type)
-{
- uint32_t ack;
-
- /* Is it after unicast retransmission? */
- switch (type)
- {
- case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "ACK due to FWD DATA retransmission\n");
- if (GNUNET_YES == t->nobuffer)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, nobuffer\n");
- return;
- }
- if (GNUNET_YES == t->reliable && NULL != t->client)
- tunnel_send_fwd_data_ack (t);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_ACK:
- if (NULL != t->owner && GNUNET_YES == t->reliable)
- return;
- case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
- break;
- case GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK:
- tunnel_send_fwd_data_ack (t);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_POLL:
- case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
- t->force_ack = GNUNET_YES;
- break;
- default:
- GNUNET_break (0);
- }
-
- /* Check if we need to transmit the ACK */
- if (NULL == t->owner &&
- t->queue_max > t->next_fc.queue_n * 4 &&
- GMC_is_pid_bigger(t->prev_fc.last_ack_sent, t->prev_fc.last_pid_recv) &&
- GNUNET_NO == t->force_ack)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, buffer free\n");
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " t->qmax: %u, t->qn: %u\n",
- t->queue_max, t->next_fc.queue_n);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " t->pid: %u, t->ack: %u\n",
- t->prev_fc.last_pid_recv, t->prev_fc.last_ack_sent);
- return;
- }
-
- /* Ok, ACK might be necessary, what PID to ACK? */
- ack = t->prev_fc.last_pid_recv + t->queue_max - t->next_fc.queue_n;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK %u\n", ack);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " last %u, qmax %u, q %u\n",
- t->prev_fc.last_pid_recv, t->queue_max, t->next_fc.queue_n);
- if (ack == t->prev_fc.last_ack_sent && GNUNET_NO == t->force_ack)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
- return;
- }
-
- t->prev_fc.last_ack_sent = ack;
- if (NULL != t->owner)
- send_local_ack (t, t->owner, GNUNET_YES);
- else if (0 != t->prev_hop)
- send_ack (t, t->prev_hop, ack);
- else
- GNUNET_break (0);
- debug_fwd_ack++;
- t->force_ack = GNUNET_NO;
-}
-
-
-/**
- * Send an ACK informing the children node/client about the available
- * buffer space.
- * If buffering is off, send only on behalf of root (can be self).
- * If buffering is on, send when sent to predecessor and buffer space is free.
- * Note that although the name is bck_ack, the BCK mean backwards *traffic*,
- * the ACK itself goes "forward" (towards children/clients).
- *
+ *
* @param t Tunnel on which to send the ACK.
* @param type Type of message that triggered the ACK transmission.
+ * @param fwd Is this FWD ACK? (Going dest->owner)
*/
static void
-tunnel_send_bck_ack (struct MeshTunnel *t, uint16_t type)
+tunnel_send_ack (struct MeshTunnel *t, uint16_t type, int fwd)
{
+ struct MeshTunnelReliability *rel;
+ struct MeshFlowControl *next_fc;
+ struct MeshFlowControl *prev_fc;
+ struct MeshClient *c;
+ struct MeshClient *o;
+ GNUNET_PEER_Id hop;
+ uint32_t delta_mid;
uint32_t ack;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Sending BCK ACK on tunnel %u [%u] due to %s\n",
- t->id.oid, t->id.tid, GNUNET_MESH_DEBUG_M2S(type));
- /* Is it after data to_origin retransmission? */
+ int delta;
+
+ rel = fwd ? t->fwd_rel : t->bck_rel;
+ c = fwd ? t->client : t->owner;
+ o = fwd ? t->owner : t->client;
+ next_fc = fwd ? &t->next_fc : &t->prev_fc;
+ prev_fc = fwd ? &t->prev_fc : &t->next_fc;
+ hop = fwd ? t->prev_hop : t->next_hop;
+
switch (type)
{
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
- if (GNUNET_YES == t->nobuffer)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "ACK due to %s\n",
+ GNUNET_MESH_DEBUG_M2S (type));
+ if (GNUNET_YES == t->nobuffer && (GNUNET_NO == t->reliable || NULL == c))
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " Not sending ACK, nobuffer + traffic\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, nobuffer\n");
return;
}
+ if (GNUNET_YES == t->reliable && NULL != c)
+ tunnel_send_data_ack (t, fwd);
break;
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK:
case GNUNET_MESSAGE_TYPE_MESH_ACK:
- /* Why was this here?!
- * This prevents the destination from starting traffic to the origin
- */
-// if (NULL != t->client && GNUNET_YES == t->reliable)
-// return;
case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
break;
- case GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK:
- tunnel_send_bck_data_ack (t);
- /* fall through */
- case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
case GNUNET_MESSAGE_TYPE_MESH_POLL:
+ case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
t->force_ack = GNUNET_YES;
break;
default:
GNUNET_break (0);
}
- /* TODO: Check if we need to transmit the ACK (as in fwd) */
-
- ack = t->next_fc.last_pid_recv + t->queue_max - t->prev_fc.queue_n;
-
- if (t->next_fc.last_ack_sent == ack && GNUNET_NO == t->force_ack)
+ /* Check if we need to transmit the ACK */
+ if (NULL == o &&
+ prev_fc->last_ack_sent - prev_fc->last_pid_recv > 3 &&
+ GNUNET_NO == t->force_ack)
{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer free\n");
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " Not sending ACK, not needed, last ack sent was %u\n",
- t->next_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 = t->queue_max - next_fc->queue_n;
+ if (NULL != o && GNUNET_YES == t->reliable && NULL != rel->head_sent)
+ delta_mid = rel->mid_sent - rel->head_sent->mid;
+ else
+ delta_mid = 0;
+ if (0 > delta || (GNUNET_YES == t->reliable &&
+ NULL != o &&
+ (10 < rel->n_sent || 64 <= delta_mid)))
+ delta = 0;
+ if (NULL != o && delta > 1)
+ delta = 1;
+ ack = prev_fc->last_pid_recv + delta;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " Sending BCK ACK %u (last sent: %u)\n",
- ack, t->next_fc.last_ack_sent);
- t->next_fc.last_ack_sent = ack;
+ " last pid %u, last ack %u, qmax %u, q %u\n",
+ prev_fc->last_pid_recv, prev_fc->last_ack_sent,
+ t->queue_max, next_fc->queue_n);
+ if (ack == prev_fc->last_ack_sent && GNUNET_NO == t->force_ack)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
+ return;
+ }
- if (NULL != t->client)
- send_local_ack (t, t->client, GNUNET_NO);
- else if (0 != t->next_hop)
- send_ack (t, t->next_hop, ack);
+ prev_fc->last_ack_sent = ack;
+ if (NULL != o)
+ send_local_ack (t, o, fwd);
+ else if (0 != hop)
+ send_ack (t, hop, ack);
else
- GNUNET_break (0);
+ GNUNET_break (GNUNET_YES == t->destroy);
t->force_ack = GNUNET_NO;
}
/**
* Modify the mesh message TID from global to local and send to client.
- *
+ *
* @param t Tunnel on which to send the message.
* @param msg Message to modify and send.
* @param c Client to send to.
* @param tid Tunnel ID to use (c can be both owner and client).
*/
static void
-tunnel_send_client_data (struct MeshTunnel *t,
- const struct GNUNET_MESH_Data *msg,
- struct MeshClient *c, MESH_TunnelNumber tid)
+tunnel_send_client_to_tid (struct MeshTunnel *t,
+ const struct GNUNET_MESH_Data *msg,
+ struct MeshClient *c, MESH_TunnelNumber tid)
{
struct GNUNET_MESH_LocalData *copy;
uint16_t size = ntohs (msg->header.size) - sizeof (struct GNUNET_MESH_Data);
/**
* Modify the unicast message TID from global to local and send to client.
- *
+ *
* @param t Tunnel on which to send the message.
* @param msg Message to modify and send.
+ * @param fwd Forward?
*/
static void
-tunnel_send_client_ucast (struct MeshTunnel *t,
- const struct GNUNET_MESH_Data *msg)
+tunnel_send_client_data (struct MeshTunnel *t,
+ const struct GNUNET_MESH_Data *msg,
+ int fwd)
{
- tunnel_send_client_data (t, msg, t->client, t->local_tid_dest);
+ if (fwd)
+ tunnel_send_client_to_tid (t, msg, t->client, t->local_tid_dest);
+ else
+ tunnel_send_client_to_tid (t, msg, t->owner, t->local_tid);
}
/**
- * Modify the to_origin message TID from global to local and send to client.
- *
- * @param t Tunnel on which to send the message.
- * @param msg Message to modify and send.
+ * Send up to 64 buffered messages to the client for in order delivery.
+ *
+ * @param t Tunnel on which to empty the message buffer.
+ * @param c Client to send to.
+ * @param rel Reliability structure to corresponding peer.
+ * If rel == t->bck_rel, this is FWD data.
+ */
+static void
+tunnel_send_client_buffered_data (struct MeshTunnel *t, struct MeshClient *c,
+ struct MeshTunnelReliability *rel)
+{
+ ;
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *next;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_buffered_data\n");
+ for (copy = rel->head_recv; NULL != copy; copy = next)
+ {
+ next = copy->next;
+ if (copy->mid == rel->mid_recv)
+ {
+ struct GNUNET_MESH_Data *msg = (struct GNUNET_MESH_Data *) ©[1];
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " have %u! now expecting %u\n",
+ copy->mid, rel->mid_recv + 1);
+ tunnel_send_client_data (t, msg, (rel == t->bck_rel));
+ rel->mid_recv++;
+ GNUNET_CONTAINER_DLL_remove (rel->head_recv, rel->tail_recv, copy);
+ GNUNET_free (copy);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " don't have %u, next is %u\n",
+ rel->mid_recv,
+ copy->mid);
+ return;
+ }
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_buffered_data END\n");
+}
+
+
+/**
+ * We have received a message out of order, buffer it until we receive
+ * the missing one and we can feed the rest to the client.
+ *
+ * @param t Tunnel to add to.
+ * @param msg Message to buffer.
+ * @param rel Reliability data to the corresponding direction.
+ */
+static void
+tunnel_add_buffered_data (struct MeshTunnel *t,
+ const struct GNUNET_MESH_Data *msg,
+ struct MeshTunnelReliability *rel)
+{
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *prev;
+ uint32_t mid;
+ uint16_t size;
+
+ size = ntohs (msg->header.size);
+ mid = ntohl (msg->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "add_buffered_data %u\n", mid);
+
+ copy = GNUNET_malloc (sizeof (*copy) + size);
+ copy->mid = mid;
+ copy->rel = rel;
+ memcpy (©[1], msg, size);
+
+ // FIXME do something better than O(n), although n < 64...
+ // FIXME start from the end (most messages are the latest ones)
+ for (prev = rel->head_recv; NULL != prev; prev = prev->next)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " prev %u\n", prev->mid);
+ if (GMC_is_pid_bigger (prev->mid, mid))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " bingo!\n");
+ GNUNET_CONTAINER_DLL_insert_before (rel->head_recv, rel->tail_recv,
+ prev, copy);
+ return;
+ }
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " insert at tail!\n");
+ GNUNET_CONTAINER_DLL_insert_tail (rel->head_recv, rel->tail_recv, copy);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "add_buffered_data END\n");
+}
+
+
+/**
+ * Destroy a reliable message after it has been acknowledged, either by
+ * direct mid ACK or bitfield. Updates the appropriate data structures and
+ * timers and frees all memory.
+ *
+ * @param copy Message that is no longer needed: remote peer got it.
+ */
+static void
+tunnel_free_reliable_message (struct MeshReliableMessage *copy)
+{
+ struct MeshTunnelReliability *rel;
+ struct GNUNET_TIME_Relative time;
+
+ rel = copy->rel;
+ time = GNUNET_TIME_absolute_get_duration (copy->timestamp);
+ rel->expected_delay.rel_value_us *= 7;
+ rel->expected_delay.rel_value_us += time.rel_value_us;
+ rel->expected_delay.rel_value_us /= 8;
+ rel->n_sent--;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! Freeing %u\n", copy->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " n_sent %u\n", rel->n_sent);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! took %s\n",
+ GNUNET_STRINGS_relative_time_to_string (time, GNUNET_NO));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! new expected delay %s\n",
+ GNUNET_STRINGS_relative_time_to_string (rel->expected_delay,
+ GNUNET_NO));
+ rel->retry_timer = rel->expected_delay;
+ GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
+ GNUNET_free (copy);
+}
+
+
+/**
+ * Destroy all reliable messages queued for a tunnel,
+ * during a tunnel destruction.
+ * Frees the reliability structure itself.
+ *
+ * @param rel Reliability data for a tunnel.
+ */
+static void
+tunnel_free_reliable_all (struct MeshTunnelReliability *rel)
+{
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *next;
+
+ if (NULL == rel)
+ return;
+
+ for (copy = rel->head_recv; NULL != copy; copy = next)
+ {
+ next = copy->next;
+ GNUNET_CONTAINER_DLL_remove (rel->head_recv, rel->tail_recv, copy);
+ GNUNET_free (copy);
+ }
+ for (copy = rel->head_sent; NULL != copy; copy = next)
+ {
+ next = copy->next;
+ GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
+ GNUNET_free (copy);
+ }
+ if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
+ GNUNET_SCHEDULER_cancel (rel->retry_task);
+ GNUNET_free (rel);
+}
+
+
+/**
+ * Mark future messages as ACK'd.
+ *
+ * @param t Tunnel whose sent buffer to clean.
+ * @param msg DataACK message with a bitfield of future ACK'd messages.
+ * @param rel Reliability data.
*/
static void
-tunnel_send_client_to_orig (struct MeshTunnel *t,
- const struct GNUNET_MESH_Data *msg)
+tunnel_free_sent_reliable (struct MeshTunnel *t,
+ const struct GNUNET_MESH_DataACK *msg,
+ struct MeshTunnelReliability *rel)
{
- tunnel_send_client_data (t, msg, t->owner, t->local_tid);
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *next;
+ uint64_t bitfield;
+ uint64_t mask;
+ uint32_t mid;
+ uint32_t target;
+ unsigned int i;
+
+ bitfield = msg->futures;
+ mid = ntohl (msg->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "free_sent_reliable %u %llX\n",
+ mid, bitfield);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " rel %p, head %p\n",
+ rel, rel->head_sent);
+ for (i = 0, copy = rel->head_sent;
+ i < 64 && NULL != copy && 0 != bitfield;
+ i++)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " trying bit %u (mid %u)\n",
+ i, mid + i + 1);
+ mask = 0x1LL << i;
+ if (0 == (bitfield & mask))
+ continue;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " set!\n");
+ /* Bit was set, clear the bit from the bitfield */
+ bitfield &= ~mask;
+
+ /* The i-th bit was set. Do we have that copy? */
+ /* Skip copies with mid < target */
+ target = mid + i + 1;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " target %u\n", target);
+ while (NULL != copy && GMC_is_pid_bigger (target, copy->mid))
+ copy = copy->next;
+
+ /* Did we run out of copies? (previously freed, it's ok) */
+ if (NULL == copy)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "run out of copies...\n");
+ return;
+ }
+
+ /* Did we overshoot the target? (previously freed, it's ok) */
+ if (GMC_is_pid_bigger (copy->mid, target))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " next copy %u\n", copy->mid);
+ continue;
+ }
+
+ /* Now copy->mid == target, free it */
+ next = copy->next;
+ tunnel_free_reliable_message (copy);
+ copy = next;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "free_sent_reliable END\n");
}
{
struct MeshTunnelReliability *rel = cls;
struct MeshReliableMessage *copy;
+ struct MeshFlowControl *fc;
+ struct MeshPeerQueue *q;
+ struct MeshPeer *pi;
struct MeshTunnel *t;
struct GNUNET_MESH_Data *payload;
GNUNET_PEER_Id hop;
t = rel->t;
copy = rel->head_sent;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! RETRANSMIT %u\n", copy->id);
+ if (NULL == copy)
+ {
+ GNUNET_break (0);
+ return;
+ }
+ /* Search the message to be retransmitted in the outgoing queue */
payload = (struct GNUNET_MESH_Data *) ©[1];
hop = rel == t->fwd_rel ? t->next_hop : t->prev_hop;
- send_prebuilt_message (&payload->header, hop, t);
- GNUNET_STATISTICS_update (stats, "# data retransmitted", 1, GNUNET_NO);
- rel->retry_timer = GNUNET_TIME_STD_BACKOFF (rel->retry_timer); // FIXME adapt
+ fc = rel == t->fwd_rel ? &t->prev_fc : &t->next_fc;
+ pi = peer_get_short (hop);
+ for (q = pi->queue_head; NULL != q; q = q->next)
+ {
+ if (ntohs (payload->header.type) == q->type)
+ {
+ struct GNUNET_MESH_Data *queued_data = q->cls;
+
+ if (queued_data->mid == payload->mid)
+ break;
+ }
+ }
+
+ /* Message not found in the queue */
+ if (NULL == q)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! RETRANSMIT %u\n", copy->mid);
+
+ fc->last_ack_sent++;
+ fc->last_pid_recv++;
+ payload->pid = htonl (fc->last_pid_recv);
+ send_prebuilt_message (&payload->header, hop, t);
+ GNUNET_STATISTICS_update (stats, "# data retransmitted", 1, GNUNET_NO);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! STILL IN QUEUE %u\n", copy->mid);
+ }
+
+ rel->retry_timer = GNUNET_TIME_STD_BACKOFF (rel->retry_timer);
rel->retry_task = GNUNET_SCHEDULER_add_delayed (rel->retry_timer,
- &tunnel_retransmit_message,
- cls);
+ &tunnel_retransmit_message,
+ cls);
}
/**
- * @brief Re-initiate traffic to this peer if necessary.
+ * Send keepalive packets for a tunnel.
*
- * 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 t Tunnel to keep alive..
+ * @param fwd Is this a FWD keepalive? (owner -> dest).
+ */
+static void
+tunnel_keepalive (struct MeshTunnel *t, int fwd)
+{
+ struct GNUNET_MESH_TunnelKeepAlive *msg;
+ size_t size = sizeof (struct GNUNET_MESH_TunnelKeepAlive);
+ char cbuf[size];
+ GNUNET_PEER_Id hop;
+ uint16_t type;
+
+ type = fwd ? GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE :
+ GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE;
+ hop = fwd ? t->next_hop : t->prev_hop;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending %s keepalive for tunnel %d\n",
+ fwd ? "FWD" : "BCK", t->id.tid);
+
+ msg = (struct GNUNET_MESH_TunnelKeepAlive *) cbuf;
+ msg->header.size = htons (size);
+ msg->header.type = htons (type);
+ msg->oid = *(GNUNET_PEER_resolve2 (t->id.oid));
+ msg->tid = htonl (t->id.tid);
+ send_prebuilt_message (&msg->header, hop, t);
+}
+
+
+/**
+ * Send create (PATH_CREATE/PATH_ACK) packets for a tunnel.
*
- * @param peer_id Short ID of peer to which initiate traffic.
+ * @param t Tunnel for which to send the message.
+ * @param fwd If GNUNET_YES, send CREATE, otherwise send ACK.
*/
static void
-peer_unlock_queue(GNUNET_PEER_Id peer_id)
+tunnel_recreate (struct MeshTunnel *t, int fwd)
{
- struct MeshPeerInfo *peer;
- struct GNUNET_PeerIdentity id;
- struct MeshPeerQueue *q;
- size_t size;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending path recreate for tunnel %s[%X]\n",
+ GNUNET_i2s (GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid);
+ if (fwd)
+ send_path_create (t);
+ else
+ send_path_ack (t);
+}
- peer = peer_get_short (peer_id);
- if (NULL != peer->core_transmit)
- return;
- q = queue_get_next (peer);
- if (NULL == q)
+/**
+ * Generic tunnel timer management.
+ * Depending on the role of the peer in the tunnel will send the
+ * appropriate message (build or keepalive)
+ *
+ * @param t Tunnel to maintain.
+ * @param fwd Is FWD?
+ */
+static void
+tunnel_maintain (struct MeshTunnel *t, int fwd)
+{
+ switch (t->state)
{
- /* Might br multicast traffic already sent to this particular peer but
- * not to other children in this tunnel.
- * This way t->queue_n would be > 0 but the queue of this particular peer
- * would be empty.
- */
- return;
+ case MESH_TUNNEL_NEW:
+ GNUNET_break (0);
+ case MESH_TUNNEL_SEARCHING:
+ /* TODO DHT GET with RO_BART */
+ break;
+ case MESH_TUNNEL_WAITING:
+ tunnel_recreate (t, fwd);
+ break;
+ case MESH_TUNNEL_READY:
+ tunnel_keepalive (t, fwd);
+ break;
+ default:
+ break;
}
- size = q->size;
- GNUNET_PEER_resolve (peer->id, &id);
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready(core_handle,
- 0,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &id,
- size,
- &queue_send,
- peer);
- return;
+}
+
+
+static void
+tunnel_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshTunnel *t = cls;
+
+ t->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) ||
+ NULL == t->owner)
+ return;
+
+ tunnel_maintain (t, GNUNET_YES);
+ t->fwd_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_path_time,
+ &tunnel_fwd_keepalive,
+ t);
+}
+
+
+static void
+tunnel_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct MeshTunnel *t = cls;
+
+ t->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) ||
+ NULL == t->client)
+ return;
+
+ tunnel_keepalive (t, GNUNET_NO);
+ t->bck_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_path_time,
+ &tunnel_bck_keepalive,
+ t);
}
GNUNET_PEER_resolve (t->next_hop, &id);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" sending forward to %s\n",
- GNUNET_i2s (&id));
- send_prebuilt_message (&msg.header, t->next_hop, t);
- }
- if (NULL == t->owner && 0 != t->prev_hop)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " parent: %u\n", t->prev_hop);
- GNUNET_PEER_resolve (t->prev_hop, &id);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " sending back to %s\n",
- GNUNET_i2s (&id));
- send_prebuilt_message (&msg.header, t->prev_hop, t);
- }
- if (NULL != t->owner)
- {
- send_client_tunnel_destroy (t->owner, t);
- }
- if (NULL != t->client)
- {
- send_client_tunnel_destroy (t->client, t);
- }
-}
-
-
-/**
- * Cancel all transmissions towards a neighbor that belongs to a certain tunnel.
- *
- * @param t Tunnel which to cancel.
- * @param neighbor Short ID of the neighbor to whom cancel the transmissions.
- */
-static void
-peer_cancel_queues (GNUNET_PEER_Id neighbor, struct MeshTunnel *t)
-{
- struct MeshPeerInfo *peer_info;
- struct MeshPeerQueue *pq;
- struct MeshPeerQueue *next;
-
- if (0 == neighbor)
- return; /* Was local peer, 0'ed in tunnel_destroy_iterator */
- peer_info = peer_get_short (neighbor);
- for (pq = peer_info->queue_head; NULL != pq; pq = next)
+ GNUNET_i2s (&id));
+ send_prebuilt_message (&msg.header, t->next_hop, t);
+ }
+ if (NULL == t->owner && 0 != t->prev_hop)
{
- next = pq->next;
- if (pq->tunnel == t)
- {
- if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == pq->type ||
- GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == pq->type)
- {
- /* Should have been removed on destroy children */
- GNUNET_break (0);
- }
- queue_destroy (pq, GNUNET_YES);
- }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " parent: %u\n", t->prev_hop);
+ GNUNET_PEER_resolve (t->prev_hop, &id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " sending back to %s\n",
+ GNUNET_i2s (&id));
+ send_prebuilt_message (&msg.header, t->prev_hop, t);
+ }
+ if (NULL != t->owner)
+ {
+ send_local_tunnel_destroy (t, GNUNET_NO);
}
- if (NULL == peer_info->queue_head && NULL != peer_info->core_transmit)
+ if (NULL != t->client)
{
- GNUNET_CORE_notify_transmit_ready_cancel(peer_info->core_transmit);
- peer_info->core_transmit = NULL;
+ send_local_tunnel_destroy (t, GNUNET_YES);
}
}
-
-/**
- * Destroy the tunnel.
- *
- * This function does not generate any warning traffic to clients or peers.
- *
- * Tasks:
- * Remove the tunnel from peer_info's and clients' hashmaps.
- * Cancel messages belonging to this tunnel queued to neighbors.
- * Free any allocated resources linked to the tunnel.
- *
- * @param t the tunnel to destroy
- *
- * @return GNUNET_OK on success
- */
static int
tunnel_destroy (struct MeshTunnel *t)
{
r = GNUNET_OK;
c = t->owner;
-#if MESH_DEBUG
- {
- struct GNUNET_PeerIdentity id;
- GNUNET_PEER_resolve (t->id.oid, &id);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying tunnel %s [%x]\n",
- GNUNET_i2s (&id), t->id.tid);
- if (NULL != c)
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
- }
-#endif
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying tunnel %s [%x]\n",
+ GNUNET_i2s (GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid);
+ if (NULL != c)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
GNUNET_CRYPTO_hash (&t->id, sizeof (struct MESH_TunnelID), &hash);
if (GNUNET_YES != GNUNET_CONTAINER_multihashmap_remove (tunnels, &hash, t))
if (NULL != c)
{
- if (GNUNET_YES !=
- GNUNET_CONTAINER_multihashmap32_remove (c->own_tunnels, t->local_tid, t))
+ if (GNUNET_YES != GNUNET_CONTAINER_multihashmap32_remove (c->own_tunnels,
+ t->local_tid, t))
{
GNUNET_break (0);
r = GNUNET_SYSERR;
}
}
- if (NULL != t->client)
+ c = t->client;
+ if (NULL != c)
{
- c = t->client;
if (GNUNET_YES !=
GNUNET_CONTAINER_multihashmap32_remove (c->incoming_tunnels,
t->local_tid_dest, t))
GNUNET_break (0);
r = GNUNET_SYSERR;
}
- if (GNUNET_YES !=
+ if (GNUNET_YES !=
GNUNET_CONTAINER_multihashmap32_remove (incoming_tunnels,
t->local_tid_dest, t))
{
}
}
+ if(GNUNET_YES == t->reliable)
+ {
+ tunnel_free_reliable_all (t->fwd_rel);
+ tunnel_free_reliable_all (t->bck_rel);
+ }
if (0 != t->prev_hop)
{
peer_cancel_queues (t->prev_hop, t);
peer_cancel_queues (t->next_hop, t);
GNUNET_PEER_change_rc (t->next_hop, -1);
}
+ if (GNUNET_SCHEDULER_NO_TASK != t->next_fc.poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (t->next_fc.poll_task);
+ t->next_fc.poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ if (GNUNET_SCHEDULER_NO_TASK != t->prev_fc.poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (t->prev_fc.poll_task);
+ t->prev_fc.poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
if (0 != t->dest) {
- peer_info_remove_tunnel (peer_get_short (t->dest), t);
+ peer_remove_tunnel (peer_get_short (t->dest), t);
}
- if (GNUNET_SCHEDULER_NO_TASK != t->maintenance_task)
- GNUNET_SCHEDULER_cancel (t->maintenance_task);
+ if (GNUNET_SCHEDULER_NO_TASK != t->fwd_maintenance_task)
+ GNUNET_SCHEDULER_cancel (t->fwd_maintenance_task);
+ if (GNUNET_SCHEDULER_NO_TASK != t->bck_maintenance_task)
+ GNUNET_SCHEDULER_cancel (t->bck_maintenance_task);
n_tunnels--;
GNUNET_STATISTICS_update (stats, "# tunnels", -1, GNUNET_NO);
/**
* Tunnel is empty: destroy it.
- *
+ *
* Notifies all participants (peers, cleints) about the destruction.
- *
- * @param t Tunnel to destroy.
+ *
+ * @param t Tunnel to destroy.
*/
static void
tunnel_destroy_empty (struct MeshTunnel *t)
/**
* Initialize a Flow Control structure to the initial state.
- *
+ *
* @param fc Flow Control structure to initialize.
*/
static void
/**
* Create a new tunnel
- *
+ *
* @param owner Who is the owner of the tunnel (short ID).
* @param tid Tunnel Number of the tunnel.
* @param client Clients that owns the tunnel, NULL for foreign tunnels.
* @param local Tunnel Number for the tunnel, for the client point of view.
- *
+ *
* @return A new initialized tunnel. NULL on error.
*/
static struct MeshTunnel *
t->owner = client;
fc_init (&t->next_fc);
fc_init (&t->prev_fc);
+ t->next_fc.t = t;
+ t->prev_fc.t = t;
t->local_tid = local;
n_tunnels++;
GNUNET_STATISTICS_update (stats, "# tunnels", 1, GNUNET_NO);
/**
* Set options in a tunnel, extracted from a bit flag field
- *
+ *
* @param t Tunnel to set options to.
* @param options Bit array in host byte order.
*/
/**
- * Timeout function, destroys tunnel if called
+ * remove client's ports from the global hashmap on diconnect.
+ *
+ * @param cls Closure (unused).
+ * @param key Port.
+ * @param value ThClient structure.
+ *
+ * @return GNUNET_OK, keep iterating.
+ */
+static int
+client_release_ports (void *cls,
+ uint32_t key,
+ void *value)
+{
+ int res;
+
+ res = GNUNET_CONTAINER_multihashmap32_remove (ports, key, value);
+ if (GNUNET_YES != res)
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Port %u by client %p was not registered.\n",
+ key, value);
+ }
+ return GNUNET_OK;
+}
+
+/**
+ * Timeout function due to lack of keepalive/traffic from the owner.
+ * Destroys tunnel if called.
*
* @param cls Closure (tunnel to destroy).
* @param tc TaskContext
*/
static void
-tunnel_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+tunnel_fwd_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct MeshTunnel *t = cls;
- struct GNUNET_PeerIdentity id;
- t->maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ t->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
return;
- GNUNET_PEER_resolve(t->id.oid, &id);
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Tunnel %s [%X] timed out. Destroying.\n",
- GNUNET_i2s(&id), t->id.tid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Tunnel %s [%X] FWD timed out. Destroying.\n",
+ GNUNET_i2s(GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid);
if (NULL != t->client)
- send_client_tunnel_destroy (t->client, t);
+ send_local_tunnel_destroy (t, GNUNET_YES);
tunnel_destroy (t); /* Do not notify other */
}
/**
- * Resets the tunnel timeout. Starts it if no timeout was running.
- *
- * @param t Tunnel whose timeout to reset.
+ * Timeout function due to lack of keepalive/traffic from the destination.
+ * Destroys tunnel if called.
*
- * TODO use heap to improve efficiency of scheduler.
+ * @param cls Closure (tunnel to destroy).
+ * @param tc TaskContext
*/
static void
-tunnel_reset_timeout (struct MeshTunnel *t)
+tunnel_bck_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
- if (NULL != t->owner || 0 != t->local_tid || 0 == t->prev_hop)
+ struct MeshTunnel *t = cls;
+
+ t->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
return;
- if (GNUNET_SCHEDULER_NO_TASK != t->maintenance_task)
- GNUNET_SCHEDULER_cancel (t->maintenance_task);
- t->maintenance_task =
- GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
- (refresh_path_time, 4), &tunnel_timeout, t);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Tunnel %s [%X] BCK timed out. Destroying.\n",
+ GNUNET_i2s(GNUNET_PEER_resolve2 (t->id.oid)), t->id.tid);
+ if (NULL != t->owner)
+ send_local_tunnel_destroy (t, GNUNET_NO);
+ tunnel_destroy (t); /* Do not notify other */
}
-/******************************************************************************/
-/**************** MESH NETWORK HANDLER HELPERS ***********************/
-/******************************************************************************/
-
/**
- * Function to send a create path packet to a peer.
+ * Resets the tunnel 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 tunnel,
+ * due to lack of activity.
+ * Starts the tiemout if no timeout was running (tunnel just created).
*
- * @param cls closure
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
+ * @param t Tunnel whose timeout to reset.
+ * @param fwd Is this forward?
+ *
+ * TODO use heap to improve efficiency of scheduler.
*/
-static size_t
-send_core_path_create (void *cls, size_t size, void *buf)
+static void
+tunnel_reset_timeout (struct MeshTunnel *t, int fwd)
{
- struct MeshTunnel *t = cls;
- struct GNUNET_MESH_CreateTunnel *msg;
- struct GNUNET_PeerIdentity *peer_ptr;
- struct MeshPeerPath *p = t->path;
- size_t size_needed;
- uint32_t opt;
- int i;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CREATE PATH sending...\n");
- size_needed =
- sizeof (struct GNUNET_MESH_CreateTunnel) +
- p->length * sizeof (struct GNUNET_PeerIdentity);
+ GNUNET_SCHEDULER_TaskIdentifier *ti;
+ GNUNET_SCHEDULER_Task f;
+ struct MeshClient *c;
- if (size < size_needed || NULL == buf)
- {
- GNUNET_break (0);
- return 0;
- }
- msg = (struct GNUNET_MESH_CreateTunnel *) buf;
- msg->header.size = htons (size_needed);
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE);
- msg->tid = ntohl (t->id.tid);
+ ti = fwd ? &t->fwd_maintenance_task : &t->bck_maintenance_task;
+ c = fwd ? t->owner : t->client;
- opt = 0;
- if (GNUNET_YES == t->nobuffer)
- opt |= GNUNET_MESH_OPTION_NOBUFFER;
- if (GNUNET_YES == t->reliable)
- opt |= GNUNET_MESH_OPTION_RELIABLE;
- msg->opt = htonl (opt);
- msg->port = htonl (t->port);
+ if (GNUNET_SCHEDULER_NO_TASK != *ti)
+ GNUNET_SCHEDULER_cancel (*ti);
- peer_ptr = (struct GNUNET_PeerIdentity *) &msg[1];
- for (i = 0; i < p->length; i++)
+ if (NULL != c)
{
- GNUNET_PEER_resolve (p->peers[i], peer_ptr++);
+ f = fwd ? &tunnel_fwd_keepalive : &tunnel_bck_keepalive;
+ *ti = GNUNET_SCHEDULER_add_delayed (refresh_path_time, f, t);
}
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "CREATE PATH (%u bytes long) sent!\n", size_needed);
- return size_needed;
-}
-
-
-/**
- * Creates a path ack message in buf and frees all unused resources.
- *
- * @param cls closure (MeshTransmissionDescriptor)
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-send_core_path_ack (void *cls, size_t size, void *buf)
-{
- struct MeshTunnel *t = cls;
- struct GNUNET_MESH_PathACK *msg = buf;
-
- GNUNET_assert (NULL != t);
- if (sizeof (struct GNUNET_MESH_PathACK) > size)
+ else
{
- GNUNET_break (0);
- return 0;
+ f = fwd ? &tunnel_fwd_timeout : &tunnel_bck_timeout;
+ *ti = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+ (refresh_path_time, 4),
+ f, t);
}
- t->prev_fc.last_ack_sent = t->nobuffer ? 0 : t->queue_max - 1;
- msg->header.size = htons (sizeof (struct GNUNET_MESH_PathACK));
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_ACK);
- GNUNET_PEER_resolve (t->id.oid, &msg->oid);
- msg->tid = htonl (t->id.tid);
- msg->peer_id = my_full_id;
- msg->ack = htonl (t->prev_fc.last_ack_sent);
-
- /* TODO add signature */
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "PATH ACK sent!\n");
- return sizeof (struct GNUNET_MESH_PathACK);
}
+/******************************************************************************/
+/**************** MESH NETWORK HANDLER HELPERS ***********************/
+/******************************************************************************/
+
/**
* Free a transmission that was already queued with all resources
* associated to the request.
case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK:
case GNUNET_MESSAGE_TYPE_MESH_POLL:
- case GNUNET_MESSAGE_TYPE_MESH_PATH_KEEPALIVE:
+ case GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE:
+ case GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE:
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" prebuilt message\n");
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
* NULL when there are no transmittable messages.
*/
struct MeshPeerQueue *
-queue_get_next (const struct MeshPeerInfo *peer)
+queue_get_next (const struct MeshPeer *peer)
{
struct MeshPeerQueue *q;
return q;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "* ACK: %u, PID: %u\n",
- ack, pid);
+ "* ACK: %u, PID: %u, MID: %u\n",
+ ack, pid, ntohl (dmsg->mid));
if (GNUNET_NO == GMC_is_pid_bigger (pid, ack))
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
static size_t
queue_send (void *cls, size_t size, void *buf)
{
- struct MeshPeerInfo *peer = cls;
+ struct MeshPeer *peer = cls;
struct GNUNET_MessageHeader *msg;
struct MeshPeerQueue *queue;
struct MeshTunnel *t;
peer->core_transmit = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Queue send\n");
+
+ if (NULL == buf || 0 == size)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Buffer size 0.\n");
+ return 0;
+ }
queue = queue_get_next (peer);
/* Queue has no internal mesh traffic nor sendable payload */
GNUNET_break (0); /* Core tmt_rdy should've been canceled */
return 0;
}
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* not empty\n");
GNUNET_PEER_resolve (peer->id, &dst_id);
case GNUNET_MESSAGE_TYPE_MESH_PATH_BROKEN:
case GNUNET_MESSAGE_TYPE_MESH_PATH_DESTROY:
case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
- case GNUNET_MESSAGE_TYPE_MESH_PATH_KEEPALIVE:
+ case GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE:
+ case GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE:
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"* raw: %s\n",
GNUNET_MESH_DEBUG_M2S (queue->type));
switch (type)
{
case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
- if (GMC_is_pid_bigger(pid, t->next_fc.last_pid_sent))
- t->next_fc.last_pid_sent = pid;
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
+ t->next_fc.last_pid_sent = pid;
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST, GNUNET_YES);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "!!! FWD %u\n",
+ ntohl ( ((struct GNUNET_MESH_Data *) buf)->mid ) );
break;
case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
- if (GMC_is_pid_bigger(pid, t->prev_fc.last_pid_sent))
- t->prev_fc.last_pid_sent = pid;
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
+ t->prev_fc.last_pid_sent = pid;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "!!! BCK %u\n",
+ ntohl ( ((struct GNUNET_MESH_Data *) buf)->mid ) );
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN, GNUNET_NO);
break;
default:
break;
}
- if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying tunnel!\n");
- tunnel_destroy (t);
- }
-
/* If more data in queue, send next */
queue = queue_get_next (peer);
if (NULL != queue)
{
- struct GNUNET_PeerIdentity id;
-
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* more data!\n");
- GNUNET_PEER_resolve (peer->id, &id);
- peer->core_transmit =
- GNUNET_CORE_notify_transmit_ready(core_handle,
- 0,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &id,
- queue->size,
- &queue_send,
- peer);
- }
- else if (NULL != peer->queue_head)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "* %s stalled\n",
- GNUNET_i2s (&my_full_id));
- if (peer->id == t->next_hop)
- fc = &t->next_fc;
- else if (peer->id == t->prev_hop)
- fc = &t->prev_fc;
- else
- {
- GNUNET_break (0);
- fc = NULL;
- }
- if (NULL != fc && GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
+ if (NULL == peer->core_transmit) {
+ peer->core_transmit =
+ GNUNET_CORE_notify_transmit_ready(core_handle,
+ 0,
+ 0,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &dst_id,
+ queue->size,
+ &queue_send,
+ peer);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "* tmt rdy called somewhere else\n");
+ }
+ }
+ if (peer->id == t->next_hop)
+ fc = &t->next_fc;
+ else if (peer->id == t->prev_hop)
+ fc = &t->prev_fc;
+ else
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "id: %u, next: %u, prev: %u\n",
+ peer->id, t->next_hop, t->prev_hop);
+ return data_size;
+ }
+ if (NULL != peer->queue_head)
+ {
+ if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task && fc->queue_n > 0)
{
- fc->t = t;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "* %s starting poll timeout\n",
+ GNUNET_i2s (&my_full_id));
fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
&tunnel_poll, fc);
}
}
+ else
+ {
+ if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
+ {
+ GNUNET_SCHEDULER_cancel (fc->poll_task);
+ fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ }
+ if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* destroying tunnel!\n");
+ tunnel_destroy (t);
+ }
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Return %d\n", data_size);
return data_size;
}
/**
* @brief Queue and pass message to core when possible.
- *
+ *
* If type is payload (UNICAST, TO_ORIGIN) checks for queue status and
* accounts for it. In case the queue is full, the message is dropped and
* a break issued.
- *
- * Otherwise, message is treated as internal and allowed to go regardless of
+ *
+ * Otherwise, message is treated as internal and allowed to go regardless of
* queue status.
*
* @param cls Closure (@c type dependant). It will be used by queue_send to
*/
static void
queue_add (void *cls, uint16_t type, size_t size,
- struct MeshPeerInfo *dst, struct MeshTunnel *t)
+ struct MeshPeer *dst, struct MeshTunnel *t)
{
struct MeshPeerQueue *queue;
- struct GNUNET_PeerIdentity id;
struct MeshFlowControl *fc;
- uint32_t pid;
- uint32_t pid_q;
int priority;
fc = NULL;
if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == type)
{
fc = &t->next_fc;
- pid = ntohl (((struct GNUNET_MESH_Data *)cls)->pid);
}
else if (GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == type)
{
fc = &t->prev_fc;
- pid = ntohl (((struct GNUNET_MESH_Data *)cls)->pid);
}
if (NULL != fc)
{
if (fc->queue_n >= t->queue_max)
{
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "queue full: %u/%u\n",
- fc->queue_n, t->queue_max);
- GNUNET_STATISTICS_update (stats,
- "# messages dropped (buffer full)",
- 1, GNUNET_NO);
- /* Get the PID of the oldest message in the queue */
- for (queue = dst->queue_head; queue != NULL; queue = queue->next)
- if (queue->type == type && queue->tunnel == t)
- {
- pid_q = ntohl (((struct GNUNET_MESH_Data *)(queue->cls))->pid);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "pid: %u, pid_q: %u\n", pid, pid_q);
- break;
- }
- GNUNET_assert (NULL != queue);
-
- /* If this is an earlier message that that, give it priority:
- * - drop the newest message in the queue
- * - instert current one at the end of the queue (first to get out)
- */
- if (GNUNET_YES == t->reliable && GMC_is_pid_bigger(pid_q, pid))
+ /* If this isn't a retransmission, drop the message */
+ if (GNUNET_NO == t->reliable ||
+ (NULL == t->owner && GNUNET_MESSAGE_TYPE_MESH_UNICAST == type) ||
+ (NULL == t->client && GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == type))
{
- for (queue = dst->queue_tail; queue != NULL; queue = queue->prev)
- if (queue->type == type && queue->tunnel == t)
- {
- /* Drop message from queue */
- pid_q = ntohl (((struct GNUNET_MESH_Data *)(queue->cls))->pid);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "dropping pid: %u\n", pid_q);
- queue_destroy (queue, GNUNET_YES);
- t->pending_messages--;
- priority = GNUNET_YES;
- break;
- }
- }
- 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, t->queue_max);
return; /* Drop this message */
+ }
+ priority = GNUNET_YES;
}
fc->queue_n++;
+ if (GMC_is_pid_bigger(fc->last_pid_sent + 1, fc->last_ack_recv) &&
+ GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
+ fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
+ &tunnel_poll,
+ fc);
}
queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
queue->cls = cls;
queue->peer = dst;
queue->tunnel = t;
if (GNUNET_YES == priority)
+ {
+ struct GNUNET_MESH_Data *d;
+ uint32_t prev;
+ uint32_t next;
+
GNUNET_CONTAINER_DLL_insert (dst->queue_head, dst->queue_tail, queue);
+ d = (struct GNUNET_MESH_Data *) queue->cls;
+ prev = d->pid;
+ for (queue = dst->queue_tail; NULL != queue; queue = queue->prev)
+ {
+ if (queue->type != type)
+ continue;
+ d = (struct GNUNET_MESH_Data *) queue->cls;
+ next = d->pid;
+ d->pid = prev;
+ prev = next;
+ }
+ }
else
GNUNET_CONTAINER_DLL_insert_tail (dst->queue_head, dst->queue_tail, queue);
+
if (NULL == dst->core_transmit)
{
- GNUNET_PEER_resolve (dst->id, &id);
dst->core_transmit =
GNUNET_CORE_notify_transmit_ready (core_handle,
0,
0,
GNUNET_TIME_UNIT_FOREVER_REL,
- &id,
+ GNUNET_PEER_resolve2 (dst->id),
size,
&queue_send,
dst);
struct GNUNET_MESH_CreateTunnel *msg;
struct GNUNET_PeerIdentity *pi;
struct MeshPeerPath *path;
- struct MeshPeerInfo *dest_peer_info;
- struct MeshPeerInfo *orig_peer_info;
+ struct MeshPeer *dest_peer_info;
+ struct MeshPeer *orig_peer_info;
struct MeshTunnel *t;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
tid = ntohl (msg->tid);
pi = (struct GNUNET_PeerIdentity *) &msg[1];
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " path is for tunnel %s[%X].\n", GNUNET_i2s (pi), tid);
+ " path is for tunnel %s[%X]:%u.\n",
+ GNUNET_i2s (pi), tid, ntohl (msg->port));
t = tunnel_get (pi, tid);
if (NULL == t) /* might be a local tunnel */
{
t->reliable = GNUNET_YES;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " nobuffer:%d\n", t->nobuffer);
-
- tunnel_reset_timeout (t);
}
- t->state = MESH_TUNNEL_WAITING;
+ tunnel_reset_timeout (t, GNUNET_YES);
+ tunnel_change_state (t, MESH_TUNNEL_WAITING);
dest_peer_info =
- GNUNET_CONTAINER_multihashmap_get (peers, &pi[size - 1].hashPubKey);
+ GNUNET_CONTAINER_multipeermap_get (peers, &pi[size - 1]);
if (NULL == dest_peer_info)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" Creating PeerInfo for destination.\n");
- dest_peer_info = GNUNET_malloc (sizeof (struct MeshPeerInfo));
+ dest_peer_info = GNUNET_malloc (sizeof (struct MeshPeer));
dest_peer_info->id = GNUNET_PEER_intern (&pi[size - 1]);
- GNUNET_CONTAINER_multihashmap_put (peers, &pi[size - 1].hashPubKey,
+ GNUNET_CONTAINER_multipeermap_put (peers, &pi[size - 1],
dest_peer_info,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
}
- orig_peer_info = GNUNET_CONTAINER_multihashmap_get (peers, &pi->hashPubKey);
+ orig_peer_info = GNUNET_CONTAINER_multipeermap_get (peers, pi);
if (NULL == orig_peer_info)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" Creating PeerInfo for origin.\n");
- orig_peer_info = GNUNET_malloc (sizeof (struct MeshPeerInfo));
+ orig_peer_info = GNUNET_new (struct MeshPeer);
orig_peer_info->id = GNUNET_PEER_intern (pi);
- GNUNET_CONTAINER_multihashmap_put (peers, &pi->hashPubKey, orig_peer_info,
+ GNUNET_CONTAINER_multipeermap_put (peers, pi, orig_peer_info,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Creating path...\n");
path_add_to_peers (path, GNUNET_NO);
tunnel_use_path (t, path);
- peer_info_add_tunnel (dest_peer_info, t);
+ peer_add_tunnel (dest_peer_info, t);
if (own_pos == size - 1)
{
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " It's for us!\n");
- peer_info_add_path_to_origin (orig_peer_info, path, GNUNET_YES);
+ peer_add_path_to_origin (orig_peer_info, path, GNUNET_YES);
+ /* This can be a retransmission due to a lost PATH ACK.
+ * Check if we already have a destination client for the tunnel. */
+ if (t->client != c)
+ {
+ /* Assign local tid */
+ while (NULL != tunnel_get_incoming (next_local_tid))
+ next_local_tid = (next_local_tid + 1) | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
+ t->local_tid_dest = next_local_tid++;
+ next_local_tid = next_local_tid | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
- /* Assign local tid */
- while (NULL != tunnel_get_incoming (next_local_tid))
- next_local_tid = (next_local_tid + 1) | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
- t->local_tid_dest = next_local_tid++;
- next_local_tid = next_local_tid | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
+ if (GNUNET_YES == t->reliable)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! Reliable\n");
+ t->bck_rel = GNUNET_malloc (sizeof (struct MeshTunnelReliability));
+ t->bck_rel->t = t;
+ t->bck_rel->expected_delay = MESH_RETRANSMIT_TIME;
+ }
- if (GNUNET_YES == t->reliable)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! Reliable\n");
- t->bck_rel = GNUNET_malloc (sizeof (struct MeshTunnelReliability));
- t->bck_rel->t = t;
- t->bck_rel->expected_delay = MESH_RETRANSMIT_TIME;
+ tunnel_add_client (t, c);
+ send_local_tunnel_create (t);
}
-
- tunnel_add_client (t, c);
- send_client_tunnel_create (t);
send_path_ack (t);
+ /* Eliminate tunnel when origin dies */
+ tunnel_reset_timeout (t, GNUNET_YES);
+ /* Keep tunnel alive in direction dest->owner*/
+ tunnel_reset_timeout (t, GNUNET_NO);
}
else
{
/* It's for somebody else! Retransmit. */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Retransmitting.\n");
path2 = path_duplicate (path);
- peer_info_add_path (dest_peer_info, path2, GNUNET_NO);
- peer_info_add_path_to_origin (orig_peer_info, path, GNUNET_NO);
- send_create_path (t);
+ peer_add_path (dest_peer_info, path2, GNUNET_NO);
+ peer_add_path_to_origin (orig_peer_info, path, GNUNET_NO);
+ send_path_create (t);
}
return GNUNET_OK;
}
const struct GNUNET_MessageHeader *message)
{
struct GNUNET_MESH_PathACK *msg;
- struct MeshPeerInfo *peer_info;
+ struct MeshPeer *peer_info;
struct MeshPeerPath *p;
struct MeshTunnel *t;
{
GNUNET_break (0);
}
- t->state = MESH_TUNNEL_READY;
+ tunnel_change_state (t, MESH_TUNNEL_READY);
+ tunnel_reset_timeout (t, GNUNET_NO);
t->next_fc.last_ack_recv = (NULL == t->client) ? ntohl (msg->ack) : 0;
t->prev_fc.last_ack_sent = ntohl (msg->ack);
GNUNET_DHT_get_stop (peer_info->dhtget);
peer_info->dhtget = NULL;
}
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_PATH_ACK);
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_PATH_ACK);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_PATH_ACK, GNUNET_YES);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_PATH_ACK, GNUNET_NO);
return GNUNET_OK;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" not for us, retransmitting...\n");
- peer_info = peer_get (&msg->oid);
send_prebuilt_message (message, t->prev_hop, t);
return GNUNET_OK;
}
/**
- * Core handler for mesh network traffic going from the origin to a peer
+ * Generic handler for mesh network payload traffic.
+ *
+ * @param peer Peer identity this notification is about.
+ * @param message Data message.
+ * @param fwd Is this FWD traffic? GNUNET_YES : GNUNET_NO;
*
- * @param cls closure
- * @param peer peer identity this notification is about
- * @param message message
* @return GNUNET_OK to keep the connection open,
* GNUNET_SYSERR to close it (signal serious error)
*/
static int
-handle_mesh_unicast (void *cls, const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+handle_mesh_data (const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message,
+ int fwd)
{
struct GNUNET_MESH_Data *msg;
+ struct MeshFlowControl *fc;
+ struct MeshTunnelReliability *rel;
struct MeshTunnel *t;
+ struct MeshClient *c;
+ GNUNET_PEER_Id hop;
uint32_t pid;
uint32_t ttl;
+ uint16_t type;
size_t size;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a unicast packet from %s\n",
- GNUNET_i2s (peer));
/* Check size */
size = ntohs (message->size);
if (size <
GNUNET_break (0);
return GNUNET_OK;
}
+ type =ntohs (message->type);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a %s message from %s\n",
+ GNUNET_MESH_DEBUG_M2S (type), GNUNET_i2s (peer));
msg = (struct GNUNET_MESH_Data *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %s\n",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " payload of type %s\n",
GNUNET_MESH_DEBUG_M2S (ntohs (msg[1].header.type)));
/* Check tunnel */
t = tunnel_get (&msg->oid, ntohl (msg->tid));
{
/* TODO notify back: we don't know this tunnel */
GNUNET_STATISTICS_update (stats, "# data on unknown tunnel", 1, GNUNET_NO);
- GNUNET_break_op (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "WARNING tunnel unknown\n");
return GNUNET_OK;
}
+
+ /* Initialize FWD/BCK data */
pid = ntohl (msg->pid);
- if (GMC_is_pid_bigger (pid, t->prev_fc.last_ack_sent))
+ fc = fwd ? &t->prev_fc : &t->next_fc;
+ c = fwd ? t->client : t->owner;
+ rel = fwd ? t->bck_rel : t->fwd_rel;
+ hop = fwd ? t->next_hop : t->prev_hop;
+ if (GMC_is_pid_bigger (pid, fc->last_ack_sent))
{
- GNUNET_STATISTICS_update (stats, "# unsolicited unicast", 1, GNUNET_NO);
- GNUNET_break_op (0);
+ GNUNET_STATISTICS_update (stats, "# unsolicited data", 1, GNUNET_NO);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received PID %u, ACK %u\n",
- pid, t->prev_fc.last_ack_sent);
- tunnel_send_fwd_ack(t, GNUNET_MESSAGE_TYPE_MESH_POLL);
+ "WARNING Received PID %u, (prev %u), ACK %u\n",
+ pid, fc->last_pid_recv, fc->last_ack_sent);
return GNUNET_OK;
}
-
- tunnel_reset_timeout (t);
- if (t->dest == myid)
+ if (NULL != c)
+ tunnel_change_state (t, MESH_TUNNEL_READY);
+ tunnel_reset_timeout (t, fwd);
+ if (NULL != c)
{
/* TODO signature verification */
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " it's for us! sending to clients...\n");
- GNUNET_STATISTICS_update (stats, "# unicast received", 1, GNUNET_NO);
-// if (GMC_is_pid_bigger(pid, t->prev_fc.last_pid_recv)) FIXME use
- if ( (GNUNET_NO == t->reliable &&
- GMC_is_pid_bigger(pid, t->prev_fc.last_pid_recv))
- ||
- (GNUNET_YES == t->reliable &&
- pid == t->prev_fc.last_pid_recv + 1) )
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " it's for us! sending to client\n");
+ GNUNET_STATISTICS_update (stats, "# data received", 1, GNUNET_NO);
+ if (GMC_is_pid_bigger (pid, fc->last_pid_recv))
{
+ uint32_t mid;
+
+ mid = ntohl (msg->mid);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " pid %u not seen yet, forwarding\n", pid);
- t->prev_fc.last_pid_recv = pid;
- tunnel_send_client_ucast (t, msg);
+ " pid %u (mid %u) not seen yet\n", pid, mid);
+ fc->last_pid_recv = pid;
+
+ if (GNUNET_NO == t->reliable ||
+ ( !GMC_is_pid_bigger (rel->mid_recv, mid) &&
+ GMC_is_pid_bigger (rel->mid_recv + 64, mid) ) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "!!! RECV %u\n", ntohl (msg->mid));
+ if (GNUNET_YES == t->reliable)
+ {
+ /* Is this the exact next expected messasge? */
+ if (mid == rel->mid_recv)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "as expected\n");
+ rel->mid_recv++;
+ tunnel_send_client_data (t, msg, fwd);
+ tunnel_send_client_buffered_data (t, c, rel);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "save for later\n");
+ tunnel_add_buffered_data (t, msg, rel);
+ }
+ }
+ else /* Tunnel unreliable, send to clients directly */
+ {
+ tunnel_send_client_data (t, msg, fwd);
+ }
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " MID %u not expected (%u - %u), dropping!\n",
+ ntohl (msg->mid), rel->mid_recv, rel->mid_recv + 64);
+ }
}
else
{
// GNUNET_STATISTICS_update (stats, "# duplicate PID", 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- " Pid %u not expected (%u), sending FWD ACK!\n",
- pid, t->prev_fc.last_pid_recv + 1);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " Pid %u not expected (%u+), dropping!\n",
+ pid, fc->last_pid_recv + 1);
}
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK);
+ tunnel_send_ack (t, type, fwd);
return GNUNET_OK;
}
- if (GMC_is_pid_bigger(pid, t->prev_fc.last_pid_recv))
- t->prev_fc.last_pid_recv = pid;
- if (0 == t->next_hop)
+ fc->last_pid_recv = pid;
+ if (0 == hop)
{
- GNUNET_break (0);
- return GNUNET_OK;
+ GNUNET_STATISTICS_update (stats, "# data on dying tunnel", 1, GNUNET_NO);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "data on dying tunnel %s[%X]\n",
+ GNUNET_PEER_resolve2 (t->id.oid), ntohl (msg->tid));
+ return GNUNET_OK; /* Next hop has destoyed the tunnel, drop */
}
ttl = ntohl (msg->ttl);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ttl: %u\n", ttl);
{
GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK, fwd);
return GNUNET_OK;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " not for us, retransmitting...\n");
- send_prebuilt_message (message, t->next_hop, t);
- GNUNET_STATISTICS_update (stats, "# unicast forwarded", 1, GNUNET_NO);
+ if (myid != hop)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " not for us, retransmitting...\n");
+ send_prebuilt_message (message, hop, t);
+ GNUNET_STATISTICS_update (stats, "# unicast forwarded", 1, GNUNET_NO);
+ }
return GNUNET_OK;
}
/**
- * Core handler for mesh network traffic toward the owner of a tunnel
+ * Core handler for mesh network traffic going from the origin to a peer
*
- * @param cls closure
- * @param message message
- * @param peer peer identity this notification is about
+ * @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)
*/
static int
-handle_mesh_to_orig (void *cls, const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+handle_mesh_unicast (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
{
- struct GNUNET_MESH_Data *msg;
- struct MeshTunnel *t;
- size_t size;
- uint32_t pid;
- uint32_t ttl;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a ToOrigin packet from %s\n",
- GNUNET_i2s (peer));
- size = ntohs (message->size);
- if (size < sizeof (struct GNUNET_MESH_Data) + /* Payload must be */
- sizeof (struct GNUNET_MessageHeader)) /* at least a header */
- {
- GNUNET_break_op (0);
- return GNUNET_OK;
- }
- msg = (struct GNUNET_MESH_Data *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %s\n",
- GNUNET_MESH_DEBUG_M2S (ntohs (msg[1].header.type)));
- t = tunnel_get (&msg->oid, ntohl (msg->tid));
- pid = ntohl (msg->pid);
- if (NULL == t)
- {
- /* TODO notify that we dont know this tunnel (whom)? */
- GNUNET_STATISTICS_update (stats, "# data on unknown tunnel", 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received to_origin with PID %u on unknown tunnel %s [%u]\n",
- pid, GNUNET_i2s (&msg->oid), ntohl (msg->tid));
- return GNUNET_OK;
- }
-
- if (GMC_is_pid_bigger (pid, t->next_fc.last_ack_sent))
- {
- GNUNET_STATISTICS_update (stats, "# unsolicited to_orig", 1, GNUNET_NO);
- GNUNET_break_op (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received PID %u, ACK %u\n",
- pid, t->next_fc.last_ack_sent);
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_POLL);
- return GNUNET_OK;
- }
-
- if (myid == t->id.oid)
- {
- /* TODO signature verification */
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " it's for us! sending to clients...\n");
- GNUNET_STATISTICS_update (stats, "# to origin received", 1, GNUNET_NO);
- if ( (GNUNET_NO == t->reliable &&
- GMC_is_pid_bigger(pid, t->next_fc.last_pid_recv))
- ||
- (GNUNET_YES == t->reliable &&
- pid == t->next_fc.last_pid_recv + 1) ) // FIXME use "futures" as accepting
- {
- t->next_fc.last_pid_recv = pid;
- tunnel_send_client_to_orig (t, msg);
- }
- else
- {
-// GNUNET_STATISTICS_update (stats, "# duplicate PID drops BCK", 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- " Pid %u not expected, sending FWD ACK!\n", pid);
- }
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK);
- return GNUNET_OK;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " not for us, retransmitting...\n");
- if (GMC_is_pid_bigger (pid, t->next_fc.last_pid_recv))
- t->next_fc.last_pid_recv = pid;
- if (0 == t->prev_hop) /* No owner AND no prev hop */
- {
- if (GNUNET_YES == t->destroy)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "to orig received on a dying tunnel %s [%X]\n",
- GNUNET_i2s (&msg->oid), ntohl(msg->tid));
- return GNUNET_OK;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "unknown to origin at %s\n",
- GNUNET_i2s (&my_full_id));
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "from peer %s\n",
- GNUNET_i2s (peer));
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "on tunnel %s [%X]\n",
- GNUNET_i2s (&msg->oid), ntohl(msg->tid));
- return GNUNET_OK;
- }
- ttl = ntohl (msg->ttl);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ttl: %u\n", ttl);
- if (ttl == 0)
- {
- GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
- return GNUNET_OK;
- }
- send_prebuilt_message (message, t->prev_hop, t);
- GNUNET_STATISTICS_update (stats, "# to origin forwarded", 1, GNUNET_NO);
+ return handle_mesh_data (peer, message, GNUNET_YES);
+}
- return GNUNET_OK;
+/**
+ * Core handler for mesh network traffic towards the owner of a tunnel.
+ *
+ * @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)
+ */
+static int
+handle_mesh_to_orig (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ return handle_mesh_data (peer, message, GNUNET_NO);
}
GNUNET_STATISTICS_update (stats, "# ack on unknown tunnel", 1, GNUNET_NO);
return GNUNET_OK;
}
- ack = ntohl (msg->pid);
+ ack = ntohl (msg->mid);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
/* Is this a forward or backward ACK? */
return GNUNET_OK;
}
rel = t->fwd_rel;
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
}
else if (t->prev_hop == id && GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK == type)
{
return GNUNET_OK;
}
rel = t->bck_rel;
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
}
else
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! ACK %u\n", ack);
for (work = GNUNET_NO, copy = rel->head_sent; copy != NULL; copy = next)
{
- struct GNUNET_TIME_Relative time;
-
- if (copy->id > ack)
+ if (GMC_is_pid_bigger (copy->mid, ack))
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! head %u, out!\n", copy->id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! head %u, out!\n", copy->mid);
+ tunnel_free_sent_reliable (t, msg, rel);
break;
}
work = GNUNET_YES;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! id %u\n", copy->id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! id %u\n", copy->mid);
next = copy->next;
- time = GNUNET_TIME_absolute_get_duration (copy->timestamp);
- rel->expected_delay.rel_value += time.rel_value;
- rel->expected_delay.rel_value /= 2;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! new expected delay %s\n",
- GNUNET_STRINGS_relative_time_to_string (rel->expected_delay,
- GNUNET_NO));
- rel->retry_timer = rel->expected_delay;
- GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
- GNUNET_free (copy);
+ tunnel_free_reliable_message (copy);
}
+ /* Once buffers have been free'd, send ACK */
+ tunnel_send_ack (t, type, GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK == type);
+ /* If some message was free'd, update the retransmission delay*/
if (GNUNET_YES == work)
{
if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
struct GNUNET_TIME_Absolute new_target;
struct GNUNET_TIME_Relative delay;
+ delay = GNUNET_TIME_relative_multiply (rel->retry_timer,
+ MESH_RETRANSMIT_MARGIN);
new_target = GNUNET_TIME_absolute_add (rel->head_sent->timestamp,
- rel->retry_timer);
+ delay);
delay = GNUNET_TIME_absolute_get_remaining (new_target);
rel->retry_task =
GNUNET_SCHEDULER_add_delayed (delay,
id = GNUNET_PEER_search (peer);
if (t->next_hop == id)
{
- debug_fwd_ack++;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK\n");
fc = &t->next_fc;
}
else if (t->prev_hop == id)
{
- debug_bck_ack++;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
fc = &t->prev_fc;
}
fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
}
+
fc->last_ack_recv = ack;
peer_unlock_queue (id);
+ tunnel_change_state (t, MESH_TUNNEL_READY);
- if (t->next_hop == id)
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
- else
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK, t->next_hop == id);
return GNUNET_OK;
}
uint32_t pid;
uint32_t old;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got an POLL packet from %s!\n",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got a POLL packet from %s!\n",
GNUNET_i2s (peer));
msg = (struct GNUNET_MESH_Poll *) message;
}
/* Is this a forward or backward ACK? */
- id = GNUNET_PEER_search(peer);
+ id = GNUNET_PEER_search (peer);
pid = ntohl (msg->pid);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " PID %u\n", pid);
if (t->next_hop == id)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " from FWD\n");
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " was %u\n", t->next_fc.last_pid_recv);
fc = &t->next_fc;
old = fc->last_pid_recv;
- fc->last_pid_recv = pid;
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_POLL);
}
else if (t->prev_hop == id)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " from BCK\n");
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " was %u\n", t->prev_fc.last_pid_recv);
fc = &t->prev_fc;
old = fc->last_pid_recv;
- fc->last_pid_recv = pid;
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_POLL);
}
else
+ {
GNUNET_break (0);
-
+ return GNUNET_OK;
+ }
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " was %u\n", fc->last_pid_recv);
+ fc->last_pid_recv = pid;
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_POLL, t->prev_hop == id);
+
if (GNUNET_YES == t->reliable)
fc->last_pid_recv = old;
{
struct GNUNET_MESH_TunnelKeepAlive *msg;
struct MeshTunnel *t;
+ struct MeshClient *c;
+ GNUNET_PEER_Id hop;
+ int fwd;
+ msg = (struct GNUNET_MESH_TunnelKeepAlive *) message;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a keepalive packet from %s\n",
GNUNET_i2s (peer));
- msg = (struct GNUNET_MESH_TunnelKeepAlive *) message;
t = tunnel_get (&msg->oid, ntohl (msg->tid));
-
if (NULL == t)
{
/* TODO notify that we dont know that tunnel */
return GNUNET_OK;
}
- tunnel_reset_timeout (t);
- if (NULL != t->client || 0 == t->next_hop || myid == t->next_hop)
+ fwd = GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE == ntohs (message->type) ?
+ GNUNET_YES : GNUNET_NO;
+ c = fwd ? t->client : t->owner;
+ hop = fwd ? t->next_hop : t->prev_hop;
+
+ if (NULL != c)
+ tunnel_change_state (t, MESH_TUNNEL_READY);
+ tunnel_reset_timeout (t, fwd);
+ if (NULL != c || 0 == hop || myid == hop)
return GNUNET_OK;
GNUNET_STATISTICS_update (stats, "# keepalives forwarded", 1, GNUNET_NO);
- send_prebuilt_message (message, t->next_hop, t);
+ send_prebuilt_message (message, hop, t);
return GNUNET_OK;
}
sizeof (struct GNUNET_MESH_DataACK)},
{&handle_mesh_data_ack, GNUNET_MESSAGE_TYPE_MESH_TO_ORIG_ACK,
sizeof (struct GNUNET_MESH_DataACK)},
- {&handle_mesh_keepalive, GNUNET_MESSAGE_TYPE_MESH_PATH_KEEPALIVE,
+ {&handle_mesh_keepalive, GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE,
+ sizeof (struct GNUNET_MESH_TunnelKeepAlive)},
+ {&handle_mesh_keepalive, GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE,
sizeof (struct GNUNET_MESH_TunnelKeepAlive)},
{&handle_mesh_ack, GNUNET_MESSAGE_TYPE_MESH_ACK,
sizeof (struct GNUNET_MESH_ACK)},
};
-
-/******************************************************************************/
-/**************** MESH LOCAL HANDLER HELPERS ***********************/
-/******************************************************************************/
-
-
-#if LATER
-/**
- * notify_client_connection_failure: notify a client that the connection to the
- * requested remote peer is not possible (for instance, no route found)
- * Function called when the socket is ready to queue more data. "buf" will be
- * NULL and "size" zero if the socket was closed for writing in the meantime.
- *
- * @param cls closure
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-notify_client_connection_failure (void *cls, size_t size, void *buf)
-{
- int size_needed;
- struct MeshPeerInfo *peer_info;
- struct GNUNET_MESH_PeerControl *msg;
- struct GNUNET_PeerIdentity id;
-
- if (0 == size && NULL == buf)
- {
- // TODO retry? cancel?
- return 0;
- }
-
- size_needed = sizeof (struct GNUNET_MESH_PeerControl);
- peer_info = (struct MeshPeerInfo *) cls;
- msg = (struct GNUNET_MESH_PeerControl *) buf;
- msg->header.size = htons (sizeof (struct GNUNET_MESH_PeerControl));
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_DISCONNECTED);
-// msg->tunnel_id = htonl(peer_info->t->tid);
- GNUNET_PEER_resolve (peer_info->id, &id);
- memcpy (&msg->peer, &id, sizeof (struct GNUNET_PeerIdentity));
-
- return size_needed;
-}
-#endif
-
-
-/**
- * Send keepalive packets for a tunnel.
- *
- * @param cls Closure (tunnel for which to send the keepalive).
- * @param tc Notification context.
- *
- * FIXME: add a refresh reset in case of normal unicast traffic is doing the job
- */
-static void
-path_refresh (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- struct MeshTunnel *t = cls;
- struct GNUNET_MESH_TunnelKeepAlive *msg;
- size_t size = sizeof (struct GNUNET_MESH_TunnelKeepAlive);
- char cbuf[size];
-
- t->maintenance_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) ||
- NULL == t->owner || 0 == t->local_tid)
- {
- return;
- }
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "sending keepalive for tunnel %d\n", t->id.tid);
-
- msg = (struct GNUNET_MESH_TunnelKeepAlive *) cbuf;
- msg->header.size = htons (size);
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_KEEPALIVE);
- msg->oid = my_full_id;
- msg->tid = htonl (t->id.tid);
- send_prebuilt_message (&msg->header, t->next_hop, t);
-
- t->maintenance_task =
- GNUNET_SCHEDULER_add_delayed (refresh_path_time, &path_refresh, t);
-}
-
-
/**
* Function to process paths received for a new peer addition. The recorded
* paths form the initial tunnel, which can be optimized later.
unsigned int put_path_length, enum GNUNET_BLOCK_Type type,
size_t size, const void *data)
{
- struct MeshPeerInfo *peer = cls;
+ struct MeshPeer *peer = cls;
struct MeshPeerPath *p;
struct GNUNET_PeerIdentity pi;
int i;
GNUNET_PEER_resolve (peer->tunnels[i]->id.oid, &id);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ... tunnel %s:%X (%X / %X)\n",
GNUNET_i2s (&id), peer->tunnels[i]->id.tid,
- peer->tunnels[i]->local_tid,
+ peer->tunnels[i]->local_tid,
peer->tunnels[i]->local_tid_dest);
if (peer->tunnels[i]->state == MESH_TUNNEL_SEARCHING)
{
/******************************************************************************/
+/**
+ * Handler for client connection.
+ *
+ * @param cls Closure (unused).
+ * @param client Client handler.
+ */
+static void
+handle_local_client_connect (void *cls, struct GNUNET_SERVER_Client *client)
+{
+ struct MeshClient *c;
+
+ if (NULL == client)
+ return;
+ c = GNUNET_malloc (sizeof (struct MeshClient));
+ c->handle = client;
+ GNUNET_SERVER_client_keep (client);
+ GNUNET_SERVER_client_set_user_context (client, c);
+ GNUNET_CONTAINER_DLL_insert (clients_head, clients_tail, c);
+}
+
+
/**
* Handler for client disconnection
*
handle_local_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
{
struct MeshClient *c;
- struct MeshClient *next;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client disconnected: %p\n", client);
if (client == NULL)
return;
}
- c = clients_head;
- while (NULL != c)
+ c = client_get (client);
+ if (NULL != c)
{
- if (c->handle != client)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " ... searching %p (%u)\n",
- c->handle, c->id);
- c = c->next;
- continue;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "matching client found (%u)\n",
- c->id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "matching client found (%u, %p)\n",
+ c->id, c);
GNUNET_SERVER_client_drop (c->handle);
c->shutting_down = GNUNET_YES;
- GNUNET_CONTAINER_multihashmap32_iterate (c->own_tunnels,
- &tunnel_destroy_iterator, c);
- GNUNET_CONTAINER_multihashmap32_iterate (c->incoming_tunnels,
- &tunnel_destroy_iterator, c);
- GNUNET_CONTAINER_multihashmap32_destroy (c->own_tunnels);
- GNUNET_CONTAINER_multihashmap32_destroy (c->incoming_tunnels);
+ if (NULL != c->own_tunnels)
+ {
+ GNUNET_CONTAINER_multihashmap32_iterate (c->own_tunnels,
+ &tunnel_destroy_iterator, c);
+ GNUNET_CONTAINER_multihashmap32_destroy (c->own_tunnels);
+ }
+
+ if (NULL != c->incoming_tunnels)
+ {
+ GNUNET_CONTAINER_multihashmap32_iterate (c->incoming_tunnels,
+ &tunnel_destroy_iterator, c);
+ GNUNET_CONTAINER_multihashmap32_destroy (c->incoming_tunnels);
+ }
if (NULL != c->ports)
+ {
+ GNUNET_CONTAINER_multihashmap32_iterate (c->ports,
+ &client_release_ports, c);
GNUNET_CONTAINER_multihashmap32_destroy (c->ports);
- next = c->next;
+ }
GNUNET_CONTAINER_DLL_remove (clients_head, clients_tail, c);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " CLIENT FREE at %p\n", c);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " client free (%p)\n", c);
GNUNET_free (c);
GNUNET_STATISTICS_update (stats, "# clients", -1, GNUNET_NO);
- c = next;
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " context NULL!\n");
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "done!\n");
return;
}
size /= sizeof (uint32_t);
- /* Create new client structure */
- c = GNUNET_malloc (sizeof (struct MeshClient));
+ /* Initialize new client structure */
+ c = GNUNET_SERVER_client_get_user_context (client, struct MeshClient);
c->id = next_client_id++; /* overflow not important: just for debug */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " client id %u\n", c->id);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " client has %u ports\n", size);
- c->handle = client;
- GNUNET_SERVER_client_keep (client);
if (size > 0)
{
uint32_t u32;
}
}
- GNUNET_CONTAINER_DLL_insert (clients_head, clients_tail, c);
c->own_tunnels = GNUNET_CONTAINER_multihashmap32_create (32);
c->incoming_tunnels = GNUNET_CONTAINER_multihashmap32_create (32);
GNUNET_SERVER_notification_context_add (nc, client);
const struct GNUNET_MessageHeader *message)
{
struct GNUNET_MESH_TunnelMessage *t_msg;
- struct MeshPeerInfo *peer_info;
+ struct MeshPeer *peer_info;
struct MeshTunnel *t;
struct MeshClient *c;
MESH_TunnelNumber tid;
}
t_msg = (struct GNUNET_MESH_TunnelMessage *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " towards %s\n",
- GNUNET_i2s (&t_msg->peer));
- /* Sanity check for tunnel numbering */
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " towards %s:%u\n",
+ GNUNET_i2s (&t_msg->peer), ntohl (t_msg->port));
tid = ntohl (t_msg->tunnel_id);
- if (0 == (tid & GNUNET_MESH_LOCAL_TUNNEL_ID_CLI))
- {
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
- }
/* Sanity check for duplicate tunnel IDs */
if (NULL != tunnel_get_by_local_id (c, tid))
{
return;
}
t->port = ntohl (t_msg->port);
- tunnel_set_options (t, ntohl (t_msg->options));
+ tunnel_set_options (t, ntohl (t_msg->opt));
if (GNUNET_YES == t->reliable)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! Reliable\n");
GNUNET_i2s (&my_full_id), t->id.tid, t->port, t->local_tid);
peer_info = peer_get (&t_msg->peer);
- peer_info_add_tunnel (peer_info, t);
+ peer_add_tunnel (peer_info, t);
peer_connect (peer_info, t);
+ tunnel_reset_timeout (t, GNUNET_YES);
GNUNET_SERVER_receive_done (client, GNUNET_OK);
return;
}
/* Retrieve tunnel */
tid = ntohl (tunnel_msg->tunnel_id);
- t = tunnel_get_by_local_id(c, tid);
+ t = tunnel_get_by_local_id (c, tid);
if (NULL == t)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, " tunnel %X not found\n", tid);
/* Cleanup after the tunnel */
client_delete_tunnel (c, t);
- if (c == t->client)
+ if (c == t->client && GNUNET_MESH_LOCAL_TUNNEL_ID_SERV <= tid)
{
t->client = NULL;
}
- if (c == t->owner)
+ else if (c == t->owner && GNUNET_MESH_LOCAL_TUNNEL_ID_SERV > tid)
{
- peer_info_remove_tunnel (peer_get_short (t->dest), t);
+ peer_remove_tunnel (peer_get_short (t->dest), t);
t->owner = NULL;
}
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ " tunnel %X client %p (%p, %p)\n",
+ tid, c, t->owner, t->client);
+ GNUNET_break (0);
+ }
/* The tunnel will be destroyed when the last message is transmitted. */
tunnel_destroy_empty (t);
/* Tunnel exists? */
tid = ntohl (data_msg->tid);
- if (tid < GNUNET_MESH_LOCAL_TUNNEL_ID_CLI)
- {
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
- }
t = tunnel_get_by_local_id (c, tid);
if (NULL == t)
{
t->owner->handle == client)
||
(tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV &&
- t->client &&
+ t->client &&
t->client->handle == client) ) )
{
GNUNET_break (0);
struct MeshTunnelReliability *rel;
struct MeshReliableMessage *copy;
+ rel = (tid < GNUNET_MESH_LOCAL_TUNNEL_ID_SERV) ? t->fwd_rel : t->bck_rel;
copy = GNUNET_malloc (sizeof (struct MeshReliableMessage)
+ sizeof(struct GNUNET_MESH_Data)
+ size);
- copy->id = fc->last_pid_recv + 1;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! DATA %u\n", copy->id);
+ copy->mid = rel->mid_sent++;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! DATA %u\n", copy->mid);
copy->timestamp = GNUNET_TIME_absolute_get ();
- rel = (tid < GNUNET_MESH_LOCAL_TUNNEL_ID_SERV) ? t->fwd_rel : t->bck_rel;
copy->rel = rel;
+ rel->n_sent++;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " n_sent %u\n", rel->n_sent);
GNUNET_CONTAINER_DLL_insert_tail (rel->head_sent, rel->tail_sent, copy);
if (GNUNET_SCHEDULER_NO_TASK == rel->retry_task)
{
- rel->retry_timer = rel->expected_delay;
+ rel->retry_timer =
+ GNUNET_TIME_relative_multiply (rel->expected_delay,
+ MESH_RETRANSMIT_MARGIN);
rel->retry_task =
GNUNET_SCHEDULER_add_delayed (rel->retry_timer,
&tunnel_retransmit_message,
rel);
}
payload = (struct GNUNET_MESH_Data *) ©[1];
+ payload->mid = htonl (copy->mid);
}
else
{
payload = (struct GNUNET_MESH_Data *) cbuf;
+ payload->mid = htonl (fc->last_pid_recv + 1);
}
memcpy (&payload[1], &data_msg[1], size);
payload->header.size = htons (sizeof (struct GNUNET_MESH_Data) + size);
}
/* Does client own tunnel? I.E: Is this an ACK for BCK traffic? */
- if (t->owner == c)
+ if (tid < GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
{
/* The client owns the tunnel, ACK is for data to_origin, send BCK ACK. */
t->prev_fc.last_ack_recv++;
- tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK, GNUNET_NO);
}
else
{
/* The client doesn't own the tunnel, this ACK is for FWD traffic. */
t->next_fc.last_ack_recv++;
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
+ tunnel_send_ack (t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK, GNUNET_YES);
}
GNUNET_SERVER_receive_done (client, GNUNET_OK);
static void
core_connect (void *cls, const struct GNUNET_PeerIdentity *peer)
{
- struct MeshPeerInfo *peer_info;
+ struct MeshPeer *peer_info;
struct MeshPeerPath *path;
DEBUG_CONN ("Peer connected\n");
}
path->peers[0] = myid;
GNUNET_PEER_change_rc (myid, 1);
- peer_info_add_path (peer_info, path, GNUNET_YES);
+ peer_add_path (peer_info, path, GNUNET_YES);
return;
}
static void
core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
{
- struct MeshPeerInfo *pi;
+ struct MeshPeer *pi;
struct MeshPeerQueue *q;
struct MeshPeerQueue *n;
DEBUG_CONN ("Peer disconnected\n");
- pi = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
+ pi = GNUNET_CONTAINER_multipeermap_get (peers, peer);
if (NULL == pi)
{
GNUNET_break (0);
GNUNET_CORE_notify_transmit_ready_cancel(pi->core_transmit);
pi->core_transmit = NULL;
}
- peer_remove_path (pi, pi->id, myid);
+ peer_remove_path (pi, pi->id, myid);
if (myid == pi->id)
{
DEBUG_CONN (" (self)\n");
server_init (void)
{
GNUNET_SERVER_add_handlers (server_handle, client_handlers);
+ GNUNET_SERVER_connect_notify (server_handle,
+ &handle_local_client_connect, NULL);
GNUNET_SERVER_disconnect_notify (server_handle,
&handle_local_client_disconnect, NULL);
nc = GNUNET_SERVER_notification_context_create (server_handle, 1);
* To be called on core init/fail.
*
* @param cls Closure (config)
- * @param server handle to the server for this service
* @param identity the public identity of this peer
*/
static void
-core_init (void *cls, struct GNUNET_CORE_Handle *server,
+core_init (void *cls,
const struct GNUNET_PeerIdentity *identity)
{
const struct GNUNET_CONFIGURATION_Handle *c = cls;
static int i = 0;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core init\n");
- GNUNET_break (core_handle == server);
- if (0 != memcmp (identity, &my_full_id, sizeof (my_full_id)) ||
- NULL == server)
+ if (0 != memcmp (identity, &my_full_id, sizeof (my_full_id)))
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Wrong CORE service\n"));
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
* GNUNET_NO if not.
*/
static int
-shutdown_peer (void *cls, const struct GNUNET_HashCode * key, void *value)
+shutdown_peer (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
{
- struct MeshPeerInfo *p = value;
+ struct MeshPeer *p = value;
struct MeshPeerQueue *q;
struct MeshPeerQueue *n;
}
q = n;
}
- peer_info_destroy (p);
+ peer_destroy (p);
return GNUNET_YES;
}
core_handle = NULL;
}
GNUNET_CONTAINER_multihashmap_iterate (tunnels, &shutdown_tunnel, NULL);
- GNUNET_CONTAINER_multihashmap_iterate (peers, &shutdown_peer, NULL);
+ GNUNET_CONTAINER_multipeermap_iterate (peers, &shutdown_peer, NULL);
if (dht_handle != NULL)
{
GNUNET_DHT_disconnect (dht_handle);
const struct GNUNET_CONFIGURATION_Handle *c)
{
char *keyfile;
- struct GNUNET_CRYPTO_EccPrivateKey *pk;
+ struct GNUNET_CRYPTO_EddsaPrivateKey *pk;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "starting to run\n");
server_handle = server;
tunnels = GNUNET_CONTAINER_multihashmap_create (32, GNUNET_NO);
incoming_tunnels = GNUNET_CONTAINER_multihashmap32_create (32);
- peers = GNUNET_CONTAINER_multihashmap_create (32, GNUNET_NO);
+ peers = GNUNET_CONTAINER_multipeermap_create (32, GNUNET_NO);
ports = GNUNET_CONTAINER_multihashmap32_create (32);
dht_handle = GNUNET_DHT_connect (c, 64);
/* Scheduled the task to clean up when shutdown is called */
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
NULL);
- pk = GNUNET_CRYPTO_ecc_key_create_from_file (keyfile);
+ pk = GNUNET_CRYPTO_eddsa_key_create_from_file (keyfile);
GNUNET_free (keyfile);
GNUNET_assert (NULL != pk);
my_private_key = pk;
- GNUNET_CRYPTO_ecc_key_get_public (my_private_key, &my_public_key);
- GNUNET_CRYPTO_hash (&my_public_key, sizeof (my_public_key),
- &my_full_id.hashPubKey);
+ GNUNET_CRYPTO_eddsa_key_get_public (my_private_key,
+ &my_full_id.public_key);
myid = GNUNET_PEER_intern (&my_full_id);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Mesh for peer [%s] starting\n",
INTERVAL_SHOW;
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Mesh for peer [%s] FWD ACKs %u, BCK ACKs %u\n",
- GNUNET_i2s(&my_full_id), debug_fwd_ack, debug_bck_ack);
-
return ret;
}
projects / oweals / gnunet.git / blobdiff