* @author Christian Grothoff
*
* TODO:
+ * - FIXME: send ACKs back to loopback clients!
+ *
* - 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!
/**
* Data message we are trying to send.
*/
- struct GNUNET_CADET_ChannelAppDataMessage data_message;
+ struct GNUNET_CADET_ChannelAppDataMessage *data_message;
- /* followed by variable-size payload */
};
};
+/**
+ * Client endpoint of a `struct CadetChannel`. A channel may be a
+ * loopback channel, in which case it has two of these endpoints.
+ * Note that flow control also is required in both directions.
+ */
+struct CadetChannelClient
+{
+ /**
+ * Client handle. Not by itself sufficient to designate
+ * the client endpoint, as the same client handle may
+ * be used for both the owner and the destination, and
+ * we thus also need the channel ID to identify the client.
+ */
+ struct CadetClient *c;
+
+ /**
+ * Head of DLL of messages received out of order or while client was unready.
+ */
+ struct CadetOutOfOrderMessage *head_recv;
+
+ /**
+ * Tail DLL of messages received out of order or while client was unready.
+ */
+ struct CadetOutOfOrderMessage *tail_recv;
+
+ /**
+ * Local tunnel number for this client.
+ * (if owner >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI,
+ * otherwise < #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
+ */
+ struct GNUNET_CADET_ClientChannelNumber ccn;
+
+ /**
+ * Can we send data to the client?
+ */
+ int client_ready;
+
+};
+
+
/**
* Struct containing all information regarding a channel to a remote client.
*/
* Client owner of the tunnel, if any.
* (Used if this channel represends the initiating end of the tunnel.)
*/
- struct CadetClient *owner;
+ struct CadetChannelClient *owner;
/**
* Client destination of the tunnel, if any.
* (Used if this channel represents the listening end of the tunnel.)
*/
- struct CadetClient *dest;
+ struct CadetChannelClient *dest;
/**
* Last entry in the tunnel's queue relating to control messages
*/
struct CadetReliableMessage *tail_sent;
- /**
- * Head of DLL of messages received out of order or while client was unready.
- */
- struct CadetOutOfOrderMessage *head_recv;
-
- /**
- * Tail DLL of messages received out of order or while client was unready.
- */
- struct CadetOutOfOrderMessage *tail_recv;
-
/**
* Task to resend/poll in case no ACK is received.
*/
*/
struct GNUNET_CADET_ChannelTunnelNumber ctn;
- /**
- * Local tunnel number for local client owning the channel.
- * ( >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI or 0 )
- */
- struct GNUNET_CADET_ClientChannelNumber ccn;
-
/**
* Channel state.
*/
enum CadetChannelState state;
- /**
- * Can we send data to the client?
- */
- int client_ready;
-
- /**
- * Can the client send data to us?
- */
- int client_allowed;
-
/**
* Is the tunnel bufferless (minimum latency)?
*/
*/
int out_of_order;
+ /**
+ * Is this channel a loopback channel, where the destination is us again?
+ */
+ int is_loopback;
+
/**
* Flag to signal the destruction of the channel. If this is set to
* #GNUNET_YES the channel will be destroyed once the queue is
GNUNET_snprintf (buf,
sizeof (buf),
- "Channel %s:%s ctn:%X(%X)",
- GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
+ "Channel %s:%s ctn:%X(%X/%X)",
+ (GNUNET_YES == ch->is_loopback)
+ ? "loopback"
+ : GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
GNUNET_h2s (&ch->port),
ch->ctn,
- ntohl (ch->ccn.channel_of_client));
+ (NULL == ch->owner) ? 0 : ntohl (ch->owner->ccn.channel_of_client),
+ (NULL == ch->dest) ? 0 : ntohl (ch->dest->ccn.channel_of_client));
return buf;
}
}
+/**
+ * Release memory associated with @a ccc
+ *
+ * @param ccc data structure to clean up
+ */
+static void
+free_channel_client (struct CadetChannelClient *ccc)
+{
+ struct CadetOutOfOrderMessage *com;
+
+ while (NULL != (com = ccc->head_recv))
+ {
+ GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
+ ccc->tail_recv,
+ com);
+ GNUNET_MQ_discard (com->env);
+ GNUNET_free (com);
+ }
+ GNUNET_free (ccc);
+}
+
+
/**
* Destroy the given channel.
*
channel_destroy (struct CadetChannel *ch)
{
struct CadetReliableMessage *crm;
- struct CadetOutOfOrderMessage *com;
while (NULL != (crm = ch->head_sent))
{
GNUNET_CONTAINER_DLL_remove (ch->head_sent,
ch->tail_sent,
crm);
+ GNUNET_free (crm->data_message);
GNUNET_free (crm);
}
- while (NULL != (com = ch->head_recv))
+ if (NULL != ch->owner)
{
- GNUNET_CONTAINER_DLL_remove (ch->head_recv,
- ch->tail_recv,
- com);
- GNUNET_MQ_discard (com->env);
- GNUNET_free (com);
+ free_channel_client (ch->owner);
+ ch->owner = NULL;
+ }
+ if (NULL != ch->dest)
+ {
+ free_channel_client (ch->dest);
+ ch->dest = NULL;
}
if (NULL != ch->last_control_qe)
{
GNUNET_SCHEDULER_cancel (ch->retry_control_task);
ch->retry_control_task = NULL;
}
- GCT_remove_channel (ch->t,
- ch,
- ch->ctn);
+ if (GNUNET_NO == ch->is_loopback)
+ {
+ GCT_remove_channel (ch->t,
+ ch,
+ ch->ctn);
+ ch->t = NULL;
+ }
GNUNET_free (ch);
}
ch->last_control_qe = NULL;
ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sent CHANNEL_OPEN on %s, retrying in %s\n",
+ "Sent CADET_CHANNEL_OPEN on %s, retrying in %s\n",
GCCH_2s (ch),
GNUNET_STRINGS_relative_time_to_string (ch->retry_time,
GNUNET_YES));
uint32_t options)
{
struct CadetChannel *ch;
+ struct CadetChannelClient *ccco;
+
+ ccco = GNUNET_new (struct CadetChannelClient);
+ ccco->c = owner;
+ ccco->ccn = ccn;
+ ccco->client_ready = GNUNET_YES;
ch = GNUNET_new (struct CadetChannel);
+ ch->mid_recv.mid = htonl (1); /* The OPEN_ACK counts as message 0! */
ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
- ch->max_pending_messages = (ch->nobuffer) ? 1 : 32; /* FIXME: 32!? Do not hardcode! */
- ch->owner = owner;
- ch->ccn = ccn;
+ ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
+ ch->owner = ccco;
ch->port = *port;
- ch->t = GCP_get_tunnel (destination,
- GNUNET_YES);
- ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
- ch->ctn = GCT_add_channel (ch->t,
- ch);
+ if (0 == memcmp (&my_full_id,
+ GCP_get_id (destination),
+ sizeof (struct GNUNET_PeerIdentity)))
+ {
+ struct CadetClient *c;
+
+ ch->is_loopback = GNUNET_YES;
+ c = GNUNET_CONTAINER_multihashmap_get (open_ports,
+ port);
+ if (NULL == c)
+ {
+ /* port closed, wait for it to possibly open */
+ (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
+ port,
+ ch,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Created loose incoming loopback channel to port %s\n",
+ GNUNET_h2s (&ch->port));
+ }
+ else
+ {
+ ch->dest = GNUNET_new (struct CadetChannelClient);
+ ch->dest->c = c;
+ ch->dest->client_ready = GNUNET_YES;
+ GCCH_bind (ch,
+ ch->dest->c);
+ }
+ }
+ else
+ {
+ ch->t = GCP_get_tunnel (destination,
+ GNUNET_YES);
+ ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
+ ch->ctn = GCT_add_channel (ch->t,
+ ch);
+ }
GNUNET_STATISTICS_update (stats,
"# channels",
1,
GNUNET_h2s (port),
GCP_2s (destination),
GSC_2s (owner),
- GCT_2s (ch->t));
+ (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
return ch;
}
ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
- ch->max_pending_messages = (ch->nobuffer) ? 1 : 32; /* FIXME: 32!? Do not hardcode! */
+ ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
GNUNET_STATISTICS_update (stats,
"# channels",
1,
/**
- * Compute and send the current ACK to the other peer.
+ * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
*
- * @param ch channel to send the ACK for
+ * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
*/
static void
send_channel_data_ack (struct CadetChannel *ch)
/**
- * Send our initial ACK to the client confirming that the
+ * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
* connection is up.
*
* @param cls the `struct CadetChannel`
struct GNUNET_CADET_ChannelManageMessage msg;
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sending CHANNEL_OPEN_ACK on channel %s\n",
+ "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);
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Retransmitting OPEN_ACK on channel %s\n",
+ "Retransmitting OPEN_ACK on %s\n",
GCCH_2s (ch));
ch->retry_control_task
= GNUNET_SCHEDULER_add_now (&send_open_ack,
/**
- * Send a LOCAL ACK to the client to solicit more messages.
+ * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK to the client to solicit more messages.
*
* @param ch channel the ack is for
- * @param c client to send the ACK to
+ * @param to_owner #GNUNET_YES to send to owner,
+ * #GNUNET_NO to send to dest
*/
static void
send_ack_to_client (struct CadetChannel *ch,
- struct CadetClient *c)
+ int to_owner)
{
struct GNUNET_MQ_Envelope *env;
struct GNUNET_CADET_LocalAck *ack;
+ struct CadetChannelClient *ccc;
env = GNUNET_MQ_msg (ack,
GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
- ack->ccn = ch->ccn;
- GSC_send_to_client (c,
+ ccc = (GNUNET_YES == to_owner) ? ch->owner : ch->dest;
+ ack->ccn = ccc->ccn;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending CADET_LOCAL_ACK to %s (%s) at ccn %X\n",
+ GSC_2s (ccc->c),
+ (GNUNET_YES == to_owner) ? "owner" : "dest",
+ ntohl (ack->ccn.channel_of_client));
+ GSC_send_to_client (ccc->c,
env);
}
GCCH_bind (struct CadetChannel *ch,
struct CadetClient *c)
{
- struct GNUNET_MQ_Envelope *env;
- struct GNUNET_CADET_LocalChannelCreateMessage *tcm;
uint32_t options;
+ struct CadetChannelClient *cccd;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Binding %s from %s to port %s of %s\n",
options |= GNUNET_CADET_OPTION_RELIABLE;
if (ch->out_of_order)
options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
- ch->dest = c;
- ch->ccn = GSC_bind (c,
- ch,
- GCT_get_destination (ch->t),
- &ch->port,
- options);
- ch->mid_recv.mid = htonl (1); /* The CONNECT counts as message 0! */
-
- /* notify other peer that we accepted the connection */
- ch->retry_control_task
- = GNUNET_SCHEDULER_add_now (&send_open_ack,
- ch);
+ cccd = GNUNET_new (struct CadetChannelClient);
+ ch->dest = cccd;
+ cccd->c = c;
+ cccd->client_ready = GNUNET_YES;
+ cccd->ccn = GSC_bind (c,
+ ch,
+ (GNUNET_YES == ch->is_loopback)
+ ? GCP_get (&my_full_id,
+ GNUNET_YES)
+ : GCT_get_destination (ch->t),
+ &ch->port,
+ options);
+ GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
+ GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
+ ch->mid_recv.mid = htonl (1); /* The OPEN counts as message 0! */
+ if (GNUNET_YES == ch->is_loopback)
+ {
+ ch->state = CADET_CHANNEL_OPEN_SENT;
+ GCCH_handle_channel_open_ack (ch);
+ }
+ else
+ {
+ /* notify other peer that we accepted the connection */
+ ch->retry_control_task
+ = GNUNET_SCHEDULER_add_now (&send_open_ack,
+ ch);
+ }
/* give client it's initial supply of ACKs */
- env = GNUNET_MQ_msg (tcm,
- GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE);
- tcm->ccn = ch->ccn;
- tcm->peer = *GCP_get_id (GCT_get_destination (ch->t));
- tcm->port = ch->port;
- tcm->opt = htonl (options);
- GSC_send_to_client (ch->dest,
- env);
+ GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
+ GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
for (unsigned int i=0;i<ch->max_pending_messages;i++)
send_ack_to_client (ch,
- ch->dest);
+ GNUNET_NO);
}
/**
- * Destroy locally created channel. Called by the
- * local client, so no need to tell the client.
+ * Destroy locally created channel. Called by the local client, so no
+ * need to tell the client.
*
* @param ch channel to destroy
+ * @param c client that caused the destruction
+ * @param ccn client number of the client @a c
*/
void
-GCCH_channel_local_destroy (struct CadetChannel *ch)
+GCCH_channel_local_destroy (struct CadetChannel *ch,
+ struct CadetClient *c,
+ struct GNUNET_CADET_ClientChannelNumber ccn)
{
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Local client asks for destruction of %s which it initiated\n",
+ "%s asks for destruction of %s\n",
+ GSC_2s (c),
GCCH_2s (ch));
- if (GNUNET_YES == ch->destroy)
+ GNUNET_assert (NULL != c);
+ if ( (NULL != ch->owner) &&
+ (c == ch->owner->c) &&
+ (ccn.channel_of_client == ch->owner->ccn.channel_of_client) )
{
- /* other end already destroyed, with the local client gone, no need
- to finish transmissions, just destroy immediately. */
- channel_destroy (ch);
- return;
+ free_channel_client (ch->owner);
+ ch->owner = NULL;
}
- if (NULL != ch->head_sent)
+ else if ( (NULL != ch->dest) &&
+ (c == ch->dest->c) &&
+ (ccn.channel_of_client == ch->dest->ccn.channel_of_client) )
{
- /* allow send queue to train first */
- ch->destroy = GNUNET_YES;
- return;
+ free_channel_client (ch->dest);
+ ch->dest = NULL;
+ }
+ else
+ {
+ GNUNET_assert (0);
}
- /* 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);
- /* Now finish our clean up */
- channel_destroy (ch);
-}
-
-/**
- * Destroy channel that was incoming. Called by the
- * local client, so no need to tell the client.
- *
- * @param ch channel to destroy
- */
-void
-GCCH_channel_incoming_destroy (struct CadetChannel *ch)
-{
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Local client asks for destruction of %s which it accepted\n",
- GCCH_2s (ch));
if (GNUNET_YES == ch->destroy)
{
- /* other end already destroyed, with the remote client gone, no need
+ /* other end already destroyed, with the local client gone, no need
to finish transmissions, just destroy immediately. */
channel_destroy (ch);
return;
}
- if (NULL != ch->head_recv)
+ if ( (NULL != ch->head_sent) ||
+ (NULL != ch->owner) ||
+ (NULL != ch->dest) )
{
- /* allow local client to see all data first */
+ /* 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);
/* Nothing left to do, just finish destruction */
- GCT_send_channel_destroy (ch->t,
- ch->ctn);
channel_destroy (ch);
}
return;
}
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Received channel OPEN_ACK for waiting %s, entering READY state\n",
+ "Received CHANNEL_OPEN_ACK for waiting %s, entering READY state\n",
GCCH_2s (ch));
- GNUNET_SCHEDULER_cancel (ch->retry_control_task);
- ch->retry_control_task = NULL;
+ if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
+ {
+ GNUNET_SCHEDULER_cancel (ch->retry_control_task);
+ ch->retry_control_task = NULL;
+ }
ch->state = CADET_CHANNEL_READY;
/* On first connect, send client as many ACKs as we allow messages
to be buffered! */
for (unsigned int i=0;i<ch->max_pending_messages;i++)
send_ack_to_client (ch,
- ch->owner);
+ GNUNET_YES);
break;
case CADET_CHANNEL_READY:
/* duplicate ACK, maybe we retried the CREATE. Ignore. */
* and send an ACK to the other end (once flow control allows it!)
*
* @param ch channel that got data
+ * @param msg message that was received
*/
void
GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
{
struct GNUNET_MQ_Envelope *env;
struct GNUNET_CADET_LocalData *ld;
+ struct CadetChannelClient *ccc;
struct CadetOutOfOrderMessage *com;
size_t payload_size;
+ GNUNET_assert (GNUNET_NO == ch->is_loopback);
+ if ( (GNUNET_YES == ch->destroy) &&
+ (NULL == ch->owner) &&
+ (NULL == ch->dest) )
+ {
+ /* This client is gone, but we still have messages to send to
+ the other end (which is why @a ch is not yet dead). However,
+ we cannot pass messages to our client anymore. */
+ 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! */
+ return;
+ }
payload_size = ntohs (msg->header.size) - sizeof (*msg);
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Receicved %u bytes of application data on %s\n",
- (unsigned int) payload_size,
- GCCH_2s (ch));
env = GNUNET_MQ_msg_extra (ld,
payload_size,
GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
- ld->ccn = ch->ccn;
+ ld->ccn = (NULL == ch->dest) ? ch->owner->ccn : ch->dest->ccn;
GNUNET_memcpy (&ld[1],
&msg[1],
payload_size);
- if ( (GNUNET_YES == ch->client_ready) &&
+ ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
+ if ( (GNUNET_YES == ccc->client_ready) &&
( (GNUNET_YES == ch->out_of_order) ||
(msg->mid.mid == ch->mid_recv.mid) ) )
{
- GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Giving %u bytes of payload from %s to client %s\n",
+ (unsigned int) payload_size,
+ GCCH_2s (ch),
+ GSC_2s (ccc->c));
+ ccc->client_ready = GNUNET_NO;
+ GSC_send_to_client (ccc->c,
env);
ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
ch->mid_futures >>= 1;
{
/* FIXME-SECURITY: if the element is WAY too far ahead,
drop it (can't buffer too much!) */
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Queuing %s payload of %u bytes on %s (mid %u, need %u first)\n",
+ (GNUNET_YES == ccc->client_ready)
+ ? "out-of-order"
+ : "client-not-ready",
+ (unsigned int) payload_size,
+ GCCH_2s (ch),
+ ntohl (msg->mid.mid),
+ ntohl (ch->mid_recv.mid));
+
com = GNUNET_new (struct CadetOutOfOrderMessage);
com->mid = msg->mid;
com->env = env;
/* sort into list ordered by "is_before" */
- if ( (NULL == ch->head_recv) ||
+ if ( (NULL == ccc->head_recv) ||
(GNUNET_YES == is_before (ch,
com,
- ch->head_recv)) )
+ ccc->head_recv)) )
{
- GNUNET_CONTAINER_DLL_insert (ch->head_recv,
- ch->tail_recv,
+ GNUNET_CONTAINER_DLL_insert (ccc->head_recv,
+ ccc->tail_recv,
com);
}
else
{
struct CadetOutOfOrderMessage *pos;
- for (pos = ch->head_recv;
+ for (pos = ccc->head_recv;
NULL != pos;
pos = pos->next)
{
if (GNUNET_YES !=
- is_before (ch,
+ is_before (NULL,
pos,
com))
break;
}
if (NULL == pos)
- GNUNET_CONTAINER_DLL_insert_tail (ch->head_recv,
- ch->tail_recv,
+ GNUNET_CONTAINER_DLL_insert_tail (ccc->head_recv,
+ ccc->tail_recv,
com);
else
- GNUNET_CONTAINER_DLL_insert_after (ch->head_recv,
- ch->tail_recv,
+ GNUNET_CONTAINER_DLL_insert_after (ccc->head_recv,
+ ccc->tail_recv,
com,
pos->prev);
}
{
struct CadetReliableMessage *crm;
+ GNUNET_break (GNUNET_NO == ch->is_loopback);
if (GNUNET_NO == ch->reliable)
{
/* not expecting ACKs on unreliable channel, odd */
for (crm = ch->head_sent;
NULL != crm;
crm = crm->next)
- if (ack->mid.mid == crm->data_message.mid.mid)
+ if (ack->mid.mid == crm->data_message->mid.mid)
break;
if (NULL == crm)
{
GNUNET_CONTAINER_DLL_remove (ch->head_sent,
ch->tail_sent,
crm);
- ch->pending_messages--;
+ GNUNET_free (crm->data_message);
GNUNET_free (crm);
+ ch->pending_messages--;
+ send_ack_to_client (ch,
+ (NULL == ch->owner)
+ ? GNUNET_NO
+ : GNUNET_YES);
GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
(unsigned int) ntohl (ack->mid.mid),
ch->pending_messages);
send_ack_to_client (ch,
- (NULL == ch->owner) ? ch->dest : ch->owner);
+ (NULL == ch->owner)
+ ? GNUNET_NO
+ : GNUNET_YES);
}
void
GCCH_handle_remote_destroy (struct CadetChannel *ch)
{
+ struct CadetChannelClient *ccc;
+
+ GNUNET_assert (GNUNET_NO == ch->is_loopback);
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Received remote channel DESTROY for %s\n",
GCCH_2s (ch));
+ if (GNUNET_YES == ch->destroy)
+ {
+ /* Local client already gone, this is instant-death. */
+ channel_destroy (ch);
+ return;
+ }
+ ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
+ if (NULL != ccc->head_recv)
+ {
+ LOG (GNUNET_ERROR_TYPE_WARNING,
+ "Lost end of transmission due to remote shutdown on %s\n",
+ GCCH_2s (ch));
+ /* FIXME: change API to notify client about truncated transmission! */
+ }
ch->destroy = GNUNET_YES;
- GSC_handle_remote_channel_destroy ((NULL != ch->owner) ? ch->owner : ch->dest,
- ch->ccn,
+ GSC_handle_remote_channel_destroy (ccc->c,
+ ccc->ccn,
ch);
channel_destroy (ch);
}
ch->retry_data_task = NULL;
GNUNET_assert (NULL == crm->qe);
crm->qe = GCT_send (ch->t,
- &crm->data_message.header,
+ &crm->data_message->header,
&data_sent_cb,
crm);
}
-/**
- * Check if we can now allow the client to transmit, and if so,
- * let the client know about it.
- *
- * @param ch channel to check
- */
-static void
-GCCH_check_allow_client (struct CadetChannel *ch)
-{
- struct GNUNET_MQ_Envelope *env;
- struct GNUNET_CADET_LocalAck *msg;
-
- if (GNUNET_YES == ch->client_allowed)
- return; /* client already allowed! */
- if (CADET_CHANNEL_READY != ch->state)
- {
- /* destination did not yet ACK our CREATE! */
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "%s not yet ready, throttling client until ACK.\n",
- GCCH_2s (ch));
- return;
- }
- if (ch->pending_messages > ch->max_pending_messages)
- {
- /* Too many messages in queue. */
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Message queue still too long on %s, throttling client until ACK.\n",
- GCCH_2s (ch));
- return;
- }
- if ( (NULL != ch->head_sent) &&
- (64 <= ntohl (ch->mid_send.mid) - ntohl (ch->head_sent->data_message.mid.mid)) )
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Gap in ACKs too big on %s, throttling client until ACK.\n",
- GCCH_2s (ch));
- return;
- }
- ch->client_allowed = GNUNET_YES;
-
-
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sending local ack to %s client\n",
- GCCH_2s (ch));
- env = GNUNET_MQ_msg (msg,
- GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
- msg->ccn = ch->ccn;
- GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
- env);
-}
-
-
/**
* Function called once the tunnel has sent one of our messages.
* If the message is unreliable, simply frees the `crm`. If the
struct CadetChannel *ch = crm->ch;
struct CadetReliableMessage *off;
+ GNUNET_assert (GNUNET_NO == ch->is_loopback);
crm->qe = NULL;
GNUNET_CONTAINER_DLL_remove (ch->head_sent,
ch->tail_sent,
{
GNUNET_free (crm);
ch->pending_messages--;
- GCCH_check_allow_client (ch);
+ send_ack_to_client (ch,
+ (NULL == ch->owner)
+ ? GNUNET_NO
+ : GNUNET_YES);
return;
}
if (0 == crm->retry_delay.rel_value_us)
* buffer space in the tunnel.
*
* @param ch Channel.
+ * @param sender_ccn ccn of the sender
* @param buf payload to transmit.
* @param buf_len number of bytes in @a buf
* @return #GNUNET_OK if everything goes well,
*/
int
GCCH_handle_local_data (struct CadetChannel *ch,
+ struct GNUNET_CADET_ClientChannelNumber sender_ccn,
const char *buf,
size_t buf_len)
{
struct CadetReliableMessage *crm;
- if (GNUNET_NO == ch->client_allowed)
+ if (ch->pending_messages > ch->max_pending_messages)
{
- GNUNET_break_op (0);
+ GNUNET_break (0);
return GNUNET_SYSERR;
}
- ch->client_allowed = GNUNET_NO;
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;
+
+ 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)
+ {
+ receiver = ch->dest;
+ to_owner = GNUNET_NO;
+ }
+ else
+ {
+ GNUNET_assert (sender_ccn.channel_of_client ==
+ ch->dest->ccn.channel_of_client);
+ receiver = ch->owner;
+ to_owner = GNUNET_YES;
+ }
+ ld->ccn = receiver->ccn;
+ GNUNET_memcpy (&ld[1],
+ buf,
+ buf_len);
+ /* FIXME: this does not provide for flow control! */
+ GSC_send_to_client (receiver->c,
+ env);
+ send_ack_to_client (ch,
+ to_owner);
+ return GNUNET_OK;
+ }
+
/* Everything is correct, send the message. */
- crm = GNUNET_malloc (sizeof (*crm) + buf_len);
+ crm = GNUNET_malloc (sizeof (*crm));
crm->ch = ch;
- crm->data_message.header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
- crm->data_message.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
+ crm->data_message = GNUNET_malloc (sizeof (struct GNUNET_CADET_ChannelAppDataMessage)
+ + buf_len);
+ crm->data_message->header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
+ crm->data_message->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
- crm->data_message.mid = ch->mid_send;
- crm->data_message.ctn = ch->ctn;
- GNUNET_memcpy (&crm[1],
+ crm->data_message->mid = ch->mid_send;
+ crm->data_message->ctn = ch->ctn;
+ GNUNET_memcpy (&crm->data_message[1],
buf,
buf_len);
GNUNET_CONTAINER_DLL_insert (ch->head_sent,
buf_len,
GCCH_2s (ch));
crm->qe = GCT_send (ch->t,
- &crm->data_message.header,
+ &crm->data_message->header,
&data_sent_cb,
crm);
- GCCH_check_allow_client (ch);
return GNUNET_OK;
}
/**
- * Try to deliver messages to the local client, if it is ready for more.
+ * Handle ACK from client on local channel. Means the client is ready
+ * for more data, see if we have any for it.
*
- * @param ch channel to process
+ * @param ch channel to destroy
+ * @param client_ccn ccn of the client sending the ack
*/
-static void
-send_client_buffered_data (struct CadetChannel *ch)
+void
+GCCH_handle_local_ack (struct CadetChannel *ch,
+ struct GNUNET_CADET_ClientChannelNumber client_ccn)
{
+ struct CadetChannelClient *ccc;
struct CadetOutOfOrderMessage *com;
- if (GNUNET_NO == ch->client_ready)
- return; /* client not ready */
- com = ch->head_recv;
+ if ( (NULL != ch->owner) &&
+ (ch->owner->ccn.channel_of_client == client_ccn.channel_of_client) )
+ ccc = ch->owner;
+ else if ( (NULL != ch->dest) &&
+ (ch->dest->ccn.channel_of_client == client_ccn.channel_of_client) )
+ ccc = ch->dest;
+ else
+ GNUNET_assert (0);
+ ccc->client_ready = GNUNET_YES;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got LOCAL_ACK, client ready to receive more data!\n");
+ com = ccc->head_recv;
if (NULL == com)
return; /* none pending */
if ( (com->mid.mid != ch->mid_recv.mid) &&
GCCH_2s (ch));
/* all good, pass next message to client */
- GNUNET_CONTAINER_DLL_remove (ch->head_recv,
- ch->tail_recv,
+ GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
+ ccc->tail_recv,
com);
/* 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 */
- GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
+ ccc->client_ready = GNUNET_NO;
+ GSC_send_to_client (ccc->c,
com->env);
GNUNET_free (com);
if ( (0xFFULL == (ch->mid_futures & 0xFFULL)) &&
send_channel_data_ack (ch);
}
- if (NULL != ch->head_recv)
+ if (NULL != ccc->head_recv)
return;
if (GNUNET_NO == ch->destroy)
return;
}
-/**
- * Handle ACK from client on local channel.
- *
- * @param ch channel to destroy
- */
-void
-GCCH_handle_local_ack (struct CadetChannel *ch)
-{
- ch->client_ready = GNUNET_YES;
- send_client_buffered_data (ch);
-}
-
-
#define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-chn",__VA_ARGS__)
{
LOG2 (level,
"CHN origin %s ready %s local-id: %u\n",
- GSC_2s (ch->owner),
- ch->client_ready ? "YES" : "NO",
- ntohl (ch->ccn.channel_of_client));
+ GSC_2s (ch->owner->c),
+ ch->owner->client_ready ? "YES" : "NO",
+ ntohl (ch->owner->ccn.channel_of_client));
}
if (NULL != ch->dest)
{
LOG2 (level,
"CHN destination %s ready %s local-id: %u\n",
- GSC_2s (ch->dest),
- ch->client_ready ? "YES" : "NO",
- ntohl (ch->ccn.channel_of_client));
+ GSC_2s (ch->dest->c),
+ ch->dest->client_ready ? "YES" : "NO",
+ ntohl (ch->dest->ccn.channel_of_client));
}
LOG2 (level,
"CHN Message IDs recv: %d (%LLX), send: %d\n",
projects / oweals / gnunet.git / blobdiff