-
/*
This file is part of GNUnet.
Copyright (C) 2001-2017 GNUnet e.V.
*
* TODO:
* - 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!)
/**
- * All the states a connection can be in.
+ * All the states a channel can be in.
*/
enum CadetChannelState
{
CADET_CHANNEL_NEW,
/**
- * Connection create message sent, waiting for ACK.
+ * Channel is to a port that is not open, we're waiting for the
+ * port to be opened.
+ */
+ CADET_CHANNEL_LOOSE,
+
+ /**
+ * CHANNEL_OPEN message sent, waiting for CHANNEL_OPEN_ACK.
*/
CADET_CHANNEL_OPEN_SENT,
*/
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_TIME_Relative retry_time;
- /**
- * How long does it usually take to get an ACK.
- */
- struct GNUNET_TIME_Relative expected_delay;
-
/**
* Bitfield of already-received messages past @e mid_recv.
*/
* 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;
msgcc.port = ch->port;
msgcc.ctn = ch->ctn;
ch->state = CADET_CHANNEL_OPEN_SENT;
+ if (NULL != ch->last_control_qe)
+ GCT_send_cancel (ch->last_control_qe);
ch->last_control_qe = GCT_send (ch->t,
&msgcc.header,
&channel_open_sent_cb,
if (NULL == c)
{
/* port closed, wait for it to possibly open */
+ ch->state = CADET_CHANNEL_LOOSE;
(void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
port,
ch,
}
else
{
- ch->dest = GNUNET_new (struct CadetChannelClient);
- ch->dest->c = c;
- ch->dest->client_ready = GNUNET_YES;
GCCH_bind (ch,
- ch->dest->c);
+ c);
}
}
else
if (NULL == c)
{
/* port closed, wait for it to possibly open */
+ ch->state = CADET_CHANNEL_LOOSE;
(void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
port,
ch,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ GNUNET_assert (NULL == ch->retry_control_task);
ch->retry_control_task
= GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
&timeout_closed_cb,
* 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;
msg.ctn = ch->ctn;
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);
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Sending DATA_ACK %u:%llX via %s\n",
(unsigned int) ntohl (msg.mid.mid),
(unsigned long long) ch->mid_futures,
GCCH_2s (ch));
+ if (NULL != ch->last_control_qe)
+ GCT_send_cancel (ch->last_control_qe);
ch->last_control_qe = GCT_send (ch->t,
&msg.header,
&send_ack_cb,
struct CadetChannel *ch = cls;
struct GNUNET_CADET_ChannelManageMessage msg;
+ ch->retry_control_task = NULL;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Sending CHANNEL_OPEN_ACK on %s\n",
GCCH_2s (ch));
- ch->retry_control_task = NULL;
msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
msg.header.size = htons (sizeof (msg));
msg.reserved = htonl (0);
* #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)
{
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Ignoring duplicate channel OPEN on %s: port is closed\n",
+ "Ignoring duplicate CHANNEL_OPEN on %s: port is closed\n",
GCCH_2s (ch));
return;
}
if (NULL != ch->retry_control_task)
{
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Ignoring duplicate channel OPEN on %s: control message is pending\n",
+ "Ignoring duplicate CHANNEL_OPEN on %s: control message is pending\n",
GCCH_2s (ch));
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Retransmitting OPEN_ACK on %s\n",
+ "Retransmitting CHANNEL_OPEN_ACK on %s\n",
GCCH_2s (ch));
ch->retry_control_task
= GNUNET_SCHEDULER_add_now (&send_open_ack,
if (ch->out_of_order)
options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
cccd = GNUNET_new (struct CadetChannelClient);
+ GNUNET_assert (NULL == ch->dest);
ch->dest = cccd;
cccd->c = c;
cccd->client_ready = GNUNET_YES;
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
{
/* notify other peer that we accepted the connection */
+ ch->state = CADET_CHANNEL_READY;
ch->retry_control_task
= GNUNET_SCHEDULER_add_now (&send_open_ack,
ch);
}
+/**
+ * One of our clients has disconnected, tell the other one that we
+ * are finished. Done asynchronously to avoid concurrent modification
+ * issues if this is the same client.
+ *
+ * @param cls the `struct CadetChannel` where one of the ends is now dead
+ */
+static void
+signal_remote_destroy_cb (void *cls)
+{
+ struct CadetChannel *ch = cls;
+ struct CadetChannelClient *ccc;
+
+ /* Find which end is left... */
+ ch->retry_control_task = NULL;
+ ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
+ GSC_handle_remote_channel_destroy (ccc->c,
+ ccc->ccn,
+ ch);
+ channel_destroy (ch);
+}
+
+
/**
* Destroy locally created channel. Called by the local client, so no
* need to tell the client.
channel_destroy (ch);
return;
}
- if ( (NULL != ch->head_sent) ||
- (NULL != ch->owner) ||
- (NULL != ch->dest) )
+ if ( (NULL != ch->head_sent) &&
+ ( (NULL != ch->owner) ||
+ (NULL != ch->dest) ) )
{
/* Wait for other end to destroy us as well,
and otherwise allow send queue to be transmitted first */
ch->destroy = GNUNET_YES;
return;
}
- /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
- if (CADET_CHANNEL_NEW != ch->state)
- GCT_send_channel_destroy (ch->t,
- ch->ctn);
+ if ( (GNUNET_YES == ch->is_loopback) &&
+ ( (NULL != ch->owner) ||
+ (NULL != ch->dest) ) )
+ {
+ if (NULL != ch->retry_control_task)
+ GNUNET_SCHEDULER_cancel (ch->retry_control_task);
+ ch->retry_control_task
+ = GNUNET_SCHEDULER_add_now (&signal_remote_destroy_cb,
+ ch);
+ return;
+ }
+ if (GNUNET_NO == ch->is_loopback)
+ {
+ /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
+ switch (ch->state)
+ {
+ case CADET_CHANNEL_NEW:
+ /* We gave up on a channel that we created as a client to a remote
+ target, but that never went anywhere. Nothing to do here. */
+ break;
+ case CADET_CHANNEL_LOOSE:
+ GSC_drop_loose_channel (&ch->port,
+ ch);
+ break;
+ default:
+ GCT_send_channel_destroy (ch->t,
+ ch->ctn);
+ }
+ }
/* Nothing left to do, just finish destruction */
channel_destroy (ch);
}
* (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)
{
/* this should be impossible */
GNUNET_break (0);
break;
+ case CADET_CHANNEL_LOOSE:
+ /* This makes no sense. */
+ GNUNET_break_op (0);
+ break;
case CADET_CHANNEL_OPEN_SENT:
if (NULL == ch->owner)
{
* 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;
/* 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,
* 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;
ntohl (crm->data_message->mid.mid),
GCCH_2s (ch));
handle_matching_ack (ch,
+ cti,
crm);
found = GNUNET_YES;
continue;
ntohl (crm->data_message->mid.mid),
GCCH_2s (ch));
handle_matching_ack (ch,
+ cti,
crm);
found = GNUNET_YES;
}
* 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;
return;
}
ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
- if (NULL != ccc->head_recv)
+ if ( (NULL != ccc) &&
+ (NULL != ccc->head_recv) )
{
LOG (GNUNET_ERROR_TYPE_WARNING,
"Lost end of transmission due to remote shutdown on %s\n",
/* FIXME: change API to notify client about truncated transmission! */
}
ch->destroy = GNUNET_YES;
- GSC_handle_remote_channel_destroy (ccc->c,
- ccc->ccn,
- ch);
+ if (NULL != ccc)
+ GSC_handle_remote_channel_destroy (ccc->c,
+ ccc->ccn,
+ ch);
channel_destroy (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)
{
struct CadetReliableMessage *crm = cls;
struct CadetChannel *ch = crm->ch;
: GNUNET_YES);
return;
}
- if (0 == crm->retry_delay.rel_value_us)
- crm->retry_delay = ch->expected_delay;
- else
- crm->retry_delay = GNUNET_TIME_STD_BACKOFF (crm->retry_delay);
+ 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) &&
+ (NULL != cid) )
+ {
+ struct CadetConnection *cc = GCC_lookup (cid);
+
+ if (NULL != cc)
+ crm->retry_delay = GCC_get_metrics (cc)->aged_latency;
+ else
+ crm->retry_delay = ch->retry_time;
+ }
+ crm->retry_delay = GNUNET_TIME_STD_BACKOFF (crm->retry_delay);
crm->retry_delay = GNUNET_TIME_relative_max (crm->retry_delay,
MIN_RTT_DELAY);
crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
GNUNET_break (0);
return GNUNET_SYSERR;
}
+ if (GNUNET_YES == ch->destroy)
+ {
+ /* we are going down, drop messages */
+ return GNUNET_OK;
+ }
ch->pending_messages++;
if (GNUNET_YES == ch->is_loopback)
struct CadetChannelClient *receiver;
struct GNUNET_MQ_Envelope *env;
struct GNUNET_CADET_LocalData *ld;
- int to_owner;
+ int ack_to_owner;
env = GNUNET_MQ_msg_extra (ld,
buf_len,
GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
- if (sender_ccn.channel_of_client ==
- ch->owner->ccn.channel_of_client)
+ if ( (NULL != ch->owner) &&
+ (sender_ccn.channel_of_client ==
+ ch->owner->ccn.channel_of_client) )
{
receiver = ch->dest;
- to_owner = GNUNET_NO;
+ ack_to_owner = GNUNET_YES;
}
- else
+ else if ( (NULL != ch->dest) &&
+ (sender_ccn.channel_of_client ==
+ ch->dest->ccn.channel_of_client) )
{
- GNUNET_assert (sender_ccn.channel_of_client ==
- ch->dest->ccn.channel_of_client);
receiver = ch->owner;
- to_owner = GNUNET_YES;
+ ack_to_owner = GNUNET_NO;
+ }
+ else
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
}
ld->ccn = receiver->ccn;
GNUNET_memcpy (&ld[1],
buf_len);
if (GNUNET_YES == receiver->client_ready)
{
+ ch->pending_messages--;
GSC_send_to_client (receiver->c,
env);
send_ack_to_client (ch,
- to_owner);
+ ack_to_owner);
}
else
{
}
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",
projects / oweals / gnunet.git / blobdiff