-
/*
This file is part of GNUnet.
Copyright (C) 2001-2017 GNUnet e.V.
* @author Christian Grothoff
*
* TODO:
- * - Optimize ACKs by using 'mid_futures' properly!
- * - calculate current RTT if possible, use that for initial retransmissions
- * (NOTE: needs us to learn which connection the tunnel uses for the message!)
- * - introduce shutdown so we can have half-closed channels, modify
- * destroy to include MID to have FIN-ACK equivalents, etc.
- * - estimate max bandwidth using bursts and use to for CONGESTION CONTROL!
- * (and figure out how/where to use this!)
- * - check that '0xFFULL' really is sufficient for flow control!
- * (this is right now a big HACK!)
- * - revisit handling of 'unreliable' traffic!
- * (has not seen enough review)
+ * - Congestion/flow control:
+ * + calculate current RTT if possible, use that for initial retransmissions
+ * (NOTE: needs us to learn which connection the tunnel uses for the message!)
+ * + estimate max bandwidth using bursts and use to for CONGESTION CONTROL!
+ * (and figure out how/where to use this!)
+ * + figure out flow control without ACKs (unreliable traffic!)
* - revisit handling of 'unbuffered' traffic!
- * (has not seen enough review)
- * - revisit handling of 'out-of-order' option, especially in combination with/without 'reliable'.
- * - figure out flow control without ACKs (unreliable traffic!)
+ * (need to push down through tunnel into connection selection)
+ * - revisit handling of 'buffered' traffic: 4 is a rather small buffer; maybe
+ * reserve more bits in 'options' to allow for buffer size control?
*/
#include "platform.h"
#include "gnunet_util_lib.h"
*/
struct CadetTunnelQueueEntry *qe;
+ /**
+ * Data message we are trying to send.
+ */
+ struct GNUNET_CADET_ChannelAppDataMessage *data_message;
+
/**
* How soon should we retry if we fail to get an ACK?
* Messages in the queue are sorted by this value.
struct GNUNET_TIME_Relative retry_delay;
/**
- * Data message we are trying to send.
+ * Time when we first successfully transmitted the message
+ * (that is, set @e num_transmissions to 1).
*/
- struct GNUNET_CADET_ChannelAppDataMessage *data_message;
+ struct GNUNET_TIME_Absolute first_transmission_time;
+
+ /**
+ * Identifier of the connection that this message took when it
+ * was first transmitted. Only useful if @e num_transmissions is 1.
+ */
+ struct GNUNET_CADET_ConnectionTunnelIdentifier connection_taken;
+
+ /**
+ * How often was this message transmitted? #GNUNET_SYSERR if there
+ * was an error transmitting the message, #GNUNET_NO if it was not
+ * yet transmitted ever, otherwise the number of (re) transmissions.
+ */
+ int num_transmissions;
};
*/
struct GNUNET_CADET_ClientChannelNumber ccn;
+ /**
+ * Number of entries currently in @a head_recv DLL.
+ */
+ unsigned int num_recv;
+
/**
* Can we send data to the client?
*/
/**
* Bitfield of already-received messages past @e mid_recv.
- *
- * FIXME: not yet properly used (bits here are never set!)
*/
uint64_t mid_futures;
*/
enum CadetChannelState state;
+ /**
+ * Count how many ACKs we skipped, used to prevent long
+ * sequences of ACK skipping.
+ */
+ unsigned int skip_ack_series;
+
/**
* Is the tunnel bufferless (minimum latency)?
*/
GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
ccc->tail_recv,
com);
+ ccc->num_recv--;
GNUNET_MQ_discard (com->env);
GNUNET_free (com);
}
* create message. Delays for a bit until we retry.
*
* @param cls our `struct CadetChannel`.
+ * @param cid identifier of the connection within the tunnel, NULL
+ * if transmission failed
*/
static void
-channel_open_sent_cb (void *cls)
+channel_open_sent_cb (void *cls,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
{
struct CadetChannel *ch = cls;
* ACKs for ACKs ;-).
*
* @param cls our `struct CadetChannel`.
+ * @param cid identifier of the connection within the tunnel, NULL
+ * if transmission failed
*/
static void
-send_ack_cb (void *cls)
+send_ack_cb (void *cls,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
{
struct CadetChannel *ch = cls;
{
struct GNUNET_CADET_ChannelDataAckMessage msg;
+ if (GNUNET_NO == ch->reliable)
+ return; /* no ACKs */
msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK);
msg.header.size = htons (sizeof (msg));
msg.ctn = ch->ctn;
- msg.mid.mid = htonl (ntohl (ch->mid_recv.mid) - 1);
+ msg.mid.mid = htonl (ntohl (ch->mid_recv.mid));
msg.futures = GNUNET_htonll (ch->mid_futures);
if (NULL != ch->last_control_qe)
GCT_send_cancel (ch->last_control_qe);
* #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
*
* @param ch channel that got the duplicate open
+ * @param cti identifier of the connection that delivered the message
*/
void
-GCCH_handle_duplicate_open (struct CadetChannel *ch)
+GCCH_handle_duplicate_open (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
{
if (NULL == ch->dest)
{
if (GNUNET_YES == ch->is_loopback)
{
ch->state = CADET_CHANNEL_OPEN_SENT;
- GCCH_handle_channel_open_ack (ch);
+ GCCH_handle_channel_open_ack (ch,
+ NULL);
}
else
{
* (the port is open on the other side). Begin transmissions.
*
* @param ch channel to destroy
+ * @param cti identifier of the connection that delivered the message
*/
void
-GCCH_handle_channel_open_ack (struct CadetChannel *ch)
+GCCH_handle_channel_open_ack (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
{
switch (ch->state)
{
* Test if element @a e1 comes before element @a e2.
*
* @param cls closure, to a flag where we indicate duplicate packets
- * @param e1 an element of to sort
- * @param e2 another element to sort
+ * @param m1 a message of to sort
+ * @param m2 another message to sort
* @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
*/
static int
}
else
{
- /* result is small, thus v2 > v1, thus e1 < e2 */
+ /* result is small, thus v2 > v1, thus m1 < m2 */
return GNUNET_YES;
}
}
* and send an ACK to the other end (once flow control allows it!)
*
* @param ch channel that got data
+ * @param cti identifier of the connection that delivered the message
* @param msg message that was received
*/
void
GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
const struct GNUNET_CADET_ChannelAppDataMessage *msg)
{
struct GNUNET_MQ_Envelope *env;
uint32_t mid_min;
uint32_t mid_max;
uint32_t mid_msg;
+ uint32_t delta;
GNUNET_assert (GNUNET_NO == ch->is_loopback);
if ( (GNUNET_YES == ch->destroy) &&
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Dropping incoming payload on %s as this end is already closed\n",
GCCH_2s (ch));
- /* FIXME: send back ACK/NACK/Closed notification
- to stop retransmissions! */
+ /* send back DESTROY notification to stop further retransmissions! */
+ GCT_send_channel_destroy (ch->t,
+ ch->ctn);
return;
}
payload_size = ntohs (msg->header.size) - sizeof (*msg);
env);
ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
ch->mid_futures >>= 1;
- if (GNUNET_YES == ch->reliable)
- send_channel_data_ack (ch);
+ send_channel_data_ack (ch);
return;
}
- /* check if message ought to be dropped because it is anicent/too distant/duplicate */
- mid_min = ntohl (ch->mid_recv.mid);
- mid_max = mid_min + MAX_OUT_OF_ORDER_DISTANCE;
- mid_msg = ntohl (msg->mid.mid);
- if ( ( (uint32_t) (mid_msg - mid_min) > MAX_OUT_OF_ORDER_DISTANCE) ||
- ( (uint32_t) (mid_max - mid_msg) > MAX_OUT_OF_ORDER_DISTANCE) )
+ if (GNUNET_YES == ch->reliable)
{
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Duplicate ancient or future payload of %u bytes on %s (mid %u) dropped\n",
- (unsigned int) payload_size,
- GCCH_2s (ch),
- ntohl (msg->mid.mid));
- GNUNET_STATISTICS_update (stats,
- "# duplicate DATA (ancient or future)",
- 1,
- GNUNET_NO);
- GNUNET_MQ_discard (env);
- return;
+ /* check if message ought to be dropped because it is ancient/too distant/duplicate */
+ mid_min = ntohl (ch->mid_recv.mid);
+ mid_max = mid_min + ch->max_pending_messages;
+ mid_msg = ntohl (msg->mid.mid);
+ if ( ( (uint32_t) (mid_msg - mid_min) > ch->max_pending_messages) ||
+ ( (uint32_t) (mid_max - mid_msg) > ch->max_pending_messages) )
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "%s at %u drops ancient or far-future message %u\n",
+ GCCH_2s (ch),
+ (unsigned int) mid_min,
+ ntohl (msg->mid.mid));
+
+ GNUNET_STATISTICS_update (stats,
+ "# duplicate DATA (ancient or future)",
+ 1,
+ GNUNET_NO);
+ GNUNET_MQ_discard (env);
+ send_channel_data_ack (ch);
+ return;
+ }
+ /* mark bit for future ACKs */
+ delta = mid_msg - mid_min - 1; /* overflow/underflow are OK here */
+ if (delta < 64)
+ {
+ if (0 != (ch->mid_futures & (1LLU << delta)))
+ {
+ /* Duplicate within the queue, drop also */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
+ (unsigned int) payload_size,
+ GCCH_2s (ch),
+ ntohl (msg->mid.mid));
+ GNUNET_STATISTICS_update (stats,
+ "# duplicate DATA",
+ 1,
+ GNUNET_NO);
+ GNUNET_MQ_discard (env);
+ send_channel_data_ack (ch);
+ return;
+ }
+ ch->mid_futures |= (1LLU << delta);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Marked bit %llX for mid %u (base: %u); now: %llX\n",
+ (1LLU << delta),
+ mid_msg,
+ mid_min,
+ ch->mid_futures);
+ }
+ }
+ else /* ! ch->reliable */
+ {
+ /* Channel is unreliable, so we do not ACK. But we also cannot
+ allow buffering everything, so check if we have space... */
+ if (ccc->num_recv >= ch->max_pending_messages)
+ {
+ struct CadetOutOfOrderMessage *drop;
+
+ /* Yep, need to drop. Drop the oldest message in
+ the buffer. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Queue full due slow client on %s, dropping oldest message\n",
+ GCCH_2s (ch));
+ GNUNET_STATISTICS_update (stats,
+ "# messages dropped due to slow client",
+ 1,
+ GNUNET_NO);
+ drop = ccc->head_recv;
+ GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
+ ccc->tail_recv,
+ drop);
+ ccc->num_recv--;
+ GNUNET_MQ_discard (drop->env);
+ GNUNET_free (drop);
+ }
}
/* Insert message into sorted out-of-order queue */
ccc->head_recv,
ccc->tail_recv,
com);
+ ccc->num_recv++;
if (GNUNET_YES == duplicate)
{
- /* Duplicate within the queue, drop also */
+ /* Duplicate within the queue, drop also (this is not covered by
+ the case above if "delta" >= 64, which could be the case if
+ max_pending_messages is also >= 64 or if our client is unready
+ and we are seeing retransmissions of the message our client is
+ blocked on. */
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Duplicate payload of %u bytes on %s (mid %u) dropped\n",
(unsigned int) payload_size,
GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
ccc->tail_recv,
com);
+ ccc->num_recv--;
GNUNET_MQ_discard (com->env);
GNUNET_free (com);
- if (GNUNET_YES == ch->reliable)
- send_channel_data_ack (ch);
+ send_channel_data_ack (ch);
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
ccc,
ntohl (msg->mid.mid),
ntohl (ch->mid_recv.mid));
+ /* NOTE: this ACK we _could_ skip, as the packet is out-of-order and
+ the sender may already be transmitting the previous one. Needs
+ experimental evaluation to see if/when this ACK helps or
+ hurts. (We might even want another option.) */
send_channel_data_ack (ch);
}
* wait for ACK (or retransmit).
*
* @param cls the `struct CadetReliableMessage` that was sent
+ * @param cid identifier of the connection within the tunnel, NULL
+ * if transmission failed
*/
static void
-data_sent_cb (void *cls);
+data_sent_cb (void *cls,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid);
/**
* the queue and tell our client that it can send more.
*
* @param ch the channel that got the PLAINTEXT_DATA_ACK
+ * @param cti identifier of the connection that delivered the message
* @param crm the message that got acknowledged
*/
static void
handle_matching_ack (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
struct CadetReliableMessage *crm)
{
GNUNET_CONTAINER_DLL_remove (ch->head_sent,
GCT_send_cancel (crm->qe);
crm->qe = NULL;
}
+ if ( (1 == crm->num_transmissions) &&
+ (NULL != cti) )
+ {
+ GCC_ack_observed (cti);
+ if (0 == memcmp (cti,
+ &crm->connection_taken,
+ sizeof (struct GNUNET_CADET_ConnectionTunnelIdentifier)))
+ {
+ GCC_latency_observed (cti,
+ GNUNET_TIME_absolute_get_duration (crm->first_transmission_time));
+ }
+ }
GNUNET_free (crm->data_message);
GNUNET_free (crm);
send_ack_to_client (ch,
* Possibly resume transmissions.
*
* @param ch channel that got the ack
+ * @param cti identifier of the connection that delivered the message
* @param ack details about what was received
*/
void
GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
const struct GNUNET_CADET_ChannelDataAckMessage *ack)
{
struct CadetReliableMessage *crm;
GNUNET_break_op (0);
return;
}
+ /* mid_base is the MID of the next message that the
+ other peer expects (i.e. that is missing!), everything
+ LOWER (but excluding mid_base itself) was received. */
mid_base = ntohl (ack->mid.mid);
mid_mask = GNUNET_htonll (ack->futures);
found = GNUNET_NO;
crm = crmn)
{
crmn = crm->next;
- if (ack->mid.mid == crm->data_message->mid.mid)
+ delta = (unsigned int) (ntohl (crm->data_message->mid.mid) - mid_base);
+ if (delta >= UINT_MAX - ch->max_pending_messages)
{
+ /* overflow, means crm was a bit in the past, so this ACK counts for it. */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got DATA_ACK with base %u satisfying past message %u on %s\n",
+ (unsigned int) mid_base,
+ ntohl (crm->data_message->mid.mid),
+ GCCH_2s (ch));
handle_matching_ack (ch,
+ cti,
crm);
found = GNUNET_YES;
continue;
}
- delta = (unsigned int) (ntohl (crm->data_message->mid.mid) - mid_base) - 1;
+ delta--;
if (delta >= 64)
continue;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Testing bit %llX for mid %u (base: %u)\n",
+ (1LLU << delta),
+ ntohl (crm->data_message->mid.mid),
+ mid_base);
if (0 != (mid_mask & (1LLU << delta)))
{
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got DATA_ACK with mask for %u on %s\n",
+ ntohl (crm->data_message->mid.mid),
+ GCCH_2s (ch));
handle_matching_ack (ch,
+ cti,
crm);
found = GNUNET_YES;
}
GNUNET_SCHEDULER_cancel (ch->retry_data_task);
ch->retry_data_task = NULL;
}
- if (NULL != ch->head_sent)
+ if ( (NULL != ch->head_sent) &&
+ (NULL == ch->head_sent->qe) )
ch->retry_data_task
= GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
&retry_transmission,
* the tunnel.
*
* @param ch channel to destroy
+ * @param cti identifier of the connection that delivered the message,
+ * NULL if we are simulating receiving a destroy due to shutdown
*/
void
-GCCH_handle_remote_destroy (struct CadetChannel *ch)
+GCCH_handle_remote_destroy (struct CadetChannel *ch,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
{
struct CadetChannelClient *ccc;
* wait for ACK (or retransmit).
*
* @param cls the `struct CadetReliableMessage` that was sent
+ * @param cid identifier of the connection within the tunnel, NULL
+ * if transmission failed
*/
static void
-data_sent_cb (void *cls)
+data_sent_cb (void *cls,
+ const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
{
struct CadetReliableMessage *crm = cls;
struct CadetChannel *ch = crm->ch;
: GNUNET_YES);
return;
}
+ if (NULL == cid)
+ {
+ /* There was an error sending. */
+ crm->num_transmissions = GNUNET_SYSERR;
+ }
+ else if (GNUNET_SYSERR != crm->num_transmissions)
+ {
+ /* Increment transmission counter, and possibly store @a cid
+ if this was the first transmission. */
+ crm->num_transmissions++;
+ if (1 == crm->num_transmissions)
+ {
+ crm->first_transmission_time = GNUNET_TIME_absolute_get ();
+ crm->connection_taken = *cid;
+ GCC_ack_expected (cid);
+ }
+ }
if (0 == crm->retry_delay.rel_value_us)
crm->retry_delay = ch->expected_delay;
else
GCCH_2s (ch),
GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (ch->head_sent->next_retry),
GNUNET_YES));
- if (crm == ch->head_sent)
+ if (NULL == ch->head_sent->qe)
{
- /* We are the new head, need to reschedule retry task */
if (NULL != ch->retry_data_task)
GNUNET_SCHEDULER_cancel (ch->retry_data_task);
ch->retry_data_task
GNUNET_CONTAINER_DLL_insert_tail (receiver->head_recv,
receiver->tail_recv,
oom);
+ receiver->num_recv++;
}
return GNUNET_OK;
}
GNUNET_memcpy (&crm->data_message[1],
buf,
buf_len);
- GNUNET_CONTAINER_DLL_insert (ch->head_sent,
- ch->tail_sent,
- crm);
+ GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent,
+ ch->tail_sent,
+ crm);
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sending %u bytes from local client to %s with MID %u\n",
- buf_len,
+ "Sending message %u from local client to %s with %u bytes\n",
+ ntohl (crm->data_message->mid.mid),
GCCH_2s (ch),
- ntohl (crm->data_message->mid.mid));
+ buf_len);
if (NULL != ch->retry_data_task)
{
GNUNET_SCHEDULER_cancel (ch->retry_data_task);
GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
ccc->tail_recv,
com);
+ ccc->num_recv--;
GSC_send_to_client (ccc->c,
com->env);
/* Notify sender that we can receive more */
}
if ( (com->mid.mid != ch->mid_recv.mid) &&
- (GNUNET_NO == ch->out_of_order) )
+ (GNUNET_NO == ch->out_of_order) &&
+ (GNUNET_YES == ch->reliable) )
{
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Got LOCAL_ACK, %s-%X ready to receive more data (but next one is out-of-order %u vs. %u)!\n",
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Got LOCAL ACK, passing payload message %u to %s-%X on %s\n",
+ "Got LOCAL_ACK, giving payload message %u to %s-%X on %s\n",
ntohl (com->mid.mid),
GSC_2s (ccc->c),
ntohl (ccc->ccn.channel_of_client),
GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
ccc->tail_recv,
com);
+ ccc->num_recv--;
/* FIXME: if unreliable, this is not aggressive
enough, as it would be OK to have lost some! */
+
ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
ch->mid_futures >>= 1; /* equivalent to division by 2 */
ccc->client_ready = GNUNET_NO;
GSC_send_to_client (ccc->c,
com->env);
GNUNET_free (com);
- if ( (0xFFULL == (ch->mid_futures & 0xFFULL)) &&
- (GNUNET_YES == ch->reliable) )
- {
- /* The next 15 messages were also already received (0xFF), this
- suggests that the sender may be blocked on flow control
- urgently waiting for an ACK from us. (As we have an inherent
- maximum of 64 bits, and 15 is getting too close for comfort.)
- So we should send one now. */
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sender on %s likely blocked on flow-control, sending ACK now.\n",
- GCCH_2s (ch));
- if (GNUNET_YES == ch->reliable)
- send_channel_data_ack (ch);
- }
-
+ send_channel_data_ack (ch);
if (NULL != ccc->head_recv)
return;
if (GNUNET_NO == ch->destroy)
projects / oweals / gnunet.git / blobdiff