#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__)
};
+/**
+ * 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;
+ uint64_t mid;
/**
* When was this message issued (to calculate ACK delay) FIXME update with traffic
};
-/**
- * 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.
+ */
+ uint64_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.
+ */
+ uint64_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
*/
- struct GNUNET_TIME_Relative expected_delay;
+ struct GNUNET_TIME_Relative expected_delay;
};
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);
}
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);
tunnel_send_fwd_data_ack (struct MeshTunnel *t)
{
struct GNUNET_MESH_DataACK msg;
+ struct MeshTunnelReliability *rel;
+ struct MeshReliableMessage *copy;
+ uint64_t mask;
+ unsigned int i;
+ unsigned int delta;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "send_fwd_data_ack for %llu\n",
+ t->bck_rel->mid_recv - 1);
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
+ msg.mid = GNUNET_htonll (t->bck_rel->mid_recv - 1);
+ msg.futures = 0;
+ rel = t->bck_rel;
+ for (i = 0, copy = rel->head_recv;
+ i < 64 && NULL != copy;
+ i++, copy = copy->next)
+ {
+ delta = copy->mid - t->bck_rel->mid_recv;
+ mask = 0x1 << 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->prev_hop, t);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_fwd_data_ack END\n");
}
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.mid = GNUNET_htonll (t->fwd_rel->mid_recv - 1);
msg.futures = 0; // FIXME set bits of other newer messages received
send_prebuilt_message (&msg.header, t->next_hop, t);
tunnel_send_fwd_ack (struct MeshTunnel *t, uint16_t type)
{
uint32_t ack;
+ int delta;
/* Is it after unicast retransmission? */
switch (type)
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;
}
/* Check if we need to transmit the ACK */
- if (NULL == t->owner &&
+ if (0 && 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)
}
/* Ok, ACK might be necessary, what PID to ACK? */
- ack = t->prev_fc.last_pid_recv + t->queue_max - t->next_fc.queue_n;
+ delta = t->queue_max - t->next_fc.queue_n;
+ if (0 > delta || (GNUNET_YES == t->reliable &&
+ NULL != t->owner &&
+ t->fwd_rel->n_sent > 10))
+ delta = 0;
+ if (NULL != t->owner && delta > 1)
+ delta = 1;
+ ack = t->prev_fc.last_pid_recv + delta;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " FWD ACK %u\n", ack);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
" last %u, qmax %u, q %u\n",
" Not sending ACK, nobuffer + traffic\n");
return;
}
+ if (GNUNET_YES == t->reliable && NULL != t->owner)
+ tunnel_send_bck_data_ack (t);
break;
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:
t->force_ack = GNUNET_YES;
}
+/**
+ * Send up to 64 buffered messages to the client for in order delivery.
+ *
+ * @param t Tunnel on which to empty the message buffer.
+ */
+static void
+tunnel_send_client_buffered_ucast (struct MeshTunnel *t)
+{
+ struct MeshTunnelReliability *rel;
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *next;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_buffered_unicast\n");
+ rel = t->bck_rel;
+ 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 %llu! now expecting %llu\n",
+ copy->mid, rel->mid_recv + 1LL);
+ tunnel_send_client_ucast (t, msg);
+ 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 %llu, (%llu)\n",
+ rel->mid_recv,
+ copy->mid);
+ return;
+ }
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send_buffered_unicast 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.
+ */
+static void
+tunnel_add_buffer_ucast (struct MeshTunnel *t,
+ const struct GNUNET_MESH_Data *msg)
+{
+ struct MeshTunnelReliability *rel;
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *prev;
+ uint64_t mid;
+ uint16_t size;
+
+ rel = t->bck_rel;
+ size = ntohs (msg->header.size);
+ mid = GNUNET_ntohll (msg->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "add_buffer_ucast %llu\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...
+ for (prev = rel->head_recv; NULL != prev; prev = prev->next)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " prev %llu\n", prev->mid);
+ if (mid < prev->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_buffer_ucast END\n");
+}
+
+
+/**
+ * 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.
+ */
+static void
+tunnel_free_buffer_ucast (struct MeshTunnel *t,
+ const struct GNUNET_MESH_DataACK *msg)
+{
+ struct MeshTunnelReliability *rel;
+ struct MeshReliableMessage *copy;
+ struct MeshReliableMessage *next;
+ uint64_t bitfield;
+ uint64_t mask;
+ uint64_t mid;
+ uint64_t target;
+ unsigned int i;
+
+ bitfield = msg->futures;
+ mid = GNUNET_ntohll (msg->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "free_sent_buffer %llu %llX\n",
+ mid, bitfield);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " rel %p, copy %p\n",
+ mid, bitfield);
+ rel = t->fwd_rel;
+ for (i = 0, copy = rel->head_recv;
+ i < 64 && NULL != copy && 0 != bitfield;
+ i++, copy = next)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " trying %u\n", i);
+ 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 %llu\n", target);
+ while (NULL != copy && copy->mid < target)
+ 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 (copy->mid > target)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " next copy %llu\n", copy->mid);
+ next = copy;
+ continue;
+ }
+
+ /* Now copy->mid == target, free it */
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! Freeing %llu\n", target);
+ copy = copy->next;
+ GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
+ GNUNET_free (copy);
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "free_sent_buffer END\n");
+}
+
+
/**
* Modify the to_origin message TID from global to local and send to client.
*
{
struct MeshTunnelReliability *rel = cls;
struct MeshReliableMessage *copy;
+ struct MeshFlowControl *fc;
+ struct MeshPeerQueue *q;
+ struct MeshPeerInfo *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)
+ {
+ struct GNUNET_TIME_Relative diff;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! RETRANSMIT %llu\n", copy->mid);
+ diff = GNUNET_TIME_absolute_get_duration (copy->timestamp);
+ diff = GNUNET_TIME_relative_divide (diff, 10);
+ copy->timestamp = GNUNET_TIME_absolute_subtract (GNUNET_TIME_absolute_get(),
+ diff);
+
+ fc->last_ack_sent++;
+ payload->pid = htonl (fc->last_pid_recv + 1);
+ 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 %llu\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);
}
return q;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "* ACK: %u, PID: %u\n",
- ack, pid);
+ "* ACK: %u, PID: %u, MID: %llu\n",
+ ack, pid, GNUNET_ntohll (dmsg->mid));
if (GNUNET_NO == GMC_is_pid_bigger (pid, ack))
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
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;
+ t->next_fc.last_pid_sent = pid;
tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
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;
+ t->prev_fc.last_pid_sent = pid;
tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
break;
default:
else
{
GNUNET_break (0);
- fc = NULL;
+ return data_size;
}
if (NULL != fc && GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
{
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++;
}
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_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->prev_fc.last_ack_sent);
+ "Received PID %u, (prev %u), ACK %u\n",
+ pid, t->prev_fc.last_pid_recv, t->prev_fc.last_ack_sent);
tunnel_send_fwd_ack(t, GNUNET_MESSAGE_TYPE_MESH_POLL);
return GNUNET_OK;
}
" 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) )
+ if (GMC_is_pid_bigger (pid, t->prev_fc.last_pid_recv))
{
+ uint64_t mid;
+
+ mid = GNUNET_ntohll (msg->mid);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- " pid %u not seen yet, forwarding\n", pid);
+ " pid %u (%llu) not seen yet\n", pid, mid);
t->prev_fc.last_pid_recv = pid;
- tunnel_send_client_ucast (t, msg);
+
+ if (GNUNET_NO == t->reliable ||
+ (mid >= t->bck_rel->mid_recv && mid <= t->bck_rel->mid_recv + 64))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "!!! RECV %llu\n", GNUNET_ntohll(msg->mid));
+ if (GNUNET_YES == t->reliable)
+ {
+ /* Is this the exact next expected messasge? */
+ if (mid == t->bck_rel->mid_recv)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "as expected\n");
+ t->bck_rel->mid_recv++;
+ tunnel_send_client_ucast (t, msg);
+ tunnel_send_client_buffered_ucast (t);
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "save for later\n");
+ tunnel_add_buffer_ucast (t, msg);
+ }
+ }
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " MID %llu not expected (%llu - %llu), dropping!\n",
+ GNUNET_ntohll (msg->mid),
+ t->bck_rel->mid_recv, t->bck_rel->mid_recv + 64LL);
+ }
}
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",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ " Pid %u not expected (%u), dropping!\n",
pid, t->prev_fc.last_pid_recv + 1);
}
- tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST_ACK);
+ tunnel_send_fwd_ack (t, GNUNET_MESSAGE_TYPE_MESH_UNICAST);
return GNUNET_OK;
}
- if (GMC_is_pid_bigger(pid, t->prev_fc.last_pid_recv))
- t->prev_fc.last_pid_recv = pid;
+ t->prev_fc.last_pid_recv = pid;
if (0 == t->next_hop)
{
GNUNET_break (0);
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);
+ tunnel_send_bck_ack (t, GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
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;
+ t->next_fc.last_pid_recv = pid;
if (0 == t->prev_hop) /* No owner AND no prev hop */
{
if (GNUNET_YES == t->destroy)
struct MeshReliableMessage *next;
struct MeshTunnel *t;
GNUNET_PEER_Id id;
- uint32_t ack;
+ uint64_t ack;
uint16_t type;
int work;
GNUNET_STATISTICS_update (stats, "# ack on unknown tunnel", 1, GNUNET_NO);
return GNUNET_OK;
}
- ack = ntohl (msg->pid);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
+ ack = GNUNET_ntohll (msg->mid);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %llu\n", ack);
/* Is this a forward or backward ACK? */
id = GNUNET_PEER_search (peer);
return GNUNET_OK;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! ACK %u\n", ack);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! ACK %llu\n", ack);
for (work = GNUNET_NO, copy = rel->head_sent; copy != NULL; copy = next)
{
struct GNUNET_TIME_Relative time;
- if (copy->id > ack)
+ if (copy->mid > ack)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! head %u, out!\n", copy->id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! head %llu, out!\n", copy->mid);
+ tunnel_free_buffer_ucast (t, msg);
break;
}
work = GNUNET_YES;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! id %u\n", copy->id);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! id %llu\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;
+ rel->n_sent--;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "!!! new expected delay %s\n",
GNUNET_STRINGS_relative_time_to_string (rel->expected_delay,
GNUNET_NO));
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,
/******************************************************************************/
+/**
+ * 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
*
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_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,
+ 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->own_tunnels);
- GNUNET_CONTAINER_multihashmap32_destroy (c->incoming_tunnels);
+ GNUNET_CONTAINER_multihashmap32_destroy (c->incoming_tunnels);
+ }
if (NULL != c->ports)
GNUNET_CONTAINER_multihashmap32_destroy (c->ports);
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);
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");
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 %llu\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_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 = GNUNET_htonll (copy->mid);
}
else
{
payload = (struct GNUNET_MESH_Data *) cbuf;
+ payload->mid = 0;
+ // FIXME FIXME FIXME FIXME FIXME FIXME FIXME FIXME FIXME FIXME
+ // use different struct for unreliable traffic, save 8 bytes
}
memcpy (&payload[1], &data_msg[1], size);
payload->header.size = htons (sizeof (struct GNUNET_MESH_Data) + size);
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);
projects / oweals / gnunet.git / blobdiff