* - error reporting (CREATE/CHANGE/ADD/DEL?) -- new message!
* - partial disconnect reporting -- same as error reporting?
* - add vs create? change vs. keep-alive? same msg or different ones? -- thinking...
- * - speed requirement specification (change?) in mesh API -- API call
* - add ping message
* - relay corking down to core
* - set ttl relative to tree depth
+ * - Add data ACK count in path ACK
+ * - Make common GNUNET_MESH_Data header for unicast, to_orig, multicast
* TODO END
*/
#define MESH_BLOOM_SIZE 128
-#define MESH_DEBUG_DHT GNUNET_YES
+#define MESH_DEBUG_DHT GNUNET_NO
#define MESH_DEBUG_CONNECTION GNUNET_NO
+#define MESH_DEBUG_TIMING __LINUX__ && GNUNET_YES
#if MESH_DEBUG_CONNECTION
#define DEBUG_CONN(...) GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
#define DEBUG_DHT(...)
#endif
+#if MESH_DEBUG_TIMING
+#include <time.h>
+double __sum;
+uint64_t __count;
+struct timespec __mesh_start;
+struct timespec __mesh_end;
+#define INTERVAL_START clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &(__mesh_start))
+#define INTERVAL_END \
+do {\
+ clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &(__mesh_end));\
+ double __diff = __mesh_end.tv_nsec - __mesh_start.tv_nsec;\
+ if (__diff < 0) __diff += 1000000000;\
+ __sum += __diff;\
+ __count++;\
+} while (0)
+#define INTERVAL_SHOW \
+if (0 < __count)\
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO, "AVG process time: %f ns\n", __sum/__count)
+#else
+#define INTERVAL_START
+#define INTERVAL_END
+#define INTERVAL_SHOW
+#endif
+
/******************************************************************************/
/************************ DATA STRUCTURES ****************************/
/******************************************************************************/
/** Tunnel it belongs to. */
struct MeshTunnel *t;
- /** In case of a multicast, task to allow a client to send more data if
- * some neighbor is too slow. */
- GNUNET_SCHEDULER_TaskIdentifier *task;
-
/** How many remaining neighbors we need to send this to. */
- unsigned int *reference_counter;
+ unsigned int reference_counter;
/** Size of the data. */
size_t data_len;
uint32_t bck_ack;
/**
- * How many messages are in the queue.
+ * How many messages are in the forward queue (towards leaves).
+ */
+ unsigned int fwd_queue_n;
+
+ /**
+ * How many messages do we accept in the forward queue.
*/
- unsigned int queue_n;
+ unsigned int fwd_queue_max;
/**
- * How many messages do we accept in the queue.
+ * How many messages are in the backward queue (towards origin).
*/
- unsigned int queue_max;
+ unsigned int bck_queue_n;
+
+ /**
+ * How many messages do we accept in the backward queue.
+ */
+ unsigned int bck_queue_max;
/**
* Last time the tunnel was used
struct MeshClient **clients;
/**
- * FWD ACK value of each active client: up to what message can we transmit
- * to a leaf client.
+ * Flow control info for each client.
*/
- uint32_t *clients_acks;
+ struct MeshTunnelClientInfo *clients_fc;
/**
- * Number of elements in clients/clients_acks
+ * Number of elements in clients/clients_fc
*/
unsigned int nclients;
/**
* Maximum PID allowed (FWD ACK received).
*/
- uint32_t max_pid;
+ uint32_t fwd_ack;
/**
* Last ACK sent to that child (BCK ACK).
*/
- uint32_t last_ack;
+ uint32_t bck_ack;
+
+ /**
+ * Circular buffer pointing to MeshPeerQueue elements.
+ * Size determined by the tunnel queue size.
+ */
+ struct MeshPeerQueue **send_buffer;
+
+ /**
+ * Index of the oldest element in the send_buffer.
+ */
+ unsigned int send_buffer_start;
+
+ /**
+ * How many elements are already in the buffer.
+ */
+ unsigned int send_buffer_n;
+};
+
+
+/**
+ * Info about a leaf client of a tunnel, needed to perform flow control.
+ */
+struct MeshTunnelClientInfo
+{
+ /**
+ * PID of the last packet sent to the client (FWD).
+ */
+ uint32_t fwd_pid;
+
+ /**
+ * PID of the last packet received from the client (BCK).
+ */
+ uint32_t bck_pid;
+
+ /**
+ * Maximum PID allowed (FWD ACK received).
+ */
+ uint32_t fwd_ack;
+
+ /**
+ * Last ACK sent to that child (BCK ACK).
+ */
+ uint32_t bck_ack;
};
+
/**
* Info collected during iteration of child nodes in order to get the ACK value
* for a tunnel.
*/
struct MeshTunnel *t;
+ /**
+ * Is this context initialized? Is the value in max_child_ack valid?
+ */
+ int init;
+
/**
* Maximum child ACK so far.
*/
}
#endif
+unsigned int debug_fwd_ack;
+unsigned int debug_bck_ack;
+
#endif
/******************************************************************************/
/*********************** GLOBAL VARIABLES ****************************/
/******************************************************************************/
-
/**
* Configuration parameters
*/
static uint32_t
tunnel_get_fwd_ack (struct MeshTunnel *t);
-/**
- * Get the current ack value for a tunnel, for data going from leaves to root,
- * taking in account the tunnel mode and the status of all children and clients.
- *
- * @param t Tunnel.
- *
- * @return Maximum PID allowed.
- */
-static uint32_t
-tunnel_get_bck_ack (struct MeshTunnel *t);
/**
* Add a client to a tunnel, initializing all needed data structures.
queue_add (void *cls, uint16_t type, size_t size,
struct MeshPeerInfo *dst, struct MeshTunnel *t);
+
/**
* Free a transmission that was already queued with all resources
* associated to the request.
static void
queue_destroy (struct MeshPeerQueue *queue, int clear_cls);
+
+/**
+ * @brief Get the next transmittable message from the queue.
+ *
+ * This will be the head, except in the case of being a data packet
+ * not allowed by the destination peer.
+ *
+ * @param peer Destination peer.
+ *
+ * @return The next viable MeshPeerQueue element to send to that peer.
+ * NULL when there are no transmittable messages.
+ */
+struct MeshPeerQueue *
+queue_get_next (const struct MeshPeerInfo *peer);
+
+
+/**
+ * Core callback to write a queued packet to core buffer
+ *
+ * @param cls Closure (peer info).
+ * @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
+queue_send (void *cls, size_t size, void *buf);
+
/******************************************************************************/
/************************ ITERATORS ****************************/
/******************************************************************************/
block.id = my_full_id;
c = GNUNET_CONTAINER_multihashmap_get (applications, key);
+ GNUNET_assert(NULL != c);
block.type = (long) GNUNET_CONTAINER_multihashmap_get (c->apps, key);
if (0 == block.type)
{
/****************** GENERAL HELPER FUNCTIONS ************************/
/******************************************************************************/
-/**
- * Check if one pid is bigger than other, accounting for overflow.
- *
- * @param bigger Argument that should be bigger.
- * @param smaller Argument that should be smaller.
- *
- * @return True if bigger (arg1) has a higher value than smaller (arg 2).
- */
-static int
-is_pid_bigger (uint32_t bigger, uint32_t smaller)
-{
- return (GNUNET_YES == PID_OVERFLOW(smaller, bigger) ||
- (bigger > smaller && GNUNET_NO == PID_OVERFLOW(bigger, smaller)));
-}
-
-/**
- * Get the higher ACK value out of two values, taking in account overflow.
- *
- * @param a First ACK value.
- * @param b Second ACK value.
- *
- * @return Highest ACK value from the two.
- */
-static uint32_t
-max_pid (uint32_t a, uint32_t b)
-{
- if (is_pid_bigger(a, b))
- return a;
- return b;
-}
-
-
-/**
- * Get the lower ACK value out of two values, taking in account overflow.
- *
- * @param a First ACK value.
- * @param b Second ACK value.
- *
- * @return Lowest ACK value from the two.
- */
-static uint32_t
-min_pid (uint32_t a, uint32_t b)
-{
- if (is_pid_bigger(a, b))
- return b;
- return a;
-}
-
-
/**
* Decrements the reference counter and frees all resources if needed
*
static void
data_descriptor_decrement_rc (struct MeshData *mesh_data)
{
- /* Make sure it's a multicast packet */
- GNUNET_assert (NULL != mesh_data->reference_counter);
-
- if (0 == --(*(mesh_data->reference_counter)))
+ if (0 == --(mesh_data->reference_counter))
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Last copy!\n");
- if (NULL != mesh_data->task)
- {
- if (GNUNET_SCHEDULER_NO_TASK != *(mesh_data->task))
- {
- GNUNET_SCHEDULER_cancel (*(mesh_data->task));
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " notifying client...\n");
- GNUNET_SERVER_receive_done (mesh_data->t->owner->handle, GNUNET_OK);
- }
- GNUNET_free (mesh_data->task);
- }
- GNUNET_free (mesh_data->reference_counter);
GNUNET_free (mesh_data->data);
GNUNET_free (mesh_data);
}
* @param c Client to check
*
* @return GNUNET_YES or GNUNET_NO, depending on subscription status
+ *
+ * FIXME: use of crypto_hash slows it down
+ * The hash function alone takes 8-10us out of the ~55us for the whole
+ * process of retransmitting the message from one local client to another.
+ * Find faster implementation!
*/
static int
client_is_subscribed (uint16_t message_type, struct MeshClient *c)
if (NULL == c->types)
return GNUNET_NO;
+
GNUNET_CRYPTO_hash (&message_type, sizeof (uint16_t), &hc);
return GNUNET_CONTAINER_multihashmap_contains (c->types, &hc);
}
-/**
- * Allow a client to send more data after transmitting a multicast message
- * which some neighbor has not yet accepted altough a reasonable time has
- * passed.
- *
- * @param cls Closure (DataDescriptor containing the task identifier)
- * @param tc Task Context
- *
- * FIXME reference counter cshould be just int
- */
-static void
-client_allow_send (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
- struct MeshData *mdata = cls;
-
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
- GNUNET_assert (NULL != mdata->reference_counter);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "CLIENT ALLOW SEND DESPITE %u COPIES PENDING\n",
- *(mdata->reference_counter));
- *(mdata->task) = GNUNET_SCHEDULER_NO_TASK;
- GNUNET_SERVER_receive_done (mdata->t->owner->handle, GNUNET_OK);
-}
-
-
/**
* Check whether client wants traffic from a tunnel.
*
*
* @param msg Pointer to the message itself
* @param payload Pointer to the payload of the message.
+ * @param t The tunnel to whose clients this message goes.
+ *
* @return number of clients this message was sent to
*/
static unsigned int
send_subscribed_clients (const struct GNUNET_MessageHeader *msg,
- const struct GNUNET_MessageHeader *payload)
+ const struct GNUNET_MessageHeader *payload,
+ struct MeshTunnel *t)
{
- struct GNUNET_PeerIdentity *oid;
struct MeshClient *c;
- struct MeshTunnel *t;
MESH_TunnelNumber *tid;
unsigned int count;
uint16_t type;
type = ntohs (payload->type);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending to clients...\n");
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "message of type %u\n", type);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "message of type %s\n",
+ GNUNET_MESH_DEBUG_M2S (type));
memcpy (cbuf, msg, sizeof (cbuf));
switch (htons (msg->type))
struct GNUNET_MESH_Multicast *mc;
struct GNUNET_MESH_ToOrigin *to;
- case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
- uc = (struct GNUNET_MESH_Unicast *) cbuf;
- tid = &uc->tid;
- oid = &uc->oid;
- break;
- case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
- mc = (struct GNUNET_MESH_Multicast *) cbuf;
- tid = &mc->tid;
- oid = &mc->oid;
- break;
- case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
- to = (struct GNUNET_MESH_ToOrigin *) cbuf;
- tid = &to->tid;
- oid = &to->oid;
- break;
- default:
- GNUNET_break (0);
- return 0;
- }
- t = tunnel_get (oid, ntohl (*tid));
- if (NULL == t)
- {
- GNUNET_break (0);
- return 0;
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ uc = (struct GNUNET_MESH_Unicast *) cbuf;
+ tid = &uc->tid;
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+ mc = (struct GNUNET_MESH_Multicast *) cbuf;
+ tid = &mc->tid;
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ to = (struct GNUNET_MESH_ToOrigin *) cbuf;
+ tid = &to->tid;
+ break;
+ default:
+ GNUNET_break (0);
+ return 0;
}
for (count = 0, c = clients; c != NULL; c = c->next)
*) cbuf, GNUNET_NO);
}
}
+
return count;
}
}
-/**
- * Notify a the client of a tunnel about how many more
- * payload packages will we accept on a given tunnel,
- * distinguiching between root and leaf clients.
- *
- * @param c Client whom to send the ACK.
- * @param t Tunnel on which to send the ACK.
- */
-static void
-send_client_tunnel_ack (struct MeshClient *c, struct MeshTunnel *t)
-{
- struct GNUNET_MESH_LocalAck msg;
- MESH_TunnelNumber tid;
- uint32_t ack;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Sending client ACK on tunnel %X\n",
- t->local_tid);
- if (NULL == c)
- return;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " to client %u\n", c->id);
-
- if (c == t->owner)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " (owner, FWD ACK)\n");
- ack = tunnel_get_fwd_ack (t);
- tid = t->local_tid;
- }
- else
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " (leaf, BCK ACK)\n");
- ack = tunnel_get_bck_ack (t);
- tid = t->local_tid_dest;
- }
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ack);
- if (t->last_fwd_ack == ack)
- return;
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending!\n");
- t->last_fwd_ack = ack;
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
- msg.tunnel_id = htonl (tid);
- msg.max_pid = htonl (ack);
-
- GNUNET_SERVER_notification_context_unicast (nc, c->handle,
- &msg.header, GNUNET_NO);
-}
-
-
/**
* Notify all clients (not depending on registration status) that the incoming
* tunnel is no longer valid.
struct MeshPeerInfo *neighbor;
struct MeshPeerPath *p;
size_t size;
+ uint16_t type;
// GNUNET_TRANSPORT_try_connect(); FIXME use?
info->mesh_data = GNUNET_malloc (sizeof (struct MeshData));
info->mesh_data->data = GNUNET_malloc (size);
memcpy (info->mesh_data->data, message, size);
- if (ntohs(message->type) == GNUNET_MESSAGE_TYPE_MESH_UNICAST)
+ type = ntohs(message->type);
+ if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == type)
{
struct GNUNET_MESH_Unicast *m;
m->ttl = htonl (ntohl (m->ttl) - 1);
}
info->mesh_data->data_len = size;
- info->mesh_data->reference_counter = GNUNET_malloc (sizeof (unsigned int));
- *info->mesh_data->reference_counter = 1;
+ info->mesh_data->reference_counter = 1;
neighbor = peer_info_get (peer);
for (p = neighbor->path_head; NULL != p; p = p->next)
{
}
if (NULL == p)
{
+#if MESH_DEBUG
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " %s IS NOT DIRECTLY CONNECTED\n",
+ GNUNET_i2s(peer));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " PATHS TO %s:\n",
+ GNUNET_i2s(peer));
+ for (p = neighbor->path_head; NULL != p; p = p->next)
+ {
+ struct GNUNET_PeerIdentity debug_id;
+ unsigned int i;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " path with %u hops through:\n",
+ p->length);
+ for (i = 0; i < p->length; i++)
+ {
+ GNUNET_PEER_resolve(p->peers[i], &debug_id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " hop %u: %s\n",
+ i, GNUNET_i2s(&debug_id));
+ }
+ }
+#endif
GNUNET_break (0); // FIXME sometimes fails (testing disconnect?)
GNUNET_free (info->mesh_data->data);
GNUNET_free (info->mesh_data);
return;
}
info->peer = neighbor;
+ if (GNUNET_MESSAGE_TYPE_MESH_PATH_ACK == type)
+ {
+ /*
+ * TODO: in this case we only need the service to retransmit
+ * the message down the path. If we pass the real type to queue_add,
+ * queue_send will try to build the message from scratch. This can
+ * probably be done by some other way instead of deleteing the type
+ * info.
+ */
+ type = 0;
+ }
queue_add (info,
- 0,
+ type,
size,
neighbor,
t);
if (t->clients[i] == c)
{
t->clients[i] = t->clients[t->nclients - 1];
- t->clients_acks[i] = t->clients_acks[t->nclients - 1];
+ t->clients_fc[i] = t->clients_fc[t->nclients - 1];
GNUNET_array_grow (t->clients, t->nclients, t->nclients - 1);
t->nclients++;
- GNUNET_array_grow (t->clients_acks, t->nclients, t->nclients - 1);
+ GNUNET_array_grow (t->clients_fc, t->nclients, t->nclients - 1);
break;
}
}
const struct GNUNET_HashCode * key,
void *value)
{
- GNUNET_free (value);
+ struct MeshTunnelChildInfo *cinfo = value;
+ struct MeshTunnel *t = cls;
+ unsigned int c;
+ unsigned int i;
+
+ for (c = 0; c < cinfo->send_buffer_n; c++)
+ {
+ i = (cinfo->send_buffer_start + c) % t->fwd_queue_max;
+ if (NULL != cinfo->send_buffer[i])
+ queue_destroy(cinfo->send_buffer[i], GNUNET_YES);
+ else
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%u %u\n", c, cinfo->send_buffer_n);
+ }
+ GNUNET_free_non_null (cinfo->send_buffer);
+ GNUNET_free (cinfo);
return GNUNET_YES;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "tunnel_add_path END\n");
}
+/**
+ * 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.
+ */
+static void
+tunnel_add_client (struct MeshTunnel *t, struct MeshClient *c)
+{
+ struct MeshTunnelClientInfo clinfo;
+
+ GNUNET_array_append (t->clients, t->nclients, c);
+ clinfo.fwd_ack = t->fwd_pid + 1;
+ clinfo.bck_ack = t->nobuffer ? 1 : INITIAL_WINDOW_SIZE - 1;
+ clinfo.fwd_pid = t->fwd_pid;
+ clinfo.bck_pid = (uint32_t) -1; // Expected next: 0
+ t->nclients--;
+ GNUNET_array_append (t->clients_fc, t->nclients, clinfo);
+}
+
/**
* Notifies a tunnel that a connection has broken that affects at least
info = GNUNET_malloc (sizeof (struct MeshTransmissionDescriptor));
info->mesh_data = mdata;
- (*(mdata->reference_counter)) ++;
+ (mdata->reference_counter) ++;
info->destination = neighbor_id;
GNUNET_PEER_resolve (neighbor_id, &neighbor);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending to %s...\n",
/**
- * Send a message in a tunnel in multicast, sending a copy to each child node
+ * Queue a message in a tunnel in multicast, sending a copy to each child node
* down the local one in the tunnel tree.
*
* @param t Tunnel in which to send the data.
* @param msg Message to be sent.
- * @param internal Has the service generated this message?
+ * @param internal DEPRECATED Has the service generated this message?
+ *
+ * FIXME remove internal if no use comes up
*/
static void
tunnel_send_multicast (struct MeshTunnel *t,
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" sending a multicast packet...\n");
+
mdata = GNUNET_malloc (sizeof (struct MeshData));
mdata->data_len = ntohs (msg->size);
- mdata->reference_counter = GNUNET_malloc (sizeof (unsigned int));
mdata->t = t;
mdata->data = GNUNET_malloc (mdata->data_len);
memcpy (mdata->data, msg, mdata->data_len);
struct GNUNET_MESH_Multicast *mcast;
mcast = (struct GNUNET_MESH_Multicast *) mdata->data;
+ if (t->fwd_queue_n >= t->fwd_queue_max)
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR, " queue full!\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ " message from %s!\n",
+ GNUNET_i2s(&mcast->oid));
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ " message at %s!\n",
+ GNUNET_i2s(&my_full_id));
+ GNUNET_free (mdata->data);
+ GNUNET_free (mdata);
+ return;
+ }
+ t->fwd_queue_n++;
mcast->ttl = htonl (ntohl (mcast->ttl) - 1);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " data packet, ttl: %u\n",
ntohl (mcast->ttl));
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " not a data packet, no ttl\n");
}
- if (NULL != t->owner && GNUNET_YES != t->owner->shutting_down
- && GNUNET_NO == internal)
- {
- mdata->task = GNUNET_malloc (sizeof (GNUNET_SCHEDULER_TaskIdentifier));
- (*(mdata->task)) =
- GNUNET_SCHEDULER_add_delayed (unacknowledged_wait_time, &client_allow_send,
- mdata);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "timeout task %u\n",
- *(mdata->task));
- }
tree_iterate_children (t->tree, &tunnel_send_multicast_iterator, mdata);
- if (*(mdata->reference_counter) == 0)
+ if (mdata->reference_counter == 0)
{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " no one to send data to\n");
GNUNET_free (mdata->data);
- GNUNET_free (mdata->reference_counter);
- if (NULL != mdata->task)
- {
- GNUNET_SCHEDULER_cancel(*(mdata->task));
- GNUNET_free (mdata->task);
- GNUNET_SERVER_receive_done (t->owner->handle, GNUNET_OK);
- }
- // FIXME change order?
GNUNET_free (mdata);
+ t->fwd_queue_n--;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" sending a multicast packet done\n");
&peer->hashPubKey);
if (NULL == cinfo)
{
+ uint32_t delta;
+
cinfo = GNUNET_malloc (sizeof (struct MeshTunnelChildInfo));
cinfo->id = GNUNET_PEER_intern (peer);
cinfo->skip = t->fwd_pid;
- cinfo->max_pid = t->fwd_pid + t->queue_max - t->queue_n; // FIXME review
+
+ delta = t->nobuffer ? 1 : INITIAL_WINDOW_SIZE - 1;
+ cinfo->fwd_ack = t->fwd_pid + delta;
+ cinfo->bck_ack = delta;
+
+ cinfo->send_buffer =
+ GNUNET_malloc (sizeof(struct MeshPeerQueue *) * t->fwd_queue_max);
GNUNET_assert (GNUNET_OK ==
GNUNET_CONTAINER_multihashmap_put (t->children_fc,
}
+/**
+ * Get the Flow Control info of a client.
+ *
+ * @param t Tunnel on which to look.
+ * @param c Client whose ACK to get.
+ *
+ * @return ACK value.
+ */
+static struct MeshTunnelClientInfo *
+tunnel_get_client_fc (struct MeshTunnel *t,
+ struct MeshClient *c)
+{
+ unsigned int i;
+
+ for (i = 0; i < t->nclients; i++)
+ {
+ if (t->clients[i] != c)
+ continue;
+ return &t->clients_fc[i];
+ }
+ GNUNET_assert (0);
+ return NULL; // avoid compiler / coverity complaints
+}
+
+
/**
* Iterator to get the appropiate ACK value from all children nodes.
*
* @param id Id of the child node.
*/
static void
-tunnel_get_child_ack (void *cls,
- GNUNET_PEER_Id id)
+tunnel_get_child_fwd_ack (void *cls,
+ GNUNET_PEER_Id id)
{
struct GNUNET_PeerIdentity peer_id;
struct MeshTunnelChildInfo *cinfo;
GNUNET_PEER_resolve (id, &peer_id);
cinfo = tunnel_get_neighbor_fc (t, &peer_id);
- ack = cinfo->max_pid;
+ ack = cinfo->fwd_ack;
- if (0 == ctx->max_child_ack)
+ ctx->nchildren++;
+ if (GNUNET_NO == ctx->init)
+ {
ctx->max_child_ack = ack;
+ ctx->init = GNUNET_YES;
+ }
if (GNUNET_YES == t->speed_min)
{
*
* @param t Tunnel.
*
- * @return Maximum PID allowed (uint32 MAX), -1 if node has no children.
+ * @return Maximum PID allowed (uint32 MAX), -1LL if node has no children.
*/
static int64_t
-tunnel_get_children_ack (struct MeshTunnel *t)
+tunnel_get_children_fwd_ack (struct MeshTunnel *t)
{
struct MeshTunnelChildIteratorContext ctx;
ctx.t = t;
ctx.max_child_ack = 0;
ctx.nchildren = 0;
- tree_iterate_children (t->tree, tunnel_get_child_ack, &ctx);
+ tree_iterate_children (t->tree, tunnel_get_child_fwd_ack, &ctx);
if (0 == ctx.nchildren)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " tunnel has no children, no FWD ACK\n");
return -1LL;
+ }
- return (int64_t) ctx.max_child_ack;
-}
-
-
-/**
- * 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.
- */
-static void
-tunnel_add_client (struct MeshTunnel *t, struct MeshClient *c)
-{
- uint32_t ack;
+ if (GNUNET_YES == t->nobuffer && GMC_is_pid_bigger(ctx.max_child_ack, t->fwd_pid))
+ ctx.max_child_ack = t->fwd_pid + 1; // Might overflow, it's ok.
- GNUNET_array_append (t->clients, t->nclients, c);
- t->nclients--;
- ack = t->fwd_pid + 1;
- GNUNET_array_append (t->clients_acks, t->nclients, ack);
+ return (int64_t) ctx.max_child_ack;
}
/**
- * Set the ACK value of a client in a particular tunnel.
+ * Set the FWD ACK value of a client in a particular tunnel.
*
* @param t Tunnel affected.
* @param c Client whose ACK to set.
{
if (t->clients[i] != c)
continue;
- t->clients_acks[i] = ack;
+ t->clients_fc[i].fwd_ack = ack;
return;
}
GNUNET_break (0);
}
-/**
- * Get the ACK value of a client in a particular tunnel.
- *
- * @param t Tunnel on which to look.
- * @param c Client whose ACK to get.
- *
- * @return ACK value.
- */
-uint32_t // FIXME static when used!!
-tunnel_get_client_ack (struct MeshTunnel *t,
- struct MeshClient *c)
-{
- unsigned int i;
-
- for (i = 0; i < t->nclients; i++)
- {
- if (t->clients[i] != c)
- continue;
- return t->clients_acks[i];
- }
- GNUNET_break (0);
- return t->clients_acks[0];
-}
-
-
/**
* Get the highest ACK value of all clients in a particular tunnel,
* according to the buffering/speed settings.
* @param t Tunnel on which to look.
*
* @return Corresponding ACK value (max uint32_t).
- * If no clients are suscribed, -1.
+ * If no clients are suscribed, -1LL.
*/
static int64_t
-tunnel_get_clients_ack (struct MeshTunnel *t)
+tunnel_get_clients_fwd_ack (struct MeshTunnel *t)
{
unsigned int i;
int64_t ack;
- for (ack = -1, i = 0; i < t->nclients; i++)
+ if (0 == t->nclients)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " tunnel has no clients, no FWD ACK\n");
+ return -1LL;
+ }
+
+ for (ack = -1LL, i = 0; i < t->nclients; i++)
{
- if (-1 == ack ||
- (GNUNET_YES == t->speed_min && t->clients_acks[i] < ack) ||
- (GNUNET_NO == t->speed_min && t->clients_acks[i] > ack))
+ if (-1LL == ack ||
+ (GNUNET_YES == t->speed_min &&
+ GNUNET_YES == GMC_is_pid_bigger (ack, t->clients_fc[i].fwd_ack)) ||
+ (GNUNET_NO == t->speed_min &&
+ GNUNET_YES == GMC_is_pid_bigger (t->clients_fc[i].fwd_ack, ack)))
{
- ack = t->clients_acks[i];
+ ack = t->clients_fc[i].fwd_ack;
}
}
- if (GNUNET_YES == t->nobuffer && ack > t->fwd_pid)
- ack = t->fwd_pid + 1;
+ if (GNUNET_YES == t->nobuffer && GMC_is_pid_bigger(ack, t->fwd_pid))
+ ack = (uint32_t) t->fwd_pid + 1; // Might overflow, it's ok.
return (uint32_t) ack;
}
/**
- * Get the current ack value for a tunnel, taking in account the tunnel
+ * Get the current fwd ack value for a tunnel, taking in account the tunnel
* mode and the status of all children nodes.
*
* @param t Tunnel.
static uint32_t
tunnel_get_fwd_ack (struct MeshTunnel *t)
{
+ uint32_t ack;
uint32_t count;
uint32_t buffer_free;
int64_t child_ack;
int64_t client_ack;
- uint32_t ack;
count = t->fwd_pid - t->skip;
- buffer_free = t->queue_max - t->queue_n;
- ack = count + buffer_free; // Might overflow 32bits, it's ok!
- child_ack = tunnel_get_children_ack (t);
- client_ack = tunnel_get_clients_ack (t);
- if (-1 == child_ack)
+ buffer_free = t->fwd_queue_max - t->fwd_queue_n;
+ ack = count + buffer_free; // Might overflow 32 bits, it's ok!
+ child_ack = tunnel_get_children_fwd_ack (t);
+ client_ack = tunnel_get_clients_fwd_ack (t);
+ if (-1LL == child_ack)
{
// Node has no children, child_ack AND core buffer are irrelevant.
- GNUNET_break (-1 != client_ack); // No children and no clients? Not good!
+ GNUNET_break (-1LL != client_ack); // No children AND no clients? Not good!
return (uint32_t) client_ack;
}
-
+ if (-1LL == client_ack)
+ {
+ client_ack = ack;
+ }
if (GNUNET_YES == t->speed_min)
{
- ack = min_pid (child_ack, ack);
- ack = min_pid (client_ack, ack);
+ ack = GMC_min_pid ((uint32_t) child_ack, ack);
+ ack = GMC_min_pid ((uint32_t) client_ack, ack);
}
else
{
- ack = max_pid (child_ack, ack);
- ack = max_pid (client_ack, ack);
+ ack = GMC_max_pid ((uint32_t) child_ack, ack);
+ ack = GMC_max_pid ((uint32_t) client_ack, ack);
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "c %u, bf %u, ch %u, cl %u\n",
- count, buffer_free, child_ack, client_ack);
+ if (GNUNET_YES == t->nobuffer && GMC_is_pid_bigger(ack, t->fwd_pid))
+ ack = t->fwd_pid + 1; // Might overflow 32 bits, it's ok!
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "c %u, bf %u, ch %lld, cl %lld, ACK: %u\n",
+ count, buffer_free, child_ack, client_ack, ack);
return ack;
}
+
/**
- * Get the current ack value for a tunnel, taking in account the tunnel
- * mode and the status of all children nodes.
- *
- * @param t Tunnel.
+ * 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 ack Value of the ACK.
+ */
+static void
+send_local_ack (struct MeshTunnel *t, struct MeshClient *c, uint32_t ack)
+{
+ 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 (t->owner == c ? t->local_tid : t->local_tid_dest);
+ msg.max_pid = htonl (ack);
+ 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, struct GNUNET_PeerIdentity *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_message (&msg.header, peer, t);
+}
+
+
+/**
+ * Notify a the owner of a tunnel about how many more
+ * payload packages will we accept on a given tunnel.
*
- * @return Maximum PID allowed.
+ * @param t Tunnel on which to send the ACK.
*/
-static uint32_t
-tunnel_get_bck_ack (struct MeshTunnel *t)
+static void
+tunnel_send_client_fwd_ack (struct MeshTunnel *t)
{
uint32_t ack;
- ack = 0;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending client FWD ACK on tunnel %X\n",
+ t->local_tid);
- return ack;
+ ack = tunnel_get_fwd_ack (t);
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ack);
+ if (t->last_fwd_ack == ack)
+ return;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " sending!\n");
+ t->last_fwd_ack = ack;
+ send_local_ack (t, t->owner, ack);
}
static void
tunnel_send_fwd_ack (struct MeshTunnel *t, uint16_t type)
{
- struct GNUNET_MESH_ACK msg;
struct GNUNET_PeerIdentity id;
uint32_t ack;
if (NULL != t->owner)
{
- send_client_tunnel_ack (t->owner, t);
+ tunnel_send_client_fwd_ack (t);
return;
}
/* Is it after unicast / multicast retransmission? */
- if (GNUNET_MESSAGE_TYPE_MESH_ACK != type)
+ switch (type)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ACK via DATA retransmission\n");
- if (GNUNET_YES == t->nobuffer)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, nobuffer\n");
- return;
- }
- if (t->queue_max > t->queue_n * 2)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer free\n");
- return;
- }
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+ 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;
+ }
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
+ break;
+ default:
+ GNUNET_break (0);
+ }
+
+ /* Check if we need no retransmit the ACK */
+ if (t->fwd_queue_max > t->fwd_queue_n * 2 &&
+ GMC_is_pid_bigger(t->last_fwd_ack, t->fwd_pid))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer free\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " t->qmax: %u, t->qn: %u\n",
+ t->fwd_queue_max, t->fwd_queue_n);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " t->pid: %u, t->ack: %u\n",
+ t->fwd_pid, t->last_fwd_ack);
+ return;
}
/* Ok, ACK might be necessary, what PID to ACK? */
/* If speed_min and not all children have ack'd, dont send yet */
if (ack == t->last_fwd_ack)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, not ready\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not ready\n");
return;
}
t->last_fwd_ack = ack;
- msg.pid = htonl (ack);
-
GNUNET_PEER_resolve (tree_get_predecessor (t->tree), &id);
+ send_ack (t, &id, ack);
+ debug_fwd_ack++;
+}
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
- msg.tid = htonl (t->id.tid);
- GNUNET_PEER_resolve(t->id.oid, &msg.oid);
- send_message (&msg.header, &id, t);
+
+/**
+ * Iterator to send a child node a BCK ACK to allow him to send more
+ * to_origin data.
+ *
+ * @param cls Closure (tunnel).
+ * @param id Id of the child node.
+ */
+static void
+tunnel_send_child_bck_ack (void *cls,
+ GNUNET_PEER_Id id)
+{
+ struct MeshTunnel *t = cls;
+ struct MeshTunnelChildInfo *cinfo;
+ struct GNUNET_PeerIdentity peer;
+
+ GNUNET_PEER_resolve (id, &peer);
+ cinfo = tunnel_get_neighbor_fc (t, &peer);
+
+ if (cinfo->bck_ack != cinfo->pid &&
+ GNUNET_NO == GMC_is_pid_bigger (cinfo->bck_ack, cinfo->pid))
+ return;
+
+ cinfo->bck_ack++;
+ send_ack (t, &peer, cinfo->bck_ack);
+}
+
+
+/**
+ * @brief Send BCK ACKs to clients to allow them more to_origin traffic
+ *
+ * Iterates over all clients and sends BCK ACKs to the ones that need it.
+ *
+ * @param t Tunnel on which to send the BCK ACKs.
+ */
+static void
+tunnel_send_clients_bck_ack (struct MeshTunnel *t)
+{
+ unsigned int i;
+ unsigned int tunnel_delta;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " Sending BCK ACK to clients\n");
+
+ tunnel_delta = t->bck_ack - t->bck_pid;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " tunnel delta: %u\n", tunnel_delta);
+
+ /* Find client whom to allow to send to origin (with lowest buffer space) */
+ for (i = 0; i < t->nclients; i++)
+ {
+ struct MeshTunnelClientInfo *clinfo;
+ unsigned int delta;
+
+ clinfo = &t->clients_fc[i];
+ delta = clinfo->bck_ack - clinfo->bck_pid;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " client %u delta: %u\n",
+ t->clients[i]->id, delta);
+
+ if ((GNUNET_NO == t->nobuffer && tunnel_delta > delta) ||
+ (GNUNET_YES == t->nobuffer && 0 == delta))
+ {
+ uint32_t ack;
+
+ ack = clinfo->bck_pid;
+ ack += t->nobuffer ? 1 : tunnel_delta;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " sending ack to client %u: %u\n",
+ t->clients[i]->id, ack);
+ send_local_ack (t, t->clients[i], ack);
+ clinfo->bck_ack = ack;
+ }
+ }
}
/**
- * Send an ACK informing the children nodes about the available buffer space.
- * In case there is no child node, inform the destination clients.
+ * Send an ACK informing the children nodes and destination clients 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*,
static void
tunnel_send_bck_ack (struct MeshTunnel *t, uint16_t type)
{
- struct GNUNET_MESH_ACK msg;
- struct GNUNET_PeerIdentity id;
- uint32_t ack;
-
- if (NULL != t->owner)
- {
- send_client_tunnel_ack (t->owner, t);
- return;
- }
- /* Is it after unicast / multicast retransmission? */
- if (GNUNET_MESSAGE_TYPE_MESH_ACK != type)
+ 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? */
+ switch (type)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ACK via DATA retransmission\n");
- if (GNUNET_YES == t->nobuffer)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, nobuffer\n");
- return;
- }
- if (t->queue_max > t->queue_n * 2)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer free\n");
- return;
- }
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ if (GNUNET_YES == t->nobuffer)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " Not sending ACK, nobuffer\n");
+ return;
+ }
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
+ break;
+ default:
+ GNUNET_break (0);
}
- /* Ok, ACK might be necessary, what PID to ACK? */
- ack = tunnel_get_bck_ack (t);
+ tunnel_send_clients_bck_ack (t);
+ tree_iterate_children (t->tree, &tunnel_send_child_bck_ack, t);
+}
+
+
+/**
+ * @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 cls Closure (unused)
+ * @param peer_id Short ID of peer to which initiate traffic.
+ */
+static void
+peer_unlock_queue(void *cls, GNUNET_PEER_Id peer_id)
+{
+ struct MeshPeerInfo *peer;
+ struct GNUNET_PeerIdentity id;
+ struct MeshPeerQueue *q;
+ size_t size;
+
+ peer = peer_info_get_short(peer_id);
+ if (NULL != peer->core_transmit)
+ return;
- /* If speed_min and not all children have ack'd, dont send yet */
- if (ack == t->last_fwd_ack)
+ q = queue_get_next(peer);
+ if (NULL == q)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, not ready\n");
+ /* 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;
}
+ 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;
+}
- t->last_fwd_ack = ack;
- msg.pid = htonl (ack);
- GNUNET_PEER_resolve (tree_get_predecessor (t->tree), &id);
+/**
+ * @brief Allow transmission of FWD traffic on this tunnel
+ *
+ * Check if there is traffic queued towards any children
+ * and the core transmit handle is NULL, and if so, call core tmt rdy.
+ *
+ * @param t Tunnel on which to unlock FWD traffic.
+ */
+static void
+tunnel_unlock_fwd_queues (struct MeshTunnel *t)
+{
+ if (0 == t->fwd_queue_n)
+ return;
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
- msg.tid = htonl (t->id.tid);
- GNUNET_PEER_resolve(t->id.oid, &msg.oid);
- send_message (&msg.header, &id, t);
+ tree_iterate_children (t->tree, &peer_unlock_queue, NULL);
+}
+
+
+/**
+ * @brief Allow transmission of BCK traffic on this tunnel
+ *
+ * Check if there is traffic queued towards the root of the tree
+ * and the core transmit handle is NULL, and if so, call core tmt rdy.
+ *
+ * @param t Tunnel on which to unlock BCK traffic.
+ */
+static void
+tunnel_unlock_bck_queue (struct MeshTunnel *t)
+{
+ if (0 == t->bck_queue_n)
+ return;
+
+ peer_unlock_queue(NULL, tree_get_predecessor(t->tree));
}
next = pq->next;
if (pq->tunnel == t)
{
+ if (GNUNET_MESSAGE_TYPE_MESH_MULTICAST == pq->type ||
+ 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);
}
}
if (NULL == t)
return GNUNET_OK;
- tree_iterate_children (t->tree, &tunnel_cancel_queues, t);
-
r = GNUNET_OK;
c = t->owner;
#if MESH_DEBUG
t);
GNUNET_CONTAINER_multihashmap_destroy (t->children_fc);
+ tree_iterate_children (t->tree, &tunnel_cancel_queues, t);
+
tree_destroy (t->tree);
if (NULL != t->regex_ctx)
{
struct MeshTunnel *t;
struct GNUNET_HashCode hash;
-
+
if (n_tunnels >= max_tunnels && NULL == client)
return NULL;
t = GNUNET_malloc (sizeof (struct MeshTunnel));
t->id.oid = owner;
t->id.tid = tid;
- t->queue_max = (max_msgs_queue / max_tunnels) + 1;
+ t->fwd_queue_max = (max_msgs_queue / max_tunnels) + 1;
+ t->bck_queue_max = t->fwd_queue_max;
t->tree = tree_new (owner);
t->owner = client;
- t->bck_ack = 1;
+ t->fwd_pid = (uint32_t) -1; // Next (expected) = 0
+ t->bck_pid = (uint32_t) -1; // Next (expected) = 0
+ t->bck_ack = INITIAL_WINDOW_SIZE - 1;
+ t->last_fwd_ack = INITIAL_WINDOW_SIZE - 1;
t->local_tid = local;
t->children_fc = GNUNET_CONTAINER_multihashmap_create (8);
n_tunnels++;
* Resets the tunnel timeout. Starts it if no timeout was running.
*
* @param t Tunnel whose timeout to reset.
+ *
+ * TODO use heap to improve efficiency of scheduler.
*/
static void
tunnel_reset_timeout (struct MeshTunnel *t)
msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE);
msg->tid = ntohl (t->id.tid);
+ opt = 0;
if (GNUNET_YES == t->speed_min)
- opt = MESH_TUNNEL_OPT_SPEED_MIN;
+ opt |= MESH_TUNNEL_OPT_SPEED_MIN;
if (GNUNET_YES == t->nobuffer)
opt |= MESH_TUNNEL_OPT_NOBUFFER;
msg->opt = htonl(opt);
mc = (struct GNUNET_MESH_Multicast *) mh;
mh = (struct GNUNET_MessageHeader *) &mc[1];
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " multicast, payload type %u\n", ntohs (mh->type));
+ " multicast, payload type %s\n",
+ GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" multicast, payload size %u\n", ntohs (mh->size));
}
else
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " type %u\n",
- ntohs (mh->type));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " type %s\n",
+ GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
}
}
#endif
break;
default:
GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " type unknown!\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ " type %u unknown!\n",
+ queue->type);
}
GNUNET_free_non_null (queue->cls);
}
}
+/**
+ * @brief Get the next transmittable message from the queue.
+ *
+ * This will be the head, except in the case of being a data packet
+ * not allowed by the destination peer.
+ *
+ * @param peer Destination peer.
+ *
+ * @return The next viable MeshPeerQueue element to send to that peer.
+ * NULL when there are no transmittable messages.
+ */
+struct MeshPeerQueue *
+queue_get_next (const struct MeshPeerInfo *peer)
+{
+ struct MeshPeerQueue *q;
+ struct MeshTunnel *t;
+ struct MeshTransmissionDescriptor *info;
+ struct MeshTunnelChildInfo *cinfo;
+ struct GNUNET_MESH_Unicast *ucast;
+ struct GNUNET_MESH_ToOrigin *to_orig;
+ struct GNUNET_MESH_Multicast *mcast;
+ struct GNUNET_PeerIdentity id;
+ uint32_t pid;
+ uint32_t ack;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* selecting message\n");
+ for (q = peer->queue_head; NULL != q; q = q->next)
+ {
+ t = q->tunnel;
+ info = q->cls;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* %s\n",
+ GNUNET_MESH_DEBUG_M2S(q->type));
+ switch (q->type)
+ {
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ ucast = (struct GNUNET_MESH_Unicast *) info->mesh_data->data;
+ pid = ntohl (ucast->pid);
+ GNUNET_PEER_resolve (info->peer->id, &id);
+ cinfo = tunnel_get_neighbor_fc(t, &id);
+ ack = cinfo->fwd_ack;
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ to_orig = (struct GNUNET_MESH_ToOrigin *) info->mesh_data->data;
+ pid = ntohl (to_orig->pid);
+ ack = t->bck_ack;
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+ mcast = (struct GNUNET_MESH_Multicast *) info->mesh_data->data;
+ if (GNUNET_MESSAGE_TYPE_MESH_MULTICAST != ntohs(mcast->header.type))
+ {
+ // Not a multicast payload: multicast control traffic (destroy, etc)
+ return q;
+ }
+ pid = ntohl (mcast->pid);
+ GNUNET_PEER_resolve (info->peer->id, &id);
+ cinfo = tunnel_get_neighbor_fc(t, &id);
+ ack = cinfo->fwd_ack;
+ break;
+ default:
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* OK!\n");
+ return q;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* ACK: %u, PID: %u\n",
+ ack, pid);
+ if (GNUNET_NO == GMC_is_pid_bigger(pid, ack))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* OK!\n");
+ return q;
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* NEXT!\n");
+ }
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* nothing found\n");
+ return NULL;
+}
+
+
/**
* Core callback to write a queued packet to core buffer
*
struct GNUNET_MessageHeader *msg;
struct MeshPeerQueue *queue;
struct MeshTunnel *t;
+ struct MeshTunnelChildInfo *cinfo;
+ struct GNUNET_PeerIdentity dst_id;
size_t data_size;
peer->core_transmit = NULL;
- queue = peer->queue_head;
-
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* Queue send\n");
+ queue = queue_get_next(peer);
- /* If queue is empty, send should have been cancelled */
+ /* Queue has no internal mesh traffic not sendable payload */
if (NULL == queue)
{
- GNUNET_break(0);
- return 0;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* not ready, return\n");
+ if (NULL == peer->queue_head)
+ GNUNET_break(0); // Should've been canceled
+ return 0;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* not empty\n");
+ GNUNET_PEER_resolve (peer->id, &dst_id);
/* Check if buffer size is enough for the message */
if (queue->size > size)
{
- struct GNUNET_PeerIdentity id;
- GNUNET_PEER_resolve (peer->id, &id);
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* not enough room, reissue\n");
peer->core_transmit =
GNUNET_CORE_notify_transmit_ready(core_handle,
0,
0,
GNUNET_TIME_UNIT_FOREVER_REL,
- &id,
+ &dst_id,
queue->size,
&queue_send,
peer);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* size ok\n");
t = queue->tunnel;
- t->queue_n--;
+ if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == queue->type)
+ {
+ t->fwd_queue_n--;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* unicast: t->q (%u/%u)\n",
+ t->fwd_queue_n, t->fwd_queue_max);
+ }
+ else if (GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == queue->type)
+ {
+ t->bck_queue_n--;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* to origin\n");
+ }
/* Fill buf */
switch (queue->type)
{
- case 0: // RAW data (preconstructed message, retransmission, etc)
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* raw\n");
- data_size = send_core_data_raw (queue->cls, size, buf);
- msg = (struct GNUNET_MessageHeader *) buf;
- switch (ntohs (msg->type)) // Type of payload
- {
- case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
- tunnel_send_bck_ack(t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
- break;
- default:
- break;
- }
- break;
- case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* multicast\n");
- data_size = send_core_data_multicast(queue->cls, size, buf);
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_MULTICAST);
- break;
- case GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* path create\n");
- data_size = send_core_path_create(queue->cls, size, buf);
+ case 0:
+ case GNUNET_MESSAGE_TYPE_MESH_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_PATH_BROKEN:
+ case GNUNET_MESSAGE_TYPE_MESH_PATH_DESTROY:
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* raw: %s\n",
+ GNUNET_MESH_DEBUG_M2S (queue->type));
+ /* Fall through */
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ data_size = send_core_data_raw (queue->cls, size, buf);
+ msg = (struct GNUNET_MessageHeader *) buf;
+ switch (ntohs (msg->type)) // Type of preconstructed message
+ {
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
break;
- case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* path ack\n");
- data_size = send_core_path_ack(queue->cls, size, buf);
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ tunnel_send_bck_ack(t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
break;
- default:
- GNUNET_break (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* type unknown\n");
- data_size = 0;
+ default:
+ break;
+ }
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* multicast\n");
+ {
+ struct MeshTransmissionDescriptor *info = queue->cls;
+ if (info->mesh_data->reference_counter == 1)
+ t->fwd_queue_n--;
+ // FIXME fc (t->fwd_queue_n--)
+ }
+ data_size = send_core_data_multicast(queue->cls, size, buf);
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_MULTICAST);
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE:
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* path create\n");
+ data_size = send_core_path_create(queue->cls, size, buf);
+ break;
+ case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* path ack\n");
+ data_size = send_core_path_ack(queue->cls, size, buf);
+ break;
+ default:
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "********* type unknown: %u\n",
+ queue->type);
+ data_size = 0;
+ }
+ switch (queue->type)
+ {
+ case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+ case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
+ case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+ cinfo = tunnel_get_neighbor_fc(t, &dst_id);
+ if (cinfo->send_buffer[cinfo->send_buffer_start] != queue)
+ {
+ GNUNET_break(0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "at pos %u (%p) != %p\n",
+ cinfo->send_buffer_start,
+ cinfo->send_buffer[cinfo->send_buffer_start],
+ queue);
+ }
+ GNUNET_break(cinfo->send_buffer_n > 0);
+ cinfo->send_buffer[cinfo->send_buffer_start] = NULL;
+ cinfo->send_buffer_n--;
+ cinfo->send_buffer_start++;
+ cinfo->send_buffer_start %= t->fwd_queue_max;
+ break;
+ default:
+ break;
}
/* Free queue, but cls was freed by send_core_* */
queue_destroy (queue, GNUNET_NO);
- if (GNUNET_YES == t->destroy && 0 == t->queue_n)
+ if (GNUNET_YES == t->destroy && 0 == t->fwd_queue_n)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* destroying tunnel!\n");
tunnel_destroy (t);
}
/* If more data in queue, send next */
- if (NULL != peer->queue_head)
+ queue = queue_get_next(peer);
+ if (NULL != queue)
{
struct GNUNET_PeerIdentity id;
0,
GNUNET_TIME_UNIT_FOREVER_REL,
&id,
- peer->queue_head->size,
+ queue->size,
&queue_send,
peer);
}
+ else
+ {
+ if (NULL != peer->queue_head)
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "********* %s stalled\n",
+ GNUNET_i2s(&my_full_id));
+ }
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* return %d\n", data_size);
return data_size;
}
queue_add (void *cls, uint16_t type, size_t size,
struct MeshPeerInfo *dst, struct MeshTunnel *t)
{
- struct MeshPeerQueue *queue;
+ struct MeshPeerQueue *queue;
+ struct MeshTunnelChildInfo *cinfo;
+ struct GNUNET_PeerIdentity id;
+ unsigned int *max;
+ unsigned int *n;
+ unsigned int i;
- if (t->queue_n >= t->queue_max)
+ n = NULL;
+ if (GNUNET_MESSAGE_TYPE_MESH_UNICAST == type ||
+ GNUNET_MESSAGE_TYPE_MESH_MULTICAST == type)
+ {
+ n = &t->fwd_queue_n;
+ max = &t->fwd_queue_max;
+ }
+ else if (GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN == type)
+ {
+ n = &t->bck_queue_n;
+ max = &t->bck_queue_max;
+ }
+ if (NULL != n) {
+ if (*n >= *max)
{
if (NULL == t->owner)
GNUNET_break_op(0); // TODO: kill connection?
else
GNUNET_break(0);
+ GNUNET_STATISTICS_update(stats, "# messages dropped (buffer full)",
+ 1, GNUNET_NO);
return; // Drop message
}
- t->queue_n++;
- queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
- queue->cls = cls;
- queue->type = type;
- queue->size = size;
- queue->peer = dst;
- queue->tunnel = t;
- GNUNET_CONTAINER_DLL_insert_tail (dst->queue_head, dst->queue_tail, queue);
- if (NULL == dst->core_transmit)
- {
- struct GNUNET_PeerIdentity id;
+ (*n)++;
+ }
+ queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
+ queue->cls = cls;
+ queue->type = type;
+ queue->size = size;
+ queue->peer = dst;
+ queue->tunnel = t;
+ GNUNET_CONTAINER_DLL_insert_tail (dst->queue_head, dst->queue_tail, queue);
+ GNUNET_PEER_resolve (dst->id, &id);
+ if (NULL == dst->core_transmit)
+ {
+ dst->core_transmit =
+ GNUNET_CORE_notify_transmit_ready(core_handle,
+ 0,
+ 0,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &id,
+ size,
+ &queue_send,
+ dst);
+ }
+ if (NULL == n) // Is this internal mesh traffic?
+ return;
- GNUNET_PEER_resolve (dst->id, &id);
- dst->core_transmit =
- GNUNET_CORE_notify_transmit_ready(core_handle,
- 0,
- 0,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &id,
- size,
- &queue_send,
- dst);
- }
+ // It's payload, keep track of buffer per peer.
+ cinfo = tunnel_get_neighbor_fc(t, &id);
+ i = (cinfo->send_buffer_start + cinfo->send_buffer_n) % t->fwd_queue_max;
+ if (NULL != cinfo->send_buffer[i])
+ {
+ GNUNET_break (cinfo->send_buffer_n == t->fwd_queue_max); // aka i == start
+ queue_destroy(cinfo->send_buffer[cinfo->send_buffer_start], GNUNET_YES);
+ cinfo->send_buffer_start++;
+ cinfo->send_buffer_start %= t->fwd_queue_max;
+ cinfo->send_buffer_n--;
+ }
+ cinfo->send_buffer[i] = queue;
+ cinfo->send_buffer_n++;
+ if (cinfo->send_buffer_n > t->fwd_queue_max)
+ {
+ GNUNET_break (0);
+ cinfo->send_buffer_n = t->fwd_queue_max;
+ }
}
GNUNET_YES : GNUNET_NO;
t->nobuffer = (0 != (opt & MESH_TUNNEL_OPT_NOBUFFER)) ?
GNUNET_YES : GNUNET_NO;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " speed_min: %d, nobuffer:%d\n",
+ t->speed_min, t->nobuffer);
if (GNUNET_YES == t->nobuffer)
- t->queue_max = 1;
+ {
+ t->bck_queue_max = 1;
+ t->fwd_queue_max = 1;
+ }
+ // FIXME only assign a local tid if a local client is interested (on demand)
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;
+ // FIXME end
tunnel_reset_timeout (t);
GNUNET_CRYPTO_hash (&t->local_tid_dest, sizeof (MESH_TunnelNumber), &hash);
/* create path: self not found in path through self */
GNUNET_break_op (0);
path_destroy (path);
- /* FIXME error. destroy tunnel? leave for timeout? */
- return 0;
+ tunnel_destroy (t);
+ return GNUNET_OK;
}
path_add_to_peers (path, GNUNET_NO);
tunnel_add_path (t, path, own_pos);
return GNUNET_OK;
}
msg = (struct GNUNET_MESH_Unicast *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %u\n",
- ntohs (msg[1].header.type));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %s\n",
+ GNUNET_MESH_DEBUG_M2S (ntohs (msg[1].header.type)));
/* Check tunnel */
t = tunnel_get (&msg->oid, ntohl (msg->tid));
if (NULL == t)
pid = ntohl (msg->pid);
if (t->fwd_pid == pid)
{
- GNUNET_STATISTICS_update (stats, "# PID drops", 1, GNUNET_NO);
+ GNUNET_STATISTICS_update (stats, "# duplicate PID drops", 1, GNUNET_NO);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
" Already seen pid %u, DROPPING!\n", pid);
return GNUNET_OK;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" pid %u not seen yet, forwarding\n", pid);
}
+
t->skip += (pid - t->fwd_pid) - 1;
t->fwd_pid = pid;
- tunnel_reset_timeout (t);
- if (is_pid_bigger (pid, t->last_fwd_ack))
+
+ if (GMC_is_pid_bigger (pid, t->last_fwd_ack))
{
- GNUNET_STATISTICS_update (stats, "# not allowed unicast", 1, GNUNET_NO);
+ GNUNET_STATISTICS_update (stats, "# unsolicited unicast", 1, GNUNET_NO);
GNUNET_break_op (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received PID %u, ACK %u\n",
+ pid, t->last_fwd_ack);
return GNUNET_OK;
- // FIXME peer sent unauthorized data. Break connection? Accept anyway?
}
+
+ tunnel_reset_timeout (t);
dest_id = GNUNET_PEER_search (&msg->destination);
if (dest_id == myid)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" it's for us! sending to clients...\n");
GNUNET_STATISTICS_update (stats, "# unicast received", 1, GNUNET_NO);
- send_subscribed_clients (message, (struct GNUNET_MessageHeader *) &msg[1]);
- // ACK sent by client, service retransmits.
+ send_subscribed_clients (message, &msg[1].header, t);
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
return GNUNET_OK;
}
ttl = ntohl (msg->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);
return GNUNET_OK;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" not for us, retransmitting...\n");
- GNUNET_STATISTICS_update (stats, "# unicast forwarded", 1, GNUNET_NO);
neighbor = tree_get_first_hop (t->tree, dest_id);
cinfo = tunnel_get_neighbor_fc (t, neighbor);
GNUNET_CONTAINER_multihashmap_iterate (t->children_fc,
&tunnel_add_skip,
&neighbor);
- if (is_pid_bigger(pid, cinfo->max_pid))
+ if (GNUNET_YES == t->nobuffer &&
+ GNUNET_YES == GMC_is_pid_bigger (pid, cinfo->fwd_ack))
{
+ GNUNET_STATISTICS_update (stats, "# unsolicited unicast", 1, GNUNET_NO);
GNUNET_break_op (0);
return GNUNET_OK;
}
send_message (message, neighbor, t);
+ GNUNET_STATISTICS_update (stats, "# unicast forwarded", 1, GNUNET_NO);
return GNUNET_OK;
}
if (t->fwd_pid == pid)
{
/* already seen this packet, drop */
- GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ GNUNET_STATISTICS_update (stats, "# duplicate PID drops", 1, GNUNET_NO);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" Already seen pid %u, DROPPING!\n", pid);
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
return GNUNET_OK;
}
else
GNUNET_CONTAINER_multihashmap_contains (t->peers, &my_full_id.hashPubKey))
{
GNUNET_STATISTICS_update (stats, "# multicast received", 1, GNUNET_NO);
- send_subscribed_clients (message, &msg[1].header);
+ send_subscribed_clients (message, &msg[1].header, t);
+ tunnel_send_fwd_ack(t, GNUNET_MESSAGE_TYPE_MESH_MULTICAST);
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ttl: %u\n", ntohl (msg->ttl));
if (ntohl (msg->ttl) == 0)
{
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);
return GNUNET_OK;
}
GNUNET_STATISTICS_update (stats, "# multicast forwarded", 1, GNUNET_NO);
struct MeshTunnel *t;
size_t size;
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a ToOrigin packet from %s\n",
GNUNET_i2s (peer));
size = ntohs (message->size);
return GNUNET_OK;
}
msg = (struct GNUNET_MESH_ToOrigin *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %u\n",
- ntohs (msg[1].header.type));
+ 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));
if (NULL == t)
return GNUNET_OK;
}
- if (t->id.oid == myid)
+ if (NULL != t->owner)
{
char cbuf[size];
struct GNUNET_MESH_ToOrigin *copy;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" it's for us! sending to clients...\n");
- if (NULL == t->owner)
- {
- /* got data packet for ownerless tunnel */
- GNUNET_STATISTICS_update (stats, "# data on ownerless tunnel",
- 1, GNUNET_NO);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " no clients!\n");
- GNUNET_break_op (0);
- return GNUNET_OK;
- }
/* TODO signature verification */
memcpy (cbuf, message, size);
copy = (struct GNUNET_MESH_ToOrigin *) cbuf;
GNUNET_STATISTICS_update (stats, "# to origin received", 1, GNUNET_NO);
GNUNET_SERVER_notification_context_unicast (nc, t->owner->handle,
©->header, GNUNET_NO);
+ tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
return GNUNET_OK;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
unsigned int atsi_count)
{
struct GNUNET_MESH_ACK *msg;
- struct MeshTunnelChildInfo *cinfo;
struct MeshTunnel *t;
uint32_t ack;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got an ACK packet from %s\n",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK packet from %s!\n",
GNUNET_i2s (peer));
msg = (struct GNUNET_MESH_ACK *) message;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " of type %u\n",
- ntohs (msg[1].header.type));
+
t = tunnel_get (&msg->oid, ntohl (msg->tid));
if (NULL == t)
return GNUNET_OK;
}
ack = ntohl (msg->pid);
- cinfo = tunnel_get_neighbor_fc (t, peer);
- cinfo->max_pid = ack;
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
+
+ /* Is this a forward or backward ACK? */
+ if (tree_get_predecessor(t->tree) != GNUNET_PEER_search(peer))
+ {
+ struct MeshTunnelChildInfo *cinfo;
+
+ debug_bck_ack++;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK\n");
+ cinfo = tunnel_get_neighbor_fc (t, peer);
+ cinfo->fwd_ack = ack;
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
+ tunnel_unlock_fwd_queues (t);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " BCK ACK\n");
+ t->bck_ack = ack;
+ tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_ACK);
+ tunnel_unlock_bck_queue (t);
+ }
return GNUNET_OK;
}
msg = (struct GNUNET_MESH_Multicast *) cbuf;
msg->header.size = htons (size);
msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_MULTICAST);
+ // FIXME: change type to != MULTICAST
msg->oid = my_full_id;
msg->tid = htonl (t->id.tid);
msg->ttl = htonl (default_ttl);
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
/* Message sanity check */
if (sizeof (struct GNUNET_MESH_TunnelMessage) != ntohs (message->size))
{
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
tunnel_msg = (struct GNUNET_MESH_TunnelMessage *) message;
/* Retrieve tunnel */
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
tunnel_msg = (struct GNUNET_MESH_TunnelMessage *) message;
/* Retrieve tunnel */
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
-
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
tunnel_msg = (struct GNUNET_MESH_TunnelMessage *) message;
/* Retrieve tunnel */
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
peer_msg = (struct GNUNET_MESH_PeerControl *) message;
+
/* Sanity check for message size */
if (sizeof (struct GNUNET_MESH_PeerControl) != ntohs (peer_msg->header.size))
{
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
peer_msg = (struct GNUNET_MESH_PeerControl *) message;
+
/* Sanity check for message size */
if (sizeof (struct GNUNET_MESH_PeerControl) != ntohs (peer_msg->header.size))
{
* @param cls closure
* @param client identification of the client
* @param message the actual message (PeerControl)
+ *
+ * FIXME implement DHT block bloomfilter
*/
static void
handle_local_blacklist (void *cls, struct GNUNET_SERVER_Client *client,
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
peer_msg = (struct GNUNET_MESH_PeerControl *) message;
/* Sanity check for message size */
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
peer_msg = (struct GNUNET_MESH_PeerControl *) message;
/* Sanity check for message size */
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
connect_msg = (struct GNUNET_MESH_ConnectPeerByType *) message;
+
/* Sanity check for message size */
if (sizeof (struct GNUNET_MESH_ConnectPeerByType) !=
ntohs (connect_msg->header.size))
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
/* Message size sanity check */
if (sizeof(struct GNUNET_MESH_ConnectPeerByString) >= size)
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
data_msg = (struct GNUNET_MESH_Unicast *) message;
+
/* Sanity check for message size */
size = ntohs (message->size);
if (sizeof (struct GNUNET_MESH_Unicast) +
return;
}
+ /* PID should be as expected */
+ if (ntohl (data_msg->pid) != t->fwd_pid + 1)
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Unicast PID, expected %u, got %u\n",
+ t->fwd_pid + 1, ntohl (data_msg->pid));
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
+ }
+
/* Ok, everything is correct, send the message
* (pretend we got it from a mesh peer)
*/
{
+ /* Work around const limitation */
char buf[ntohs (message->size)] GNUNET_ALIGN;
struct GNUNET_MESH_Unicast *copy;
- /* Work around const limitation */
copy = (struct GNUNET_MESH_Unicast *) buf;
memcpy (buf, data_msg, size);
copy->oid = my_full_id;
copy->tid = htonl (t->id.tid);
copy->ttl = htonl (default_ttl);
- copy->pid = htonl (t->fwd_pid + 1);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" calling generic handler...\n");
handle_mesh_data_unicast (NULL, &my_full_id, ©->header, NULL, 0);
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "receive done OK\n");
GNUNET_SERVER_receive_done (client, GNUNET_OK);
+
return;
}
const struct GNUNET_MessageHeader *message)
{
struct GNUNET_MESH_ToOrigin *data_msg;
- struct GNUNET_PeerIdentity id;
+ struct MeshTunnelClientInfo *clinfo;
struct MeshClient *c;
struct MeshTunnel *t;
MESH_TunnelNumber tid;
size_t size;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Got a ToOrigin request from a client!\n");
/* Sanity check for client registration */
if (NULL == (c = client_get (client)))
{
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
data_msg = (struct GNUNET_MESH_ToOrigin *) message;
+
/* Sanity check for message size */
size = ntohs (message->size);
if (sizeof (struct GNUNET_MESH_ToOrigin) +
/* Tunnel exists? */
tid = ntohl (data_msg->tid);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Got a ToOrigin request from a client! Tunnel %X\n", tid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " on tunnel %X\n", tid);
if (tid < GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
{
GNUNET_break (0);
t = tunnel_get_by_local_id (c, tid);
if (NULL == t)
{
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Tunnel %X unknown.\n", tid);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " for client %u.\n", c->id);
GNUNET_break (0);
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
/* It should be sent by someone who has this as incoming tunnel. */
- if (-1 == client_knows_tunnel (c, t))
+ if (GNUNET_NO == client_knows_tunnel (c, t))
+ {
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
+ }
+
+ /* PID should be as expected */
+ clinfo = tunnel_get_client_fc (t, c);
+ if (ntohl (data_msg->pid) != clinfo->bck_pid + 1)
{
GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "To Origin PID, expected %u, got %u\n",
+ clinfo->bck_pid + 1,
+ ntohl (data_msg->pid));
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
- GNUNET_PEER_resolve (t->id.oid, &id);
/* Ok, everything is correct, send the message
* (pretend we got it from a mesh peer)
*/
+ clinfo->bck_pid++;
{
char buf[ntohs (message->size)] GNUNET_ALIGN;
struct GNUNET_MESH_ToOrigin *copy;
/* Work around const limitation */
copy = (struct GNUNET_MESH_ToOrigin *) buf;
memcpy (buf, data_msg, size);
- copy->oid = id;
+ GNUNET_PEER_resolve (t->id.oid, ©->oid);
copy->tid = htonl (t->id.tid);
+ copy->ttl = htonl (default_ttl);
+ if (ntohl (copy->pid) != (t->bck_pid + 1))
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "To Origin PID, expected %u, got %u\n",
+ t->bck_pid + 1,
+ ntohl (copy->pid));
+ return;
+ }
+ t->bck_pid++;
copy->sender = my_full_id;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" calling generic handler...\n");
handle_mesh_data_to_orig (NULL, &my_full_id, ©->header, NULL, 0);
}
GNUNET_SERVER_receive_done (client, GNUNET_OK);
+
return;
}
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
+
data_msg = (struct GNUNET_MESH_Multicast *) message;
+
/* Sanity check for message size */
if (sizeof (struct GNUNET_MESH_Multicast) +
sizeof (struct GNUNET_MessageHeader) > ntohs (data_msg->header.size))
t = tunnel_get_by_local_id (c, tid);
if (NULL == t)
{
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Tunnel %X unknown.\n", tid);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " for client %u.\n", c->id);
GNUNET_break (0);
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
return;
}
+ /* PID should be as expected */
+ if (ntohl (data_msg->pid) != t->fwd_pid + 1)
+ {
+ GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Multicast PID, expected %u, got %u\n",
+ t->fwd_pid + 1, ntohl (data_msg->pid));
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
+ }
+
{
char buf[ntohs (message->size)] GNUNET_ALIGN;
struct GNUNET_MESH_Multicast *copy;
copy->oid = my_full_id;
copy->tid = htonl (t->id.tid);
copy->ttl = htonl (default_ttl);
- copy->pid = htonl (t->fwd_pid + 1);
+ GNUNET_assert (ntohl (copy->pid) == (t->fwd_pid + 1));
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" calling generic handler...\n");
handle_mesh_data_multicast (client, &my_full_id, ©->header, NULL, 0);
}
- /* receive done gets called when last copy is sent to a neighbor */
+ GNUNET_SERVER_receive_done (t->owner->handle, GNUNET_OK);
return;
}
struct MeshTunnel *t;
struct MeshClient *c;
MESH_TunnelNumber tid;
+ uint32_t ack;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got a local ACK\n");
/* Sanity check for client registration */
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " by client %u\n", c->id);
msg = (struct GNUNET_MESH_LocalAck *) message;
+
/* Tunnel exists? */
tid = ntohl (msg->tunnel_id);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " on tunnel %X\n", tid);
if (NULL == t)
{
GNUNET_break (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Tunnel %X unknown.\n", tid);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " for client %u.\n", c->id);
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
- /* Does client own tunnel? Is this and ACK for BCK traffic? */
+ ack = ntohl (msg->max_pid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ack);
+
+ /* Does client own tunnel? I.E: Is this and ACK for BCK traffic? */
if (NULL != t->owner && t->owner->handle == client)
{
- GNUNET_break (0);
- // FIXME TODO
+ /* The client owns the tunnel, ACK is for data to_origin, send BCK ACK. */
+ t->bck_ack = ack;
+ tunnel_send_bck_ack(t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
}
else
{
/* The client doesn't own the tunnel, this ACK is for FWD traffic. */
- tunnel_set_client_fwd_ack (t, c, ntohl (msg->max_pid));
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
+ tunnel_set_client_fwd_ack (t, c, ack);
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
}
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+
return;
}
core_init (void *cls, struct GNUNET_CORE_Handle *server,
const struct GNUNET_PeerIdentity *identity)
{
+ static int i = 0;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core init\n");
core_handle = server;
if (0 != memcmp (identity, &my_full_id, sizeof (my_full_id)) ||
NULL == server)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Wrong CORE service\n"));
- GNUNET_SCHEDULER_shutdown ();
+ GNUNET_SCHEDULER_shutdown (); // Try gracefully
+ if (10 < i++)
+ GNUNET_abort(); // Try harder
}
return;
}
GNUNET_CRYPTO_hash (&my_public_key, sizeof (my_public_key),
&my_full_id.hashPubKey);
myid = GNUNET_PEER_intern (&my_full_id);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Mesh for peer [%s] starting\n",
+ GNUNET_i2s(&my_full_id));
// transport_handle = GNUNET_TRANSPORT_connect(c,
// &my_full_id,
NULL)) ? 0 : 1;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "main() END\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