2 This file is part of GNUnet.
3 Copyright (C) 2001-2017 GNUnet e.V.
5 GNUnet is free software: you can redistribute it and/or modify it
6 under the terms of the GNU Affero General Public License as published
7 by the Free Software Foundation, either version 3 of the License,
8 or (at your option) any later version.
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 Affero General Public License for more details.
15 You should have received a copy of the GNU Affero General Public License
16 along with this program. If not, see <http://www.gnu.org/licenses/>.
18 SPDX-License-Identifier: AGPL3.0-or-later
21 * @file cadet/gnunet-service-cadet_channel.c
22 * @brief logical links between CADET clients
23 * @author Bartlomiej Polot
24 * @author Christian Grothoff
27 * - Congestion/flow control:
28 * + estimate max bandwidth using bursts and use to for CONGESTION CONTROL!
29 * (and figure out how/where to use this!)
30 * + figure out flow control without ACKs (unreliable traffic!)
31 * - revisit handling of 'unbuffered' traffic!
32 * (need to push down through tunnel into connection selection)
33 * - revisit handling of 'buffered' traffic: 4 is a rather small buffer; maybe
34 * reserve more bits in 'options' to allow for buffer size control?
38 #include "gnunet_statistics_service.h"
39 #include "gnunet-service-cadet_channel.h"
40 #include "gnunet-service-cadet_connection.h"
41 #include "gnunet-service-cadet_tunnels.h"
42 #include "gnunet-service-cadet_paths.h"
44 #define LOG(level, ...) GNUNET_log_from (level, "cadet-chn", __VA_ARGS__)
47 * How long do we initially wait before retransmitting?
49 #define CADET_INITIAL_RETRANSMIT_TIME \
50 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 250)
53 * How long do we wait before dropping state about incoming
54 * connection to closed port?
56 #define TIMEOUT_CLOSED_PORT \
57 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
60 * How long do we wait at least before retransmitting ever?
62 #define MIN_RTT_DELAY \
63 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 75)
66 * Maximum message ID into the future we accept for out-of-order messages.
67 * If the message is more than this into the future, we drop it. This is
68 * important both to detect values that are actually in the past, as well
69 * as to limit adversarially triggerable memory consumption.
71 * Note that right now we have "max_pending_messages = 4" hard-coded in
72 * the logic below, so a value of 4 would suffice here. But we plan to
73 * allow larger windows in the future...
75 #define MAX_OUT_OF_ORDER_DISTANCE 1024
79 * All the states a channel can be in.
81 enum CadetChannelState
84 * Uninitialized status, should never appear in operation.
89 * Channel is to a port that is not open, we're waiting for the
95 * CHANNEL_OPEN message sent, waiting for CHANNEL_OPEN_ACK.
97 CADET_CHANNEL_OPEN_SENT,
100 * Connection confirmed, ready to carry traffic.
107 * Info needed to retry a message in case it gets lost.
108 * Note that we DO use this structure also for unreliable
111 struct CadetReliableMessage
114 * Double linked list, FIFO style
116 struct CadetReliableMessage *next;
119 * Double linked list, FIFO style
121 struct CadetReliableMessage *prev;
124 * Which channel is this message in?
126 struct CadetChannel *ch;
129 * Entry in the tunnels queue for this message, NULL if it has left
130 * the tunnel. Used to cancel transmission in case we receive an
133 struct CadetTunnelQueueEntry *qe;
136 * Data message we are trying to send.
138 struct GNUNET_CADET_ChannelAppDataMessage *data_message;
141 * How soon should we retry if we fail to get an ACK?
142 * Messages in the queue are sorted by this value.
144 struct GNUNET_TIME_Absolute next_retry;
147 * How long do we wait for an ACK after transmission?
148 * Use for the back-off calculation.
150 struct GNUNET_TIME_Relative retry_delay;
153 * Time when we first successfully transmitted the message
154 * (that is, set @e num_transmissions to 1).
156 struct GNUNET_TIME_Absolute first_transmission_time;
159 * Identifier of the connection that this message took when it
160 * was first transmitted. Only useful if @e num_transmissions is 1.
162 struct GNUNET_CADET_ConnectionTunnelIdentifier connection_taken;
165 * How often was this message transmitted? #GNUNET_SYSERR if there
166 * was an error transmitting the message, #GNUNET_NO if it was not
167 * yet transmitted ever, otherwise the number of (re) transmissions.
169 int num_transmissions;
174 * List of received out-of-order data messages.
176 struct CadetOutOfOrderMessage
179 * Double linked list, FIFO style
181 struct CadetOutOfOrderMessage *next;
184 * Double linked list, FIFO style
186 struct CadetOutOfOrderMessage *prev;
189 * ID of the message (messages up to this point needed
190 * before we give this one to the client).
192 struct ChannelMessageIdentifier mid;
195 * The envelope with the payload of the out-of-order message
197 struct GNUNET_MQ_Envelope *env;
202 * Client endpoint of a `struct CadetChannel`. A channel may be a
203 * loopback channel, in which case it has two of these endpoints.
204 * Note that flow control also is required in both directions.
206 struct CadetChannelClient
209 * Client handle. Not by itself sufficient to designate
210 * the client endpoint, as the same client handle may
211 * be used for both the owner and the destination, and
212 * we thus also need the channel ID to identify the client.
214 struct CadetClient *c;
217 * Head of DLL of messages received out of order or while client was unready.
219 struct CadetOutOfOrderMessage *head_recv;
222 * Tail DLL of messages received out of order or while client was unready.
224 struct CadetOutOfOrderMessage *tail_recv;
227 * Local tunnel number for this client.
228 * (if owner >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI,
229 * otherwise < #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
231 struct GNUNET_CADET_ClientChannelNumber ccn;
234 * Number of entries currently in @a head_recv DLL.
236 unsigned int num_recv;
239 * Can we send data to the client?
246 * Struct containing all information regarding a channel to a remote client.
251 * Tunnel this channel is in.
253 struct CadetTunnel *t;
256 * Client owner of the tunnel, if any.
257 * (Used if this channel represends the initiating end of the tunnel.)
259 struct CadetChannelClient *owner;
262 * Client destination of the tunnel, if any.
263 * (Used if this channel represents the listening end of the tunnel.)
265 struct CadetChannelClient *dest;
268 * Last entry in the tunnel's queue relating to control messages
269 * (#GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN or
270 * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK). Used to cancel
271 * transmission in case we receive updated information.
273 struct CadetTunnelQueueEntry *last_control_qe;
276 * Head of DLL of messages sent and not yet ACK'd.
278 struct CadetReliableMessage *head_sent;
281 * Tail of DLL of messages sent and not yet ACK'd.
283 struct CadetReliableMessage *tail_sent;
286 * Task to resend/poll in case no ACK is received.
288 struct GNUNET_SCHEDULER_Task *retry_control_task;
291 * Task to resend/poll in case no ACK is received.
293 struct GNUNET_SCHEDULER_Task *retry_data_task;
296 * Last time the channel was used
298 struct GNUNET_TIME_Absolute timestamp;
301 * Destination port of the channel.
303 struct GNUNET_HashCode port;
306 * Hash'ed port of the channel with initiator and destination PID.
308 struct GNUNET_HashCode h_port;
311 * Counter for exponential backoff.
313 struct GNUNET_TIME_Relative retry_time;
316 * Bitfield of already-received messages past @e mid_recv.
318 uint64_t mid_futures;
321 * Next MID expected for incoming traffic.
323 struct ChannelMessageIdentifier mid_recv;
326 * Next MID to use for outgoing traffic.
328 struct ChannelMessageIdentifier mid_send;
331 * Total (reliable) messages pending ACK for this channel.
333 unsigned int pending_messages;
336 * Maximum (reliable) messages pending ACK for this channel
337 * before we throttle the client.
339 unsigned int max_pending_messages;
342 * Number identifying this channel in its tunnel.
344 struct GNUNET_CADET_ChannelTunnelNumber ctn;
349 enum CadetChannelState state;
352 * Count how many ACKs we skipped, used to prevent long
353 * sequences of ACK skipping.
355 unsigned int skip_ack_series;
358 * Is the tunnel bufferless (minimum latency)?
363 * Is the tunnel reliable?
368 * Is the tunnel out-of-order?
373 * Is this channel a loopback channel, where the destination is us again?
378 * Flag to signal the destruction of the channel. If this is set to
379 * #GNUNET_YES the channel will be destroyed once the queue is
387 * Get the static string for identification of the channel.
391 * @return Static string with the channel IDs.
394 GCCH_2s (const struct CadetChannel *ch)
396 static char buf[128];
398 GNUNET_snprintf (buf,
400 "Channel %s:%s ctn:%X(%X/%X)",
401 (GNUNET_YES == ch->is_loopback)
403 : GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
404 GNUNET_h2s (&ch->port),
408 : ntohl (ch->owner->ccn.channel_of_client),
411 : ntohl (ch->dest->ccn.channel_of_client));
417 * Hash the @a port and @a initiator and @a listener to
418 * calculate the "challenge" @a h_port we send to the other
419 * peer on #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN.
421 * @param[out] h_port set to the hash of @a port, @a initiator and @a listener
422 * @param port cadet port, as seen by CADET clients
423 * @param listener peer that is listining on @a port
426 GCCH_hash_port (struct GNUNET_HashCode *h_port,
427 const struct GNUNET_HashCode *port,
428 const struct GNUNET_PeerIdentity *listener)
430 struct GNUNET_HashContext *hc;
432 hc = GNUNET_CRYPTO_hash_context_start ();
433 GNUNET_CRYPTO_hash_context_read (hc, port, sizeof(*port));
434 GNUNET_CRYPTO_hash_context_read (hc, listener, sizeof(*listener));
435 GNUNET_CRYPTO_hash_context_finish (hc, h_port);
436 LOG (GNUNET_ERROR_TYPE_DEBUG,
437 "Calculated port hash %s\n",
438 GNUNET_h2s (h_port));
443 * Get the channel's public ID.
447 * @return ID used to identify the channel with the remote peer.
449 struct GNUNET_CADET_ChannelTunnelNumber
450 GCCH_get_id (const struct CadetChannel *ch)
457 * Release memory associated with @a ccc
459 * @param ccc data structure to clean up
462 free_channel_client (struct CadetChannelClient *ccc)
464 struct CadetOutOfOrderMessage *com;
466 while (NULL != (com = ccc->head_recv))
468 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, com);
470 GNUNET_MQ_discard (com->env);
478 * Destroy the given channel.
480 * @param ch channel to destroy
483 channel_destroy (struct CadetChannel *ch)
485 struct CadetReliableMessage *crm;
487 while (NULL != (crm = ch->head_sent))
489 GNUNET_assert (ch == crm->ch);
492 GCT_send_cancel (crm->qe);
495 GNUNET_CONTAINER_DLL_remove (ch->head_sent, ch->tail_sent, crm);
496 GNUNET_free (crm->data_message);
499 if (CADET_CHANNEL_LOOSE == ch->state)
501 GSC_drop_loose_channel (&ch->h_port, ch);
503 if (NULL != ch->owner)
505 free_channel_client (ch->owner);
508 if (NULL != ch->dest)
510 free_channel_client (ch->dest);
513 if (NULL != ch->last_control_qe)
515 GCT_send_cancel (ch->last_control_qe);
516 ch->last_control_qe = NULL;
518 if (NULL != ch->retry_data_task)
520 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
521 ch->retry_data_task = NULL;
523 if (NULL != ch->retry_control_task)
525 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
526 ch->retry_control_task = NULL;
528 if (GNUNET_NO == ch->is_loopback)
530 GCT_remove_channel (ch->t, ch, ch->ctn);
538 * Send a channel create message.
540 * @param cls Channel for which to send.
543 send_channel_open (void *cls);
547 * Function called once the tunnel confirms that we sent the
548 * create message. Delays for a bit until we retry.
550 * @param cls our `struct CadetChannel`.
551 * @param cid identifier of the connection within the tunnel, NULL
552 * if transmission failed
555 channel_open_sent_cb (void *cls,
556 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
558 struct CadetChannel *ch = cls;
560 GNUNET_assert (NULL != ch->last_control_qe);
561 ch->last_control_qe = NULL;
562 ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
563 LOG (GNUNET_ERROR_TYPE_DEBUG,
564 "Sent CADET_CHANNEL_OPEN on %s, retrying in %s\n",
566 GNUNET_STRINGS_relative_time_to_string (ch->retry_time, GNUNET_YES));
567 ch->retry_control_task =
568 GNUNET_SCHEDULER_add_delayed (ch->retry_time, &send_channel_open, ch);
573 * Send a channel open message.
575 * @param cls Channel for which to send.
578 send_channel_open (void *cls)
580 struct CadetChannel *ch = cls;
581 struct GNUNET_CADET_ChannelOpenMessage msgcc;
583 ch->retry_control_task = NULL;
584 LOG (GNUNET_ERROR_TYPE_DEBUG,
585 "Sending CHANNEL_OPEN message for %s\n",
587 msgcc.header.size = htons (sizeof(msgcc));
588 msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN);
589 // TODO This will be removed in a major release, because this will be a protocol breaking change. We set the deprecated "reliable" bit here that was removed.
591 msgcc.h_port = ch->h_port;
593 ch->state = CADET_CHANNEL_OPEN_SENT;
594 if (NULL != ch->last_control_qe)
595 GCT_send_cancel (ch->last_control_qe);
596 ch->last_control_qe =
597 GCT_send (ch->t, &msgcc.header, &channel_open_sent_cb, ch);
598 GNUNET_assert (NULL == ch->retry_control_task);
603 * Function called once and only once after a channel was bound
604 * to its tunnel via #GCT_add_channel() is ready for transmission.
605 * Note that this is only the case for channels that this peer
606 * initiates, as for incoming channels we assume that they are
607 * ready for transmission immediately upon receiving the open
608 * message. Used to bootstrap the #GCT_send() process.
610 * @param ch the channel for which the tunnel is now ready
613 GCCH_tunnel_up (struct CadetChannel *ch)
615 GNUNET_assert (NULL == ch->retry_control_task);
616 LOG (GNUNET_ERROR_TYPE_DEBUG,
617 "Tunnel up, sending CHANNEL_OPEN on %s now\n",
619 ch->retry_control_task = GNUNET_SCHEDULER_add_now (&send_channel_open, ch);
624 * Create a new channel.
626 * @param owner local client owning the channel
627 * @param ccn local number of this channel at the @a owner
628 * @param destination peer to which we should build the channel
629 * @param port desired port at @a destination
630 * @param options options for the channel
631 * @return handle to the new channel
633 struct CadetChannel *
634 GCCH_channel_local_new (struct CadetClient *owner,
635 struct GNUNET_CADET_ClientChannelNumber ccn,
636 struct CadetPeer *destination,
637 const struct GNUNET_HashCode *port,
640 struct CadetChannel *ch;
641 struct CadetChannelClient *ccco;
643 ccco = GNUNET_new (struct CadetChannelClient);
646 ccco->client_ready = GNUNET_YES;
648 ch = GNUNET_new (struct CadetChannel);
649 ch->mid_recv.mid = htonl (1); /* The OPEN_ACK counts as message 0! */
650 ch->nobuffer = GNUNET_NO;
651 ch->reliable = GNUNET_YES;
652 ch->out_of_order = GNUNET_NO;
653 ch->max_pending_messages =
654 (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
657 GCCH_hash_port (&ch->h_port, port, GCP_get_id (destination));
658 if (0 == GNUNET_memcmp (&my_full_id, GCP_get_id (destination)))
662 ch->is_loopback = GNUNET_YES;
663 op = GNUNET_CONTAINER_multihashmap_get (open_ports, &ch->h_port);
666 /* port closed, wait for it to possibly open */
667 ch->state = CADET_CHANNEL_LOOSE;
668 (void) GNUNET_CONTAINER_multihashmap_put (
672 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
673 LOG (GNUNET_ERROR_TYPE_DEBUG,
674 "Created loose incoming loopback channel to port %s\n",
675 GNUNET_h2s (&ch->port));
679 GCCH_bind (ch, op->c, &op->port);
684 ch->t = GCP_get_tunnel (destination, GNUNET_YES);
685 ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
686 ch->ctn = GCT_add_channel (ch->t, ch);
688 GNUNET_STATISTICS_update (stats, "# channels", 1, GNUNET_NO);
689 LOG (GNUNET_ERROR_TYPE_DEBUG,
690 "Created channel to port %s at peer %s for %s using %s\n",
692 GCP_2s (destination),
694 (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
700 * We had an incoming channel to a port that is closed.
701 * It has not been opened for a while, drop it.
703 * @param cls the channel to drop
706 timeout_closed_cb (void *cls)
708 struct CadetChannel *ch = cls;
710 ch->retry_control_task = NULL;
711 LOG (GNUNET_ERROR_TYPE_DEBUG,
712 "Closing incoming channel to port %s from peer %s due to timeout\n",
713 GNUNET_h2s (&ch->port),
714 GCP_2s (GCT_get_destination (ch->t)));
715 channel_destroy (ch);
720 * Create a new channel based on a request coming in over the network.
722 * @param t tunnel to the remote peer
723 * @param ctn identifier of this channel in the tunnel
724 * @param h_port desired hash of local port
725 * @param options options for the channel
726 * @return handle to the new channel
728 struct CadetChannel *
729 GCCH_channel_incoming_new (struct CadetTunnel *t,
730 struct GNUNET_CADET_ChannelTunnelNumber ctn,
731 const struct GNUNET_HashCode *h_port,
734 struct CadetChannel *ch;
737 ch = GNUNET_new (struct CadetChannel);
738 ch->h_port = *h_port;
741 ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
742 ch->nobuffer = GNUNET_NO;
743 ch->reliable = GNUNET_YES;
744 ch->out_of_order = GNUNET_NO;
745 ch->max_pending_messages =
746 (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
747 GNUNET_STATISTICS_update (stats, "# channels", 1, GNUNET_NO);
749 op = GNUNET_CONTAINER_multihashmap_get (open_ports, h_port);
752 /* port closed, wait for it to possibly open */
753 ch->state = CADET_CHANNEL_LOOSE;
754 (void) GNUNET_CONTAINER_multihashmap_put (
758 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
759 GNUNET_assert (NULL == ch->retry_control_task);
760 ch->retry_control_task = GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
763 LOG (GNUNET_ERROR_TYPE_DEBUG,
764 "Created loose incoming channel to port %s from peer %s\n",
765 GNUNET_h2s (&ch->port),
766 GCP_2s (GCT_get_destination (ch->t)));
770 GCCH_bind (ch, op->c, &op->port);
772 GNUNET_STATISTICS_update (stats, "# channels", 1, GNUNET_NO);
778 * Function called once the tunnel confirms that we sent the
779 * ACK message. Just remembers it was sent, we do not expect
782 * @param cls our `struct CadetChannel`.
783 * @param cid identifier of the connection within the tunnel, NULL
784 * if transmission failed
787 send_ack_cb (void *cls,
788 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
790 struct CadetChannel *ch = cls;
792 GNUNET_assert (NULL != ch->last_control_qe);
793 ch->last_control_qe = NULL;
798 * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
800 * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
803 send_channel_data_ack (struct CadetChannel *ch)
805 struct GNUNET_CADET_ChannelDataAckMessage msg;
807 if (GNUNET_NO == ch->reliable)
808 return; /* no ACKs */
809 msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK);
810 msg.header.size = htons (sizeof(msg));
812 msg.mid.mid = htonl (ntohl (ch->mid_recv.mid));
813 msg.futures = GNUNET_htonll (ch->mid_futures);
814 LOG (GNUNET_ERROR_TYPE_DEBUG,
815 "Sending DATA_ACK %u:%llX via %s\n",
816 (unsigned int) ntohl (msg.mid.mid),
817 (unsigned long long) ch->mid_futures,
819 if (NULL != ch->last_control_qe)
820 GCT_send_cancel (ch->last_control_qe);
821 ch->last_control_qe = GCT_send (ch->t, &msg.header, &send_ack_cb, ch);
826 * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
829 * @param cls the `struct CadetChannel`
832 send_open_ack (void *cls)
834 struct CadetChannel *ch = cls;
835 struct GNUNET_CADET_ChannelOpenAckMessage msg;
837 ch->retry_control_task = NULL;
838 LOG (GNUNET_ERROR_TYPE_DEBUG,
839 "Sending CHANNEL_OPEN_ACK on %s\n",
841 msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
842 msg.header.size = htons (sizeof(msg));
843 msg.reserved = htonl (0);
846 if (NULL != ch->last_control_qe)
847 GCT_send_cancel (ch->last_control_qe);
848 ch->last_control_qe = GCT_send (ch->t, &msg.header, &send_ack_cb, ch);
853 * We got a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN message again for
854 * this channel. If the binding was successful, (re)transmit the
855 * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
857 * @param ch channel that got the duplicate open
858 * @param cti identifier of the connection that delivered the message
861 GCCH_handle_duplicate_open (
862 struct CadetChannel *ch,
863 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
865 if (NULL == ch->dest)
867 LOG (GNUNET_ERROR_TYPE_DEBUG,
868 "Ignoring duplicate CHANNEL_OPEN on %s: port is closed\n",
872 if (NULL != ch->retry_control_task)
874 LOG (GNUNET_ERROR_TYPE_DEBUG,
875 "Ignoring duplicate CHANNEL_OPEN on %s: control message is pending\n",
879 LOG (GNUNET_ERROR_TYPE_DEBUG,
880 "Retransmitting CHANNEL_OPEN_ACK on %s\n",
882 ch->retry_control_task = GNUNET_SCHEDULER_add_now (&send_open_ack, ch);
887 * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK to the client to solicit more messages.
889 * @param ch channel the ack is for
890 * @param to_owner #GNUNET_YES to send to owner,
891 * #GNUNET_NO to send to dest
894 send_ack_to_client (struct CadetChannel *ch, int to_owner)
896 struct GNUNET_MQ_Envelope *env;
897 struct GNUNET_CADET_LocalAck *ack;
898 struct CadetChannelClient *ccc;
900 ccc = (GNUNET_YES == to_owner) ? ch->owner : ch->dest;
903 /* This can happen if we are just getting ACKs after
904 our local client already disconnected. */
905 GNUNET_assert (GNUNET_YES == ch->destroy);
908 env = GNUNET_MQ_msg (ack, GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
910 LOG (GNUNET_ERROR_TYPE_DEBUG,
911 "Sending CADET_LOCAL_ACK to %s (%s) at ccn %X (%u/%u pending)\n",
913 (GNUNET_YES == to_owner) ? "owner" : "dest",
914 ntohl (ack->ccn.channel_of_client),
915 ch->pending_messages,
916 ch->max_pending_messages);
917 GSC_send_to_client (ccc->c, env);
922 * A client is bound to the port that we have a channel
923 * open to. Send the acknowledgement for the connection
924 * request and establish the link with the client.
926 * @param ch open incoming channel
927 * @param c client listening on the respective @a port
928 * @param port the port @a is listening on
931 GCCH_bind (struct CadetChannel *ch,
932 struct CadetClient *c,
933 const struct GNUNET_HashCode *port)
936 struct CadetChannelClient *cccd;
938 LOG (GNUNET_ERROR_TYPE_DEBUG,
939 "Binding %s from %s to port %s of %s\n",
942 GNUNET_h2s (&ch->port),
944 if (NULL != ch->retry_control_task)
946 /* there might be a timeout task here */
947 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
948 ch->retry_control_task = NULL;
951 cccd = GNUNET_new (struct CadetChannelClient);
952 GNUNET_assert (NULL == ch->dest);
956 cccd->client_ready = GNUNET_YES;
957 cccd->ccn = GSC_bind (c,
959 (GNUNET_YES == ch->is_loopback)
960 ? GCP_get (&my_full_id, GNUNET_YES)
961 : GCT_get_destination (ch->t),
964 GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
965 GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
966 ch->mid_recv.mid = htonl (1); /* The OPEN counts as message 0! */
967 if (GNUNET_YES == ch->is_loopback)
969 ch->state = CADET_CHANNEL_OPEN_SENT;
970 GCCH_handle_channel_open_ack (ch, NULL, port);
974 /* notify other peer that we accepted the connection */
975 ch->state = CADET_CHANNEL_READY;
976 ch->retry_control_task = GNUNET_SCHEDULER_add_now (&send_open_ack, ch);
978 /* give client it's initial supply of ACKs */
979 GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
980 GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
981 for (unsigned int i = 0; i < ch->max_pending_messages; i++)
982 send_ack_to_client (ch, GNUNET_NO);
987 * One of our clients has disconnected, tell the other one that we
988 * are finished. Done asynchronously to avoid concurrent modification
989 * issues if this is the same client.
991 * @param cls the `struct CadetChannel` where one of the ends is now dead
994 signal_remote_destroy_cb (void *cls)
996 struct CadetChannel *ch = cls;
997 struct CadetChannelClient *ccc;
999 /* Find which end is left... */
1000 ch->retry_control_task = NULL;
1001 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1002 GSC_handle_remote_channel_destroy (ccc->c, ccc->ccn, ch);
1003 channel_destroy (ch);
1008 * Destroy locally created channel. Called by the local client, so no
1009 * need to tell the client.
1011 * @param ch channel to destroy
1012 * @param c client that caused the destruction
1013 * @param ccn client number of the client @a c
1016 GCCH_channel_local_destroy (struct CadetChannel *ch,
1017 struct CadetClient *c,
1018 struct GNUNET_CADET_ClientChannelNumber ccn)
1020 LOG (GNUNET_ERROR_TYPE_DEBUG,
1021 "%s asks for destruction of %s\n",
1024 GNUNET_assert (NULL != c);
1025 if ((NULL != ch->owner) && (c == ch->owner->c) &&
1026 (ccn.channel_of_client == ch->owner->ccn.channel_of_client))
1028 free_channel_client (ch->owner);
1031 else if ((NULL != ch->dest) && (c == ch->dest->c) &&
1032 (ccn.channel_of_client == ch->dest->ccn.channel_of_client))
1034 free_channel_client (ch->dest);
1042 if (GNUNET_YES == ch->destroy)
1044 /* other end already destroyed, with the local client gone, no need
1045 to finish transmissions, just destroy immediately. */
1046 channel_destroy (ch);
1049 if ((NULL != ch->head_sent) && ((NULL != ch->owner) || (NULL != ch->dest)))
1051 /* Wait for other end to destroy us as well,
1052 and otherwise allow send queue to be transmitted first */
1053 ch->destroy = GNUNET_YES;
1056 if ((GNUNET_YES == ch->is_loopback) &&
1057 ((NULL != ch->owner) || (NULL != ch->dest)))
1059 if (NULL != ch->retry_control_task)
1060 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
1061 ch->retry_control_task =
1062 GNUNET_SCHEDULER_add_now (&signal_remote_destroy_cb, ch);
1065 if (GNUNET_NO == ch->is_loopback)
1067 /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
1070 case CADET_CHANNEL_NEW:
1071 /* We gave up on a channel that we created as a client to a remote
1072 target, but that never went anywhere. Nothing to do here. */
1075 case CADET_CHANNEL_LOOSE:
1079 GCT_send_channel_destroy (ch->t, ch->ctn);
1082 /* Nothing left to do, just finish destruction */
1083 channel_destroy (ch);
1088 * We got an acknowledgement for the creation of the channel
1089 * (the port is open on the other side). Verify that the
1090 * other end really has the right port, and begin transmissions.
1092 * @param ch channel to destroy
1093 * @param cti identifier of the connection that delivered the message
1094 * @param port port number (needed to verify receiver knows the port)
1097 GCCH_handle_channel_open_ack (
1098 struct CadetChannel *ch,
1099 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1100 const struct GNUNET_HashCode *port)
1104 case CADET_CHANNEL_NEW:
1105 /* this should be impossible */
1109 case CADET_CHANNEL_LOOSE:
1110 /* This makes no sense. */
1111 GNUNET_break_op (0);
1114 case CADET_CHANNEL_OPEN_SENT:
1115 if (NULL == ch->owner)
1117 /* We're not the owner, wrong direction! */
1118 GNUNET_break_op (0);
1121 if (0 != GNUNET_memcmp (&ch->port, port))
1123 /* Other peer failed to provide the right port,
1124 refuse connection. */
1125 GNUNET_break_op (0);
1128 LOG (GNUNET_ERROR_TYPE_DEBUG,
1129 "Received CHANNEL_OPEN_ACK for waiting %s, entering READY state\n",
1131 if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
1133 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
1134 ch->retry_control_task = NULL;
1136 ch->state = CADET_CHANNEL_READY;
1137 /* On first connect, send client as many ACKs as we allow messages
1139 for (unsigned int i = 0; i < ch->max_pending_messages; i++)
1140 send_ack_to_client (ch, GNUNET_YES);
1143 case CADET_CHANNEL_READY:
1144 /* duplicate ACK, maybe we retried the CREATE. Ignore. */
1145 LOG (GNUNET_ERROR_TYPE_DEBUG,
1146 "Received duplicate channel OPEN_ACK for %s\n",
1148 GNUNET_STATISTICS_update (stats, "# duplicate CREATE_ACKs", 1, GNUNET_NO);
1155 * Test if element @a e1 comes before element @a e2.
1157 * @param cls closure, to a flag where we indicate duplicate packets
1158 * @param m1 a message of to sort
1159 * @param m2 another message to sort
1160 * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1163 is_before (void *cls,
1164 struct CadetOutOfOrderMessage *m1,
1165 struct CadetOutOfOrderMessage *m2)
1167 int *duplicate = cls;
1168 uint32_t v1 = ntohl (m1->mid.mid);
1169 uint32_t v2 = ntohl (m2->mid.mid);
1174 *duplicate = GNUNET_YES;
1175 if (delta > (uint32_t) INT_MAX)
1177 /* in overflow range, we can safely assume we wrapped around */
1182 /* result is small, thus v2 > v1, thus m1 < m2 */
1189 * We got payload data for a channel. Pass it on to the client
1190 * and send an ACK to the other end (once flow control allows it!)
1192 * @param ch channel that got data
1193 * @param cti identifier of the connection that delivered the message
1194 * @param msg message that was received
1197 GCCH_handle_channel_plaintext_data (
1198 struct CadetChannel *ch,
1199 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1200 const struct GNUNET_CADET_ChannelAppDataMessage *msg)
1202 struct GNUNET_MQ_Envelope *env;
1203 struct GNUNET_CADET_LocalData *ld;
1204 struct CadetChannelClient *ccc;
1205 size_t payload_size;
1206 struct CadetOutOfOrderMessage *com;
1213 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1214 if ((NULL == ch->owner) && (NULL == ch->dest))
1216 /* This client is gone, but we still have messages to send to
1217 the other end (which is why @a ch is not yet dead). However,
1218 we cannot pass messages to our client anymore. */
1219 LOG (GNUNET_ERROR_TYPE_DEBUG,
1220 "Dropping incoming payload on %s as this end is already closed\n",
1222 /* send back DESTROY notification to stop further retransmissions! */
1223 if (GNUNET_YES == ch->destroy)
1224 GCT_send_channel_destroy (ch->t, ch->ctn);
1227 payload_size = ntohs (msg->header.size) - sizeof(*msg);
1228 env = GNUNET_MQ_msg_extra (ld,
1230 GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1231 ld->ccn = (NULL == ch->dest) ? ch->owner->ccn : ch->dest->ccn;
1232 GNUNET_memcpy (&ld[1], &msg[1], payload_size);
1233 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1234 if (GNUNET_YES == ccc->client_ready)
1237 * We ad-hoc send the message if
1238 * - The channel is out-of-order
1239 * - The channel is reliable and MID matches next expected MID
1240 * - The channel is unreliable and MID is before lowest seen MID
1241 */if ((GNUNET_YES == ch->out_of_order) ||
1242 ((msg->mid.mid == ch->mid_recv.mid) && (GNUNET_YES == ch->reliable)) ||
1243 ((GNUNET_NO == ch->reliable) &&
1244 (ntohl (msg->mid.mid) >= ntohl (ch->mid_recv.mid)) &&
1245 ((NULL == ccc->head_recv) ||
1246 (ntohl (msg->mid.mid) < ntohl (ccc->head_recv->mid.mid)))))
1248 LOG (GNUNET_ERROR_TYPE_DEBUG,
1249 "Giving %u bytes of payload with MID %u from %s to client %s\n",
1250 (unsigned int) payload_size,
1251 ntohl (msg->mid.mid),
1254 ccc->client_ready = GNUNET_NO;
1255 GSC_send_to_client (ccc->c, env);
1256 if (GNUNET_NO == ch->out_of_order)
1257 ch->mid_recv.mid = htonl (1 + ntohl (msg->mid.mid));
1259 ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
1260 ch->mid_futures >>= 1;
1261 if ((GNUNET_YES == ch->out_of_order) && (GNUNET_NO == ch->reliable))
1263 /* possibly shift by more if we skipped messages */
1264 uint64_t delta = htonl (msg->mid.mid) - 1 - ntohl (ch->mid_recv.mid);
1267 ch->mid_futures = 0;
1269 ch->mid_futures >>= delta;
1270 ch->mid_recv.mid = htonl (1 + ntohl (msg->mid.mid));
1272 send_channel_data_ack (ch);
1277 if (GNUNET_YES == ch->reliable)
1279 /* check if message ought to be dropped because it is ancient/too distant/duplicate */
1280 mid_min = ntohl (ch->mid_recv.mid);
1281 mid_max = mid_min + ch->max_pending_messages;
1282 mid_msg = ntohl (msg->mid.mid);
1283 if (((uint32_t) (mid_msg - mid_min) > ch->max_pending_messages) ||
1284 ((uint32_t) (mid_max - mid_msg) > ch->max_pending_messages))
1286 LOG (GNUNET_ERROR_TYPE_DEBUG,
1287 "%s at %u drops ancient or far-future message %u\n",
1289 (unsigned int) mid_min,
1290 ntohl (msg->mid.mid));
1292 GNUNET_STATISTICS_update (stats,
1293 "# duplicate DATA (ancient or future)",
1296 GNUNET_MQ_discard (env);
1297 send_channel_data_ack (ch);
1300 /* mark bit for future ACKs */
1301 delta = mid_msg - mid_min - 1; /* overflow/underflow are OK here */
1304 if (0 != (ch->mid_futures & (1LLU << delta)))
1306 /* Duplicate within the queue, drop also */
1307 LOG (GNUNET_ERROR_TYPE_DEBUG,
1308 "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1309 (unsigned int) payload_size,
1311 ntohl (msg->mid.mid));
1312 GNUNET_STATISTICS_update (stats, "# duplicate DATA", 1, GNUNET_NO);
1313 GNUNET_MQ_discard (env);
1314 send_channel_data_ack (ch);
1317 ch->mid_futures |= (1LLU << delta);
1318 LOG (GNUNET_ERROR_TYPE_DEBUG,
1319 "Marked bit %llX for mid %u (base: %u); now: %llX\n",
1326 else /* ! ch->reliable */
1328 struct CadetOutOfOrderMessage *next_msg;
1331 * We always send if possible in this case.
1332 * It is guaranteed that the queued MID < received MID
1334 if ((NULL != ccc->head_recv) && (GNUNET_YES == ccc->client_ready))
1336 next_msg = ccc->head_recv;
1337 LOG (GNUNET_ERROR_TYPE_DEBUG,
1338 "Giving queued MID %u from %s to client %s\n",
1339 ntohl (next_msg->mid.mid),
1342 ccc->client_ready = GNUNET_NO;
1343 GSC_send_to_client (ccc->c, next_msg->env);
1344 ch->mid_recv.mid = htonl (1 + ntohl (next_msg->mid.mid));
1345 ch->mid_futures >>= 1;
1346 send_channel_data_ack (ch);
1347 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, next_msg);
1349 /* Do not process duplicate MID */
1350 if (msg->mid.mid == next_msg->mid.mid) /* Duplicate */
1352 /* Duplicate within the queue, drop */
1353 LOG (GNUNET_ERROR_TYPE_DEBUG,
1354 "Message on %s (mid %u) dropped, duplicate\n",
1356 ntohl (msg->mid.mid));
1357 GNUNET_free (next_msg);
1358 GNUNET_MQ_discard (env);
1361 GNUNET_free (next_msg);
1364 if (ntohl (msg->mid.mid) < ntohl (ch->mid_recv.mid)) /* Old */
1366 /* Duplicate within the queue, drop */
1367 LOG (GNUNET_ERROR_TYPE_DEBUG,
1368 "Message on %s (mid %u) dropped, old.\n",
1370 ntohl (msg->mid.mid));
1371 GNUNET_MQ_discard (env);
1375 /* Channel is unreliable, so we do not ACK. But we also cannot
1376 allow buffering everything, so check if we have space... */
1377 if (ccc->num_recv >= ch->max_pending_messages)
1379 struct CadetOutOfOrderMessage *drop;
1381 /* Yep, need to drop. Drop the oldest message in
1383 LOG (GNUNET_ERROR_TYPE_DEBUG,
1384 "Queue full due slow client on %s, dropping oldest message\n",
1386 GNUNET_STATISTICS_update (stats,
1387 "# messages dropped due to slow client",
1390 drop = ccc->head_recv;
1391 GNUNET_assert (NULL != drop);
1392 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, drop);
1394 GNUNET_MQ_discard (drop->env);
1399 /* Insert message into sorted out-of-order queue */
1400 com = GNUNET_new (struct CadetOutOfOrderMessage);
1401 com->mid = msg->mid;
1403 duplicate = GNUNET_NO;
1404 GNUNET_CONTAINER_DLL_insert_sorted (struct CadetOutOfOrderMessage,
1411 if (GNUNET_YES == duplicate)
1413 /* Duplicate within the queue, drop also (this is not covered by
1414 the case above if "delta" >= 64, which could be the case if
1415 max_pending_messages is also >= 64 or if our client is unready
1416 and we are seeing retransmissions of the message our client is
1417 blocked on. */LOG (GNUNET_ERROR_TYPE_DEBUG,
1418 "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1419 (unsigned int) payload_size,
1421 ntohl (msg->mid.mid));
1422 GNUNET_STATISTICS_update (stats, "# duplicate DATA", 1, GNUNET_NO);
1423 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, com);
1425 GNUNET_MQ_discard (com->env);
1427 send_channel_data_ack (ch);
1430 LOG (GNUNET_ERROR_TYPE_DEBUG,
1431 "Queued %s payload of %u bytes on %s-%X(%p) (mid %u, need %u first)\n",
1432 (GNUNET_YES == ccc->client_ready) ? "out-of-order" : "client-not-ready",
1433 (unsigned int) payload_size,
1435 ntohl (ccc->ccn.channel_of_client),
1437 ntohl (msg->mid.mid),
1438 ntohl (ch->mid_recv.mid));
1439 /* NOTE: this ACK we _could_ skip, as the packet is out-of-order and
1440 the sender may already be transmitting the previous one. Needs
1441 experimental evaluation to see if/when this ACK helps or
1442 hurts. (We might even want another option.) */
1443 send_channel_data_ack (ch);
1448 * Function called once the tunnel has sent one of our messages.
1449 * If the message is unreliable, simply frees the `crm`. If the
1450 * message was reliable, calculate retransmission time and
1451 * wait for ACK (or retransmit).
1453 * @param cls the `struct CadetReliableMessage` that was sent
1454 * @param cid identifier of the connection within the tunnel, NULL
1455 * if transmission failed
1458 data_sent_cb (void *cls,
1459 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid);
1463 * We need to retry a transmission, the last one took too long to
1466 * @param cls the `struct CadetChannel` where we need to retransmit
1469 retry_transmission (void *cls)
1471 struct CadetChannel *ch = cls;
1472 struct CadetReliableMessage *crm = ch->head_sent;
1474 ch->retry_data_task = NULL;
1475 GNUNET_assert (NULL == crm->qe);
1476 LOG (GNUNET_ERROR_TYPE_DEBUG,
1477 "Retrying transmission on %s of message %u\n",
1479 (unsigned int) ntohl (crm->data_message->mid.mid));
1480 crm->qe = GCT_send (ch->t, &crm->data_message->header, &data_sent_cb, crm);
1481 GNUNET_assert (NULL == ch->retry_data_task);
1486 * We got an PLAINTEXT_DATA_ACK for a message in our queue, remove it from
1487 * the queue and tell our client that it can send more.
1489 * @param ch the channel that got the PLAINTEXT_DATA_ACK
1490 * @param cti identifier of the connection that delivered the message
1491 * @param crm the message that got acknowledged
1494 handle_matching_ack (struct CadetChannel *ch,
1495 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1496 struct CadetReliableMessage *crm)
1498 GNUNET_CONTAINER_DLL_remove (ch->head_sent, ch->tail_sent, crm);
1499 ch->pending_messages--;
1500 GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
1501 LOG (GNUNET_ERROR_TYPE_DEBUG,
1502 "Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
1504 (unsigned int) ntohl (crm->data_message->mid.mid),
1505 ch->pending_messages);
1506 if (NULL != crm->qe)
1508 GCT_send_cancel (crm->qe);
1511 if ((1 == crm->num_transmissions) && (NULL != cti))
1513 GCC_ack_observed (cti);
1514 if (0 == GNUNET_memcmp (cti, &crm->connection_taken))
1516 GCC_latency_observed (cti,
1517 GNUNET_TIME_absolute_get_duration (
1518 crm->first_transmission_time));
1521 GNUNET_free (crm->data_message);
1523 send_ack_to_client (ch, (NULL == ch->owner) ? GNUNET_NO : GNUNET_YES);
1528 * We got an acknowledgement for payload data for a channel.
1529 * Possibly resume transmissions.
1531 * @param ch channel that got the ack
1532 * @param cti identifier of the connection that delivered the message
1533 * @param ack details about what was received
1536 GCCH_handle_channel_plaintext_data_ack (
1537 struct CadetChannel *ch,
1538 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1539 const struct GNUNET_CADET_ChannelDataAckMessage *ack)
1541 struct CadetReliableMessage *crm;
1542 struct CadetReliableMessage *crmn;
1548 GNUNET_break (GNUNET_NO == ch->is_loopback);
1549 if (GNUNET_NO == ch->reliable)
1551 /* not expecting ACKs on unreliable channel, odd */
1552 GNUNET_break_op (0);
1555 /* mid_base is the MID of the next message that the
1556 other peer expects (i.e. that is missing!), everything
1557 LOWER (but excluding mid_base itself) was received. */
1558 mid_base = ntohl (ack->mid.mid);
1559 mid_mask = GNUNET_htonll (ack->futures);
1561 for (crm = ch->head_sent; NULL != crm; crm = crmn)
1564 delta = (unsigned int) (ntohl (crm->data_message->mid.mid) - mid_base);
1565 if (delta >= UINT_MAX - ch->max_pending_messages)
1567 /* overflow, means crm was a bit in the past, so this ACK counts for it. */
1568 LOG (GNUNET_ERROR_TYPE_DEBUG,
1569 "Got DATA_ACK with base %u satisfying past message %u on %s\n",
1570 (unsigned int) mid_base,
1571 ntohl (crm->data_message->mid.mid),
1573 handle_matching_ack (ch, cti, crm);
1580 LOG (GNUNET_ERROR_TYPE_DEBUG,
1581 "Testing bit %llX for mid %u (base: %u)\n",
1583 ntohl (crm->data_message->mid.mid),
1585 if (0 != (mid_mask & (1LLU << delta)))
1587 LOG (GNUNET_ERROR_TYPE_DEBUG,
1588 "Got DATA_ACK with mask for %u on %s\n",
1589 ntohl (crm->data_message->mid.mid),
1591 handle_matching_ack (ch, cti, crm);
1595 if (GNUNET_NO == found)
1597 /* ACK for message we already dropped, might have been a
1598 duplicate ACK? Ignore. */
1599 LOG (GNUNET_ERROR_TYPE_DEBUG,
1600 "Duplicate DATA_ACK on %s, ignoring\n",
1602 GNUNET_STATISTICS_update (stats, "# duplicate DATA_ACKs", 1, GNUNET_NO);
1605 if (NULL != ch->retry_data_task)
1607 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1608 ch->retry_data_task = NULL;
1610 if ((NULL != ch->head_sent) && (NULL == ch->head_sent->qe))
1611 ch->retry_data_task = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1612 &retry_transmission,
1618 * Destroy channel, based on the other peer closing the
1619 * connection. Also needs to remove this channel from
1622 * @param ch channel to destroy
1623 * @param cti identifier of the connection that delivered the message,
1624 * NULL if we are simulating receiving a destroy due to shutdown
1627 GCCH_handle_remote_destroy (
1628 struct CadetChannel *ch,
1629 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
1631 struct CadetChannelClient *ccc;
1633 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1634 LOG (GNUNET_ERROR_TYPE_DEBUG,
1635 "Received remote channel DESTROY for %s\n",
1637 if (GNUNET_YES == ch->destroy)
1639 /* Local client already gone, this is instant-death. */
1640 channel_destroy (ch);
1643 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1644 if ((NULL != ccc) && (NULL != ccc->head_recv))
1646 LOG (GNUNET_ERROR_TYPE_WARNING,
1647 "Lost end of transmission due to remote shutdown on %s\n",
1649 /* FIXME: change API to notify client about truncated transmission! */
1651 ch->destroy = GNUNET_YES;
1653 GSC_handle_remote_channel_destroy (ccc->c, ccc->ccn, ch);
1654 channel_destroy (ch);
1659 * Test if element @a e1 comes before element @a e2.
1661 * @param cls closure, to a flag where we indicate duplicate packets
1662 * @param crm1 an element of to sort
1663 * @param crm2 another element to sort
1664 * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1667 cmp_crm_by_next_retry (void *cls,
1668 struct CadetReliableMessage *crm1,
1669 struct CadetReliableMessage *crm2)
1671 if (crm1->next_retry.abs_value_us < crm2->next_retry.abs_value_us)
1678 * Function called once the tunnel has sent one of our messages.
1679 * If the message is unreliable, simply frees the `crm`. If the
1680 * message was reliable, calculate retransmission time and
1681 * wait for ACK (or retransmit).
1683 * @param cls the `struct CadetReliableMessage` that was sent
1684 * @param cid identifier of the connection within the tunnel, NULL
1685 * if transmission failed
1688 data_sent_cb (void *cls,
1689 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
1691 struct CadetReliableMessage *crm = cls;
1692 struct CadetChannel *ch = crm->ch;
1694 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1695 GNUNET_assert (NULL != crm->qe);
1697 GNUNET_CONTAINER_DLL_remove (ch->head_sent, ch->tail_sent, crm);
1698 if (GNUNET_NO == ch->reliable)
1700 GNUNET_free (crm->data_message);
1702 ch->pending_messages--;
1703 send_ack_to_client (ch, (NULL == ch->owner) ? GNUNET_NO : GNUNET_YES);
1708 /* There was an error sending. */
1709 crm->num_transmissions = GNUNET_SYSERR;
1711 else if (GNUNET_SYSERR != crm->num_transmissions)
1713 /* Increment transmission counter, and possibly store @a cid
1714 if this was the first transmission. */
1715 crm->num_transmissions++;
1716 if (1 == crm->num_transmissions)
1718 crm->first_transmission_time = GNUNET_TIME_absolute_get ();
1719 crm->connection_taken = *cid;
1720 GCC_ack_expected (cid);
1723 if ((0 == crm->retry_delay.rel_value_us) && (NULL != cid))
1725 struct CadetConnection *cc = GCC_lookup (cid);
1728 crm->retry_delay = GCC_get_metrics (cc)->aged_latency;
1730 crm->retry_delay = ch->retry_time;
1732 crm->retry_delay = GNUNET_TIME_STD_BACKOFF (crm->retry_delay);
1733 crm->retry_delay = GNUNET_TIME_relative_max (crm->retry_delay, MIN_RTT_DELAY);
1734 crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
1736 GNUNET_CONTAINER_DLL_insert_sorted (struct CadetReliableMessage,
1737 cmp_crm_by_next_retry,
1742 LOG (GNUNET_ERROR_TYPE_DEBUG,
1743 "Message %u sent, next transmission on %s in %s\n",
1744 (unsigned int) ntohl (crm->data_message->mid.mid),
1746 GNUNET_STRINGS_relative_time_to_string (
1747 GNUNET_TIME_absolute_get_remaining (
1748 ch->head_sent->next_retry),
1750 if (NULL == ch->head_sent->qe)
1752 if (NULL != ch->retry_data_task)
1753 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1754 ch->retry_data_task = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1755 &retry_transmission,
1762 * Handle data given by a client.
1764 * Check whether the client is allowed to send in this tunnel, save if
1765 * channel is reliable and send an ACK to the client if there is still
1766 * buffer space in the tunnel.
1768 * @param ch Channel.
1769 * @param sender_ccn ccn of the sender
1770 * @param buf payload to transmit.
1771 * @param buf_len number of bytes in @a buf
1772 * @return #GNUNET_OK if everything goes well,
1773 * #GNUNET_SYSERR in case of an error.
1776 GCCH_handle_local_data (struct CadetChannel *ch,
1777 struct GNUNET_CADET_ClientChannelNumber sender_ccn,
1781 struct CadetReliableMessage *crm;
1783 if (ch->pending_messages >= ch->max_pending_messages)
1785 GNUNET_break (0); /* Fails: #5370 */
1786 return GNUNET_SYSERR;
1788 if (GNUNET_YES == ch->destroy)
1790 /* we are going down, drop messages */
1793 ch->pending_messages++;
1795 if (GNUNET_YES == ch->is_loopback)
1797 struct CadetChannelClient *receiver;
1798 struct GNUNET_MQ_Envelope *env;
1799 struct GNUNET_CADET_LocalData *ld;
1803 GNUNET_MQ_msg_extra (ld, buf_len, GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1804 if ((NULL != ch->owner) &&
1805 (sender_ccn.channel_of_client == ch->owner->ccn.channel_of_client))
1807 receiver = ch->dest;
1808 ack_to_owner = GNUNET_YES;
1810 else if ((NULL != ch->dest) &&
1811 (sender_ccn.channel_of_client == ch->dest->ccn.channel_of_client))
1813 receiver = ch->owner;
1814 ack_to_owner = GNUNET_NO;
1819 return GNUNET_SYSERR;
1821 GNUNET_assert (NULL != receiver);
1822 ld->ccn = receiver->ccn;
1823 GNUNET_memcpy (&ld[1], buf, buf_len);
1824 if (GNUNET_YES == receiver->client_ready)
1826 ch->pending_messages--;
1827 GSC_send_to_client (receiver->c, env);
1828 send_ack_to_client (ch, ack_to_owner);
1832 struct CadetOutOfOrderMessage *oom;
1834 oom = GNUNET_new (struct CadetOutOfOrderMessage);
1836 GNUNET_CONTAINER_DLL_insert_tail (receiver->head_recv,
1837 receiver->tail_recv,
1839 receiver->num_recv++;
1844 /* Everything is correct, send the message. */
1845 crm = GNUNET_malloc (sizeof(*crm));
1847 crm->data_message = GNUNET_malloc (
1848 sizeof(struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1849 crm->data_message->header.size =
1850 htons (sizeof(struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1851 crm->data_message->header.type =
1852 htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
1853 ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
1854 crm->data_message->mid = ch->mid_send;
1855 crm->data_message->ctn = ch->ctn;
1856 GNUNET_memcpy (&crm->data_message[1], buf, buf_len);
1857 GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent, ch->tail_sent, crm);
1858 LOG (GNUNET_ERROR_TYPE_DEBUG,
1859 "Sending message %u from local client to %s with %u bytes\n",
1860 ntohl (crm->data_message->mid.mid),
1863 if (NULL != ch->retry_data_task)
1865 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1866 ch->retry_data_task = NULL;
1868 crm->qe = GCT_send (ch->t, &crm->data_message->header, &data_sent_cb, crm);
1869 GNUNET_assert (NULL == ch->retry_data_task);
1875 * Handle ACK from client on local channel. Means the client is ready
1876 * for more data, see if we have any for it.
1878 * @param ch channel to destroy
1879 * @param client_ccn ccn of the client sending the ack
1882 GCCH_handle_local_ack (struct CadetChannel *ch,
1883 struct GNUNET_CADET_ClientChannelNumber client_ccn)
1885 struct CadetChannelClient *ccc;
1886 struct CadetOutOfOrderMessage *com;
1888 if ((NULL != ch->owner) &&
1889 (ch->owner->ccn.channel_of_client == client_ccn.channel_of_client))
1891 else if ((NULL != ch->dest) &&
1892 (ch->dest->ccn.channel_of_client == client_ccn.channel_of_client))
1896 ccc->client_ready = GNUNET_YES;
1897 com = ccc->head_recv;
1900 LOG (GNUNET_ERROR_TYPE_DEBUG,
1901 "Got LOCAL_ACK, %s-%X ready to receive more data, but none pending on %s-%X(%p)!\n",
1903 ntohl (client_ccn.channel_of_client),
1905 ntohl (ccc->ccn.channel_of_client),
1907 return; /* none pending */
1909 if (GNUNET_YES == ch->is_loopback)
1913 /* Messages are always in-order, just send */
1914 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, com);
1916 GSC_send_to_client (ccc->c, com->env);
1917 /* Notify sender that we can receive more */
1918 if ((NULL != ch->owner) &&
1919 (ccc->ccn.channel_of_client == ch->owner->ccn.channel_of_client))
1921 to_owner = GNUNET_NO;
1925 GNUNET_assert ((NULL != ch->dest) && (ccc->ccn.channel_of_client ==
1926 ch->dest->ccn.channel_of_client));
1927 to_owner = GNUNET_YES;
1929 send_ack_to_client (ch, to_owner);
1934 if ((com->mid.mid != ch->mid_recv.mid) && (GNUNET_NO == ch->out_of_order) &&
1935 (GNUNET_YES == ch->reliable))
1937 LOG (GNUNET_ERROR_TYPE_DEBUG,
1938 "Got LOCAL_ACK, %s-%X ready to receive more data (but next one is out-of-order %u vs. %u)!\n",
1940 ntohl (ccc->ccn.channel_of_client),
1941 ntohl (com->mid.mid),
1942 ntohl (ch->mid_recv.mid));
1943 return; /* missing next one in-order */
1946 LOG (GNUNET_ERROR_TYPE_DEBUG,
1947 "Got LOCAL_ACK, giving payload message %u to %s-%X on %s\n",
1948 ntohl (com->mid.mid),
1950 ntohl (ccc->ccn.channel_of_client),
1953 /* all good, pass next message to client */
1954 GNUNET_CONTAINER_DLL_remove (ccc->head_recv, ccc->tail_recv, com);
1956 /* FIXME: if unreliable, this is not aggressive
1957 enough, as it would be OK to have lost some! */
1959 ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
1960 ch->mid_futures >>= 1; /* equivalent to division by 2 */
1961 ccc->client_ready = GNUNET_NO;
1962 GSC_send_to_client (ccc->c, com->env);
1964 send_channel_data_ack (ch);
1965 if (NULL != ccc->head_recv)
1967 if (GNUNET_NO == ch->destroy)
1969 GCT_send_channel_destroy (ch->t, ch->ctn);
1970 channel_destroy (ch);
1974 #define LOG2(level, ...) \
1975 GNUNET_log_from_nocheck (level, "cadet-chn", __VA_ARGS__)
1981 * @param ch Channel.
1982 * @param level Debug level to use.
1985 GCCH_debug (struct CadetChannel *ch, enum GNUNET_ErrorType level)
1987 #if ! defined(GNUNET_CULL_LOGGING)
1990 do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
2000 LOG2 (level, "CHN *** DEBUG NULL CHANNEL ***\n");
2003 LOG2 (level, "CHN %s:%X (%p)\n", GCT_2s (ch->t), ch->ctn, ch);
2004 if (NULL != ch->owner)
2007 "CHN origin %s ready %s local-id: %u\n",
2008 GSC_2s (ch->owner->c),
2009 ch->owner->client_ready ? "YES" : "NO",
2010 ntohl (ch->owner->ccn.channel_of_client));
2012 if (NULL != ch->dest)
2015 "CHN destination %s ready %s local-id: %u\n",
2016 GSC_2s (ch->dest->c),
2017 ch->dest->client_ready ? "YES" : "NO",
2018 ntohl (ch->dest->ccn.channel_of_client));
2021 "CHN Message IDs recv: %d (%LLX), send: %d\n",
2022 ntohl (ch->mid_recv.mid),
2023 (unsigned long long) ch->mid_futures,
2024 ntohl (ch->mid_send.mid));
2029 /* end of gnunet-service-cadet_channel.c */