#define LOG(level, ...) GNUNET_log_from(level,"mesh-chn",__VA_ARGS__)
-#define MESH_RETRANSMIT_TIME GNUNET_TIME_UNIT_SECONDS
+#define MESH_RETRANSMIT_TIME GNUNET_TIME_relative_multiply(\
+ GNUNET_TIME_UNIT_MILLISECONDS, 250)
#define MESH_RETRANSMIT_MARGIN 4
/**
* Tunnel Queue.
*/
- struct MeshTunnel3Queue *q;
+ struct MeshTunnel3Queue *tq;
/**
* Message type (DATA/DATA_ACK)
/**
* Tunnel Queue.
*/
- struct MeshChannelQueue *q;
+ struct MeshChannelQueue *chq;
/**
* When was this message issued (to calculate ACK delay)
* is skewed by the retransmission, count only for the
* retransmitted message.
*/
-static void
+static int
rel_message_free (struct MeshReliableMessage *copy, int update_time);
/**
static void
send_create (struct MeshChannel *ch);
+/**
+ * Confirm we got a channel create, FWD ack.
+ *
+ * @param ch The channel to confirm.
+ * @param fwd Should we send a FWD ACK? (going dest->root)
+ */
+static void
+send_ack (struct MeshChannel *ch, int fwd);
+
+
+
+/**
+ * Test if the channel is loopback: both root and dest are on the local peer.
+ *
+ * @param ch Channel to test.
+ *
+ * @return #GNUNET_YES if channel is loopback, #GNUNET_NO otherwise.
+ */
+static int
+is_loopback (const struct MeshChannel *ch)
+{
+ if (NULL != ch->t)
+ return GMT_is_loopback (ch->t);
+
+ return (NULL != ch->root && NULL != ch->dest);
+}
/**
LOG (GNUNET_ERROR_TYPE_DEBUG, "add_buffered_data END\n");
}
+
/**
* Add a destination client to a channel, initializing all data structures
* in the channel and the client.
ch->dest_rel = GNUNET_new (struct MeshChannelReliability);
ch->dest_rel->ch = ch;
ch->dest_rel->expected_delay.rel_value_us = 0;
+ ch->dest_rel->retry_timer = MESH_RETRANSMIT_TIME;
ch->dest = c;
}
send_client_ack (struct MeshChannel *ch, int fwd)
{
struct MeshChannelReliability *rel = fwd ? ch->root_rel : ch->dest_rel;
+ struct MeshClient *c = fwd ? ch->root : ch->dest;
+ if (NULL == c)
+ {
+ GNUNET_break (GNUNET_NO != ch->destroy);
+ return;
+ }
LOG (GNUNET_ERROR_TYPE_DEBUG,
" sending %s ack to client on channel %s\n",
GM_f2s (fwd), GMCH_2s (ch));
}
rel->client_allowed = GNUNET_YES;
- GML_send_ack (fwd ? ch->root : ch->dest, fwd ? ch->lid_root : ch->lid_dest);
+ GML_send_ack (c, fwd ? ch->lid_root : ch->lid_dest);
}
GNUNET_break (0);
return;
}
- GML_send_nack (ch->root, ch->lid_root);
+ GML_send_channel_nack (ch->root, ch->lid_root);
}
{
send_create (rel->ch);
}
+ else if (rel == rel->ch->dest_rel)
+ {
+ send_ack (rel->ch, GNUNET_YES);
+ }
else
{
-
+ GNUNET_break (0);
}
}
struct MeshTunnel3Queue *q,
uint16_t type, size_t size)
{
- struct MeshChannelQueue *ch_q = cls;
- struct MeshReliableMessage *copy = ch_q->copy;
+ struct MeshChannelQueue *chq = cls;
+ struct MeshReliableMessage *copy = chq->copy;
struct MeshChannelReliability *rel;
- switch (ch_q->type)
+ switch (chq->type)
{
case GNUNET_MESSAGE_TYPE_MESH_DATA:
- LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! SENT %u %s (c: %p, q: %p)\n",
- copy->mid, GM_m2s (type), copy, copy->q);
- GNUNET_assert (ch_q == copy->q);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! SENT DATA MID %u\n", copy->mid);
+ GNUNET_assert (chq == copy->chq);
copy->timestamp = GNUNET_TIME_absolute_get ();
rel = copy->rel;
if (GNUNET_SCHEDULER_NO_TASK == rel->retry_task)
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!! scheduling retry in 4 * %s\n",
+ GNUNET_STRINGS_relative_time_to_string (rel->expected_delay,
+ GNUNET_YES));
if (0 != rel->expected_delay.rel_value_us)
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!! delay != 0\n");
rel->retry_timer =
GNUNET_TIME_relative_multiply (rel->expected_delay,
MESH_RETRANSMIT_MARGIN);
}
else
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!! delay reset\n");
rel->retry_timer = MESH_RETRANSMIT_TIME;
}
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!! using delay %s\n",
+ GNUNET_STRINGS_relative_time_to_string (rel->retry_timer,
+ GNUNET_NO));
rel->retry_task =
GNUNET_SCHEDULER_add_delayed (rel->retry_timer,
&channel_retransmit_message, rel);
}
- copy->q = NULL;
+ else
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!! retry task %u\n", rel->retry_task);
+ }
+ copy->chq = NULL;
break;
case GNUNET_MESSAGE_TYPE_MESH_DATA_ACK:
- rel = ch_q->rel;
- GNUNET_assert (rel->uniq == ch_q);
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_ACK:
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! SENT %s\n", GM_m2s (chq->type));
+ rel = chq->rel;
+ GNUNET_assert (rel->uniq == chq);
rel->uniq = NULL;
- break;
-
- case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
- rel = ch_q->rel;
- GNUNET_assert (rel->uniq == ch_q);
- if (MESH_CHANNEL_READY != rel->ch->state)
+ if (MESH_CHANNEL_READY != rel->ch->state
+ && GNUNET_MESSAGE_TYPE_MESH_DATA_ACK != type
+ && GNUNET_NO == rel->ch->destroy)
{
- struct GNUNET_TIME_Relative delay;
-
GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == rel->retry_task);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! STD BACKOFF %s\n",
+ GNUNET_STRINGS_relative_time_to_string (rel->retry_timer,
+ GNUNET_NO));
rel->retry_timer = GNUNET_TIME_STD_BACKOFF (rel->retry_timer);
rel->retry_task = GNUNET_SCHEDULER_add_delayed (rel->retry_timer,
&channel_recreate, rel);
}
- rel->uniq = NULL;
break;
GNUNET_break (0);
}
- GNUNET_free (ch_q);
+ GNUNET_free (chq);
}
send_create (struct MeshChannel *ch)
{
struct GNUNET_MESH_ChannelCreate msgcc;
- struct MeshChannelQueue *q;
msgcc.header.size = htons (sizeof (msgcc));
msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE);
msgcc.port = htonl (ch->port);
msgcc.opt = htonl (channel_get_options (ch));
- q = GNUNET_new (struct MeshChannelQueue);
- q->rel = ch->root_rel;
+ GMCH_send_prebuilt_message (&msgcc.header, ch, GNUNET_YES, NULL);
+}
+
+
+/**
+ * Confirm we got a channel create or FWD ack.
+ *
+ * @param ch The channel to confirm.
+ * @param fwd Should we send a FWD ACK? (going dest->root)
+ */
+static void
+send_ack (struct MeshChannel *ch, int fwd)
+{
+ struct GNUNET_MESH_ChannelManage msg;
+
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_ACK);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " sending channel %s ack for channel %s\n",
+ GM_f2s (fwd), GMCH_2s (ch));
- /* FIXME cancel on confirm */
- q->q = GMT_send_prebuilt_message (&msgcc.header, ch->t, ch,
- GNUNET_YES, GNUNET_YES,
- ch_message_sent, q);
- q->rel->uniq = q;
+ msg.chid = htonl (ch->gid);
+ GMCH_send_prebuilt_message (&msg.header, ch, !fwd, NULL);
+}
+
+
+/**
+ * Send a message and don't keep any info about it: we won't need to cancel it
+ * or resend it.
+ *
+ * @param msg Header of the message to fire away.
+ * @param ch Channel on which the message should go.
+ * @param force Is this a forced (undroppable) message?
+ */
+static void
+fire_and_forget (const struct GNUNET_MessageHeader *msg,
+ struct MeshChannel *ch,
+ int force)
+{
+ GNUNET_break (NULL == GMT_send_prebuilt_message (msg, ch->t, force,
+ NULL, NULL));
+}
+
+
+/**
+ * Notify that a channel create didn't succeed.
+ *
+ * @param ch The channel to reject.
+ */
+static void
+send_nack (struct MeshChannel *ch)
+{
+ struct GNUNET_MESH_ChannelManage msg;
+
+ msg.header.size = htons (sizeof (msg));
+ msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_NACK);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " sending channel NACK for channel %s\n",
+ GMCH_2s (ch));
+
+ msg.chid = htonl (ch->gid);
+ GMCH_send_prebuilt_message (&msg.header, ch, GNUNET_NO, NULL);
}
next = copy->next;
GNUNET_CONTAINER_DLL_remove (rel->head_recv, rel->tail_recv, copy);
LOG (GNUNET_ERROR_TYPE_DEBUG, " COPYFREE BATCH RECV %p\n", copy);
+ GNUNET_break (NULL == copy->chq);
GNUNET_free (copy);
}
for (copy = rel->head_sent; NULL != copy; copy = next)
next = copy->next;
GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
LOG (GNUNET_ERROR_TYPE_DEBUG, " COPYFREE BATCH %p\n", copy);
+ if (NULL != copy->chq)
+ {
+ if (NULL != copy->chq->tq)
+ {
+ GMT_cancel (copy->chq->tq);
+ /* ch_message_sent will free copy->q */
+ }
+ else
+ {
+ GNUNET_free (copy->chq);
+ GNUNET_break (0);
+ }
+ }
GNUNET_free (copy);
}
+ if (NULL != rel->uniq && NULL != rel->uniq->tq)
+ {
+ GMT_cancel (rel->uniq->tq);
+ /* ch_message_sent is called freeing uniq */
+ }
if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
{
GNUNET_SCHEDULER_cancel (rel->retry_task);
+ rel->retry_task = GNUNET_SCHEDULER_NO_TASK;
}
- if (NULL != rel->uniq)
- GMT_cancel (rel->uniq->q);
GNUNET_free (rel);
}
target = mid + i + 1;
LOG (GNUNET_ERROR_TYPE_DEBUG, " target %u\n", target);
while (NULL != copy && GM_is_pid_bigger (target, copy->mid))
- copy = copy->next;
+ copy = copy->next;
/* Did we run out of copies? (previously freed, it's ok) */
if (NULL == copy)
/* Now copy->mid == target, free it */
next = copy->next;
- rel_message_free (copy, GNUNET_YES);
+ GNUNET_break (GNUNET_YES != rel_message_free (copy, GNUNET_YES));
copy = next;
}
LOG (GNUNET_ERROR_TYPE_DEBUG, "free_sent_reliable END\n");
* If this message is ACK in a batch the timing information
* is skewed by the retransmission, count only for the
* retransmitted message.
+ *
+ * @return #GNUNET_YES if channel was destroyed as a result of the call,
+ * #GNUNET_NO otherwise.
*/
-static void
+static int
rel_message_free (struct MeshReliableMessage *copy, int update_time)
{
struct MeshChannelReliability *rel;
LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! batch free, ignoring timing\n");
}
rel->ch->pending_messages--;
- if (GNUNET_NO != rel->ch->destroy && 0 == rel->ch->pending_messages)
- {
- struct MeshTunnel3 *t = rel->ch->t;
- GMCH_destroy (rel->ch);
- GMT_destroy_if_empty (t);
- }
- if (NULL != copy->q)
+ if (NULL != copy->chq)
{
- GMT_cancel (copy->q->q);
+ GMT_cancel (copy->chq->tq);
+ /* copy->q is set to NULL by ch_message_sent */
}
GNUNET_CONTAINER_DLL_remove (rel->head_sent, rel->tail_sent, copy);
LOG (GNUNET_ERROR_TYPE_DEBUG, " COPYFREE %p\n", copy);
GNUNET_free (copy);
-}
-
-/**
- * Confirm we got a channel create.
- *
- * @param ch The channel to confirm.
- * @param fwd Should we send a FWD ACK? (going dest->root)
- */
-static void
-channel_send_ack (struct MeshChannel *ch, int fwd)
-{
- struct GNUNET_MESH_ChannelManage msg;
-
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_ACK);
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- " sending channel %s ack for channel %s\n",
- GM_f2s (fwd), GMCH_2s (ch));
-
- msg.chid = htonl (ch->gid);
- GMCH_send_prebuilt_message (&msg.header, ch, !fwd, NULL);
-}
-
-
-/**
- * Notify that a channel create didn't succeed.
- *
- * @param ch The channel to reject.
- */
-static void
-channel_send_nack (struct MeshChannel *ch)
-{
- struct GNUNET_MESH_ChannelManage msg;
-
- msg.header.size = htons (sizeof (msg));
- msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_NACK);
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- " sending channel NACK for channel %s\n",
- GMCH_2s (ch));
-
- msg.chid = htonl (ch->gid);
- GMCH_send_prebuilt_message (&msg.header, ch, GNUNET_NO, NULL);
+ if (GNUNET_NO != rel->ch->destroy && 0 == rel->ch->pending_messages)
+ {
+ struct MeshTunnel3 *t = rel->ch->t;
+ GMCH_destroy (rel->ch);
+ GMT_destroy_if_empty (t);
+ return GNUNET_YES;
+ }
+ return GNUNET_NO;
}
channel_confirm (struct MeshChannel *ch, int fwd)
{
struct MeshChannelReliability *rel;
-
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- " channel confirm %s %s:%X\n",
- GM_f2s (fwd), GMT_2s (ch->t), ch->gid);
- ch->state = MESH_CHANNEL_READY;
+ enum MeshChannelState oldstate;
rel = fwd ? ch->root_rel : ch->dest_rel;
- rel->client_ready = GNUNET_YES;
- rel->expected_delay = rel->retry_timer;
- send_client_ack (ch, fwd);
-
- if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
- {
- GNUNET_SCHEDULER_cancel (rel->retry_task);
- rel->retry_task = GNUNET_SCHEDULER_NO_TASK;
- }
- else if (NULL != rel->uniq)
+ if (NULL == rel)
{
- GMT_cancel (rel->uniq->q);
- /* ch_sent_message will free and NULL uniq */
+ GNUNET_break (GNUNET_NO != ch->destroy);
+ return;
}
- else
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " channel confirm %s %s\n",
+ GM_f2s (fwd), GMCH_2s (ch));
+ oldstate = ch->state;
+ ch->state = MESH_CHANNEL_READY;
+
+ if (MESH_CHANNEL_READY != oldstate || GNUNET_YES == is_loopback (ch))
{
- /* We SHOULD have been trying to retransmit this! */
- GNUNET_break (0);
+ rel->client_ready = GNUNET_YES;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " !! retry timer confirm %s\n",
+ GNUNET_STRINGS_relative_time_to_string (rel->retry_timer, GNUNET_NO));
+ rel->expected_delay = rel->retry_timer;
+ if (GMT_get_connections_buffer (ch->t) > 0 || GMT_is_loopback (ch->t))
+ send_client_ack (ch, fwd);
+
+ if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
+ {
+ GNUNET_SCHEDULER_cancel (rel->retry_task);
+ rel->retry_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ else if (NULL != rel->uniq)
+ {
+ GMT_cancel (rel->uniq->tq);
+ /* ch_message_sent will free and NULL uniq */
+ }
+ else
+ {
+ if (GNUNET_NO == is_loopback (ch))
+ {
+ /* We SHOULD have been trying to retransmit this! */
+ GNUNET_break (0);
+ }
+ }
}
/* In case of a FWD ACK (SYNACK) send a BCK ACK (ACK). */
if (GNUNET_YES == fwd)
- channel_send_ack (ch, GNUNET_NO);
+ send_ack (ch, GNUNET_NO);
}
}
-/**
- * Test if the channel is loopback: both root and dest are on the local peer.
- *
- * @param ch Channel to test.
- *
- * @return #GNUNET_YES if channel is loopback, #NGUNET_NO otherwise.
- */
-static int
-is_loopback (const struct MeshChannel *ch)
-{
- if (NULL != ch->t)
- return GMT_is_loopback (ch->t);
-
- return (NULL != ch->root && NULL != ch->dest);
-}
-
-
/**
* Handle a loopback message: call the appropriate handler for the message type.
*
{
case GNUNET_MESSAGE_TYPE_MESH_DATA:
/* Don't send hop ACK, wait for client to ACK */
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! SEND loopback %u (%u)\n",
+ ntohl (((struct GNUNET_MESH_Data *) msgh)->mid), ntohs (msgh->size));
GMCH_handle_data (ch, (struct GNUNET_MESH_Data *) msgh, fwd);
break;
if (NULL == ch)
return;
+ if (2 == ch->destroy)
+ return; /* recursive call */
+ ch->destroy = 2;
LOG (GNUNET_ERROR_TYPE_DEBUG, "destroying channel %s:%u\n",
GMT_2s (ch->t), ch->gid);
/**
- * Get channel ID.
+ * Get the channel's public ID.
*
* @param ch Channel.
*
- * @return ID
+ * @return ID used to identify the channel with the remote peer.
*/
MESH_ChannelNumber
GMCH_get_id (const struct MeshChannel *ch)
GNUNET_break (GNUNET_NO != ch->destroy);
return;
}
- if (NULL != rel->head_sent && 64 <= rel->mid_send - rel->head_sent->mid)
+ if (NULL != rel->head_sent)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " too big MID gap! Wait for ACK.\n");
- return;
+ if (64 <= rel->mid_send - rel->head_sent->mid)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " too big MID gap! Wait for ACK.\n");
+ return;
+ }
+ else
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " gap ok: %u - %u\n",
+ rel->head_sent->mid, rel->mid_send);
+ }
+ else
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " head sent is NULL\n");
}
}
if (is_loopback (ch))
{
if (GMCH_get_buffer (ch, fwd) > 0)
- send_client_ack (ch, fwd);
+ GMCH_allow_client (ch, fwd);
return GNUNET_OK;
}
if (GMT_get_connections_buffer (ch->t) > 0)
{
- send_client_ack (ch, fwd);
+ GMCH_allow_client (ch, fwd);
}
return GNUNET_OK;
}
else
{
+ /* FIXME change to a tunnel API, eliminate ch <-> peer connection */
GMP_connect (peer);
}
/* In unreliable channels, we'll use the DLL to buffer BCK data */
ch->root_rel = GNUNET_new (struct MeshChannelReliability);
ch->root_rel->ch = ch;
- ch->root_rel->retry_timer = GNUNET_TIME_UNIT_SECONDS;
+ ch->root_rel->retry_timer = MESH_RETRANSMIT_TIME;
ch->root_rel->expected_delay.rel_value_us = 0;
LOG (GNUNET_ERROR_TYPE_DEBUG, "CREATED CHANNEL %s\n", GMCH_2s (ch));
if (NULL == c)
{
- GNUNET_break (0);
+ GNUNET_break (GNUNET_NO != ch->destroy);
return;
}
+ if (MESH_CHANNEL_READY != ch->state)
+ {
+ if (GNUNET_NO == fwd)
+ {
+ /* If we are the root, this means the other peer has sent traffic before
+ * receiving our ACK. Even if the SYNACK goes missing, no traffic should
+ * be sent before the ACK.
+ */
+ GNUNET_break_op (0);
+ return;
+ }
+ /* If we are the dest, this means that the SYNACK got to the root but
+ * the ACK went missing. Treat this as an ACK.
+ */
+ channel_confirm (ch, GNUNET_NO);
+ }
+
GNUNET_STATISTICS_update (stats, "# data received", 1, GNUNET_NO);
mid = ntohl (msg->mid);
( !GM_is_pid_bigger (rel->mid_recv, mid) &&
GM_is_pid_bigger (rel->mid_recv + 64, mid) ) )
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! RECV %u\n", mid);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! RECV %u (%u)\n",
+ mid, ntohs (msg->header.size));
if (GNUNET_YES == ch->reliable)
{
/* Is this the exact next expected messasge? */
}
if (NULL == rel)
{
- GNUNET_break_op (0);
+ GNUNET_break_op (GNUNET_NO != ch->destroy);
return;
}
- /* Free ACK'd copies: no need to retransmit those anymore */
+ /* Free ACK'd copies: no need to retransmit those anymore FIXME refactor */
for (work = GNUNET_NO, copy = rel->head_sent; copy != NULL; copy = next)
{
if (GM_is_pid_bigger (copy->mid, ack))
work = GNUNET_YES;
LOG (GNUNET_ERROR_TYPE_DEBUG, " !! id %u\n", copy->mid);
next = copy->next;
- rel_message_free (copy, GNUNET_YES);
+ if (GNUNET_YES == rel_message_free (copy, GNUNET_YES))
+ return;
}
- /* ACK client if needed */
+ /* ACK client if needed and possible */
GMCH_allow_client (ch, fwd);
/* If some message was free'd, update the retransmission delay */
if (GNUNET_SCHEDULER_NO_TASK != rel->retry_task)
{
GNUNET_SCHEDULER_cancel (rel->retry_task);
- if (NULL != rel->head_sent && NULL == rel->head_sent->q)
+ rel->retry_task = GNUNET_SCHEDULER_NO_TASK;
+ if (NULL != rel->head_sent && NULL == rel->head_sent->chq)
{
struct GNUNET_TIME_Absolute new_target;
struct GNUNET_TIME_Relative delay;
&channel_retransmit_message,
rel);
}
- else /* either no more traffic to ack or traffic has just been queued */
- {
- rel->retry_task = GNUNET_SCHEDULER_NO_TASK;
- }
}
- else /* work was done but no task was pending? shouldn't happen! */
+ else
{
+ /* Work was done but no task was pending? Shouldn't happen! */
GNUNET_break (0);
}
}
MESH_ChannelNumber chid;
struct MeshChannel *ch;
struct MeshClient *c;
+ int new_channel;
chid = ntohl (msg->chid);
-
ch = GMT_get_channel (t, chid);
if (NULL == ch)
{
/* Create channel */
ch = channel_new (t, NULL, 0);
ch->gid = chid;
+ channel_set_options (ch, ntohl (msg->opt));
+ new_channel = GNUNET_YES;
+ }
+ else
+ {
+ new_channel = GNUNET_NO;
}
- channel_set_options (ch, ntohl (msg->opt));
- /* Find a destination client */
- ch->port = ntohl (msg->port);
- LOG (GNUNET_ERROR_TYPE_DEBUG, " port %u\n", ch->port);
- c = GML_client_get_by_port (ch->port);
- if (NULL == c)
+ if (GNUNET_YES == new_channel || GMT_is_loopback (t))
{
- /* TODO send reject */
- LOG (GNUNET_ERROR_TYPE_DEBUG, " no client has port registered\n");
- if (is_loopback (ch))
+ /* Find a destination client */
+ ch->port = ntohl (msg->port);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " port %u\n", ch->port);
+ c = GML_client_get_by_port (ch->port);
+ if (NULL == c)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " loopback: destroy on handler\n");
- channel_send_nack (ch);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " no client has port registered\n");
+ if (is_loopback (ch))
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " loopback: destroy on handler\n");
+ send_nack (ch);
+ }
+ else
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " not loopback: destroy now\n");
+ send_nack (ch);
+ GMCH_destroy (ch);
+ }
+ return NULL;
}
else
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " not loopback: destroy now\n");
- channel_send_nack (ch);
- GMCH_destroy (ch);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " client %p has port registered\n", c);
}
- return NULL;
+
+ add_destination (ch, c);
+ if (GNUNET_YES == ch->reliable)
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! Reliable\n");
+ else
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! Not Reliable\n");
+
+ send_client_create (ch);
+ ch->state = MESH_CHANNEL_SENT;
}
else
{
- LOG (GNUNET_ERROR_TYPE_DEBUG, " client %p has port registered\n", c);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " duplicate create channel\n");
}
-
- add_destination (ch, c);
- if (GNUNET_YES == ch->reliable)
- LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! Reliable\n");
- else
- LOG (GNUNET_ERROR_TYPE_DEBUG, "!!! Not Reliable\n");
-
- send_client_create (ch);
- channel_send_ack (ch, GNUNET_YES);
+ send_ack (ch, GNUNET_YES);
return ch;
}
struct MeshChannel *ch, int fwd,
void *existing_copy)
{
+ struct MeshChannelQueue *chq;
uint16_t type;
type = ntohs (message->type);
if (GNUNET_YES == ch->reliable)
{
- struct MeshChannelQueue *q;
-
- q = GNUNET_new (struct MeshChannelQueue);
- q->type = type;
+ chq = GNUNET_new (struct MeshChannelQueue);
+ chq->type = type;
if (NULL == existing_copy)
- q->copy = channel_save_copy (ch, message, fwd);
+ chq->copy = channel_save_copy (ch, message, fwd);
else
{
- q->copy = (struct MeshReliableMessage *) existing_copy;
- if (NULL != q->copy->q)
+ chq->copy = (struct MeshReliableMessage *) existing_copy;
+ if (NULL != chq->copy->chq)
{
/* Last retransmission was queued but not yet sent!
* This retransmission was scheduled by a ch_message_sent which
* retransmission leaves the peer and ch_message_sent starts
* the timer for the next one.
*/
- GNUNET_free (q);
+ GNUNET_free (chq);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ " exisitng copy not yet transmitted!\n");
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
" using existing copy: %p {r:%p q:%p t:%u}\n",
existing_copy,
- q->copy->rel, q->copy->q, q->copy->type);
+ chq->copy->rel, chq->copy->chq, chq->copy->type);
}
- LOG (GNUNET_ERROR_TYPE_DEBUG, " new q: %p\n", q);
- q->copy->q = q;
- q->q = GMT_send_prebuilt_message (message, ch->t, ch,
- fwd, NULL != existing_copy,
- &ch_message_sent, q);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " new chq: %p\n", chq);
+ chq->copy->chq = chq;
+ chq->tq = GMT_send_prebuilt_message (message, ch->t,
+ NULL != existing_copy,
+ &ch_message_sent, chq);
/* q itself is stored in copy */
+ GNUNET_assert (NULL != chq->tq || GNUNET_NO != ch->destroy);
}
else
{
- GNUNET_break (NULL == GMT_send_prebuilt_message (message, ch->t, ch,
- fwd, GNUNET_NO,
- NULL, NULL));
+ fire_and_forget (message, ch, GNUNET_NO);
}
break;
-
+
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_ACK:
+ if (GNUNET_YES == fwd)
+ {
+ /* BCK ACK (going FWD) is just a response for a SYNACK, don't keep*/
+ fire_and_forget (message, ch, GNUNET_YES);
+ return;
+ }
+ /* fall-trough */
case GNUNET_MESSAGE_TYPE_MESH_DATA_ACK:
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
+ chq = GNUNET_new (struct MeshChannelQueue);
+ chq->type = type;
+ chq->rel = fwd ? ch->root_rel : ch->dest_rel;
+ if (NULL != chq->rel->uniq)
{
- struct MeshChannelReliability *rel;
-
- rel = fwd ? ch->root_rel : ch->dest_rel;
- if (NULL != rel->uniq)
+ if (NULL != chq->rel->uniq->tq)
+ {
+ GMT_cancel (chq->rel->uniq->tq);
+ /* ch_message_sent is called, freeing and NULLing uniq */
+ }
+ else
{
- GMT_cancel (rel->uniq->q);
- /* ch_message_sent is called, freeing ack_q */
+ GNUNET_break (0);
+ GNUNET_free (chq->rel->uniq);
}
- rel->uniq = GNUNET_new (struct MeshChannelQueue);
- rel->uniq->type = type;
- rel->uniq->rel = rel;
- rel->uniq->q = GMT_send_prebuilt_message (message, ch->t, ch,
- fwd, GNUNET_YES,
- &ch_message_sent,
- rel->uniq);
}
+ chq->tq = GMT_send_prebuilt_message (message, ch->t, GNUNET_YES,
+ &ch_message_sent, chq);
+ if (NULL == chq->tq)
+ {
+ GNUNET_break (0);
+ GNUNET_free (chq);
+ chq = NULL;
+ return;
+ }
+ chq->rel->uniq = chq;
+ break;
+
+
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
+ case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_NACK:
+ fire_and_forget (message, ch, GNUNET_YES);
break;
default:
- GNUNET_break (NULL == GMT_send_prebuilt_message (message, ch->t, ch,
- fwd, GNUNET_YES,
- NULL, NULL));
+ GNUNET_break (0);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "type %s unknown!\n", GM_m2s (type));
+ fire_and_forget (message, ch, GNUNET_YES);
}
}
projects / oweals / gnunet.git / blobdiff