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 GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, 250)
52 * How long do we wait before dropping state about incoming
53 * connection to closed port?
55 #define TIMEOUT_CLOSED_PORT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 30)
58 * How long do we wait at least before retransmitting ever?
60 #define MIN_RTT_DELAY GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, 75)
63 * Maximum message ID into the future we accept for out-of-order messages.
64 * If the message is more than this into the future, we drop it. This is
65 * important both to detect values that are actually in the past, as well
66 * as to limit adversarially triggerable memory consumption.
68 * Note that right now we have "max_pending_messages = 4" hard-coded in
69 * the logic below, so a value of 4 would suffice here. But we plan to
70 * allow larger windows in the future...
72 #define MAX_OUT_OF_ORDER_DISTANCE 1024
76 * All the states a channel can be in.
78 enum CadetChannelState
81 * Uninitialized status, should never appear in operation.
86 * Channel is to a port that is not open, we're waiting for the
92 * CHANNEL_OPEN message sent, waiting for CHANNEL_OPEN_ACK.
94 CADET_CHANNEL_OPEN_SENT,
97 * Connection confirmed, ready to carry traffic.
104 * Info needed to retry a message in case it gets lost.
105 * Note that we DO use this structure also for unreliable
108 struct CadetReliableMessage
111 * Double linked list, FIFO style
113 struct CadetReliableMessage *next;
116 * Double linked list, FIFO style
118 struct CadetReliableMessage *prev;
121 * Which channel is this message in?
123 struct CadetChannel *ch;
126 * Entry in the tunnels queue for this message, NULL if it has left
127 * the tunnel. Used to cancel transmission in case we receive an
130 struct CadetTunnelQueueEntry *qe;
133 * Data message we are trying to send.
135 struct GNUNET_CADET_ChannelAppDataMessage *data_message;
138 * How soon should we retry if we fail to get an ACK?
139 * Messages in the queue are sorted by this value.
141 struct GNUNET_TIME_Absolute next_retry;
144 * How long do we wait for an ACK after transmission?
145 * Use for the back-off calculation.
147 struct GNUNET_TIME_Relative retry_delay;
150 * Time when we first successfully transmitted the message
151 * (that is, set @e num_transmissions to 1).
153 struct GNUNET_TIME_Absolute first_transmission_time;
156 * Identifier of the connection that this message took when it
157 * was first transmitted. Only useful if @e num_transmissions is 1.
159 struct GNUNET_CADET_ConnectionTunnelIdentifier connection_taken;
162 * How often was this message transmitted? #GNUNET_SYSERR if there
163 * was an error transmitting the message, #GNUNET_NO if it was not
164 * yet transmitted ever, otherwise the number of (re) transmissions.
166 int num_transmissions;
172 * List of received out-of-order data messages.
174 struct CadetOutOfOrderMessage
177 * Double linked list, FIFO style
179 struct CadetOutOfOrderMessage *next;
182 * Double linked list, FIFO style
184 struct CadetOutOfOrderMessage *prev;
187 * ID of the message (messages up to this point needed
188 * before we give this one to the client).
190 struct ChannelMessageIdentifier mid;
193 * The envelope with the payload of the out-of-order message
195 struct GNUNET_MQ_Envelope *env;
201 * Client endpoint of a `struct CadetChannel`. A channel may be a
202 * loopback channel, in which case it has two of these endpoints.
203 * Note that flow control also is required in both directions.
205 struct CadetChannelClient
208 * Client handle. Not by itself sufficient to designate
209 * the client endpoint, as the same client handle may
210 * be used for both the owner and the destination, and
211 * we thus also need the channel ID to identify the client.
213 struct CadetClient *c;
216 * Head of DLL of messages received out of order or while client was unready.
218 struct CadetOutOfOrderMessage *head_recv;
221 * Tail DLL of messages received out of order or while client was unready.
223 struct CadetOutOfOrderMessage *tail_recv;
226 * Local tunnel number for this client.
227 * (if owner >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI,
228 * otherwise < #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
230 struct GNUNET_CADET_ClientChannelNumber ccn;
233 * Number of entries currently in @a head_recv DLL.
235 unsigned int num_recv;
238 * 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
388 * Get the static string for identification of the channel.
392 * @return Static string with the channel IDs.
395 GCCH_2s (const struct CadetChannel *ch)
397 static char buf[128];
399 GNUNET_snprintf (buf,
401 "Channel %s:%s ctn:%X(%X/%X)",
402 (GNUNET_YES == ch->is_loopback)
404 : GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
405 GNUNET_h2s (&ch->port),
407 (NULL == ch->owner) ? 0 : ntohl (ch->owner->ccn.channel_of_client),
408 (NULL == ch->dest) ? 0 : ntohl (ch->dest->ccn.channel_of_client));
414 * Hash the @a port and @a initiator and @a listener to
415 * calculate the "challenge" @a h_port we send to the other
416 * peer on #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN.
418 * @param[out] h_port set to the hash of @a port, @a initiator and @a listener
419 * @param port cadet port, as seen by CADET clients
420 * @param listener peer that is listining on @a port
423 GCCH_hash_port (struct GNUNET_HashCode *h_port,
424 const struct GNUNET_HashCode *port,
425 const struct GNUNET_PeerIdentity *listener)
427 struct GNUNET_HashContext *hc;
429 hc = GNUNET_CRYPTO_hash_context_start ();
430 GNUNET_CRYPTO_hash_context_read (hc,
433 GNUNET_CRYPTO_hash_context_read (hc,
436 GNUNET_CRYPTO_hash_context_finish (hc,
438 LOG (GNUNET_ERROR_TYPE_DEBUG,
439 "Calculated port hash %s\n",
440 GNUNET_h2s (h_port));
445 * Get the channel's public ID.
449 * @return ID used to identify the channel with the remote peer.
451 struct GNUNET_CADET_ChannelTunnelNumber
452 GCCH_get_id (const struct CadetChannel *ch)
459 * Release memory associated with @a ccc
461 * @param ccc data structure to clean up
464 free_channel_client (struct CadetChannelClient *ccc)
466 struct CadetOutOfOrderMessage *com;
468 while (NULL != (com = ccc->head_recv))
470 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
474 GNUNET_MQ_discard (com->env);
482 * Destroy the given channel.
484 * @param ch channel to destroy
487 channel_destroy (struct CadetChannel *ch)
489 struct CadetReliableMessage *crm;
491 while (NULL != (crm = ch->head_sent))
493 GNUNET_assert (ch == crm->ch);
496 GCT_send_cancel (crm->qe);
499 GNUNET_CONTAINER_DLL_remove (ch->head_sent,
502 GNUNET_free (crm->data_message);
505 if (CADET_CHANNEL_LOOSE == ch->state)
507 GSC_drop_loose_channel (&ch->h_port,
510 if (NULL != ch->owner)
512 free_channel_client (ch->owner);
515 if (NULL != ch->dest)
517 free_channel_client (ch->dest);
520 if (NULL != ch->last_control_qe)
522 GCT_send_cancel (ch->last_control_qe);
523 ch->last_control_qe = NULL;
525 if (NULL != ch->retry_data_task)
527 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
528 ch->retry_data_task = NULL;
530 if (NULL != ch->retry_control_task)
532 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
533 ch->retry_control_task = NULL;
535 if (GNUNET_NO == ch->is_loopback)
537 GCT_remove_channel (ch->t,
547 * Send a channel create message.
549 * @param cls Channel for which to send.
552 send_channel_open (void *cls);
556 * Function called once the tunnel confirms that we sent the
557 * create message. Delays for a bit until we retry.
559 * @param cls our `struct CadetChannel`.
560 * @param cid identifier of the connection within the tunnel, NULL
561 * if transmission failed
564 channel_open_sent_cb (void *cls,
565 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
567 struct CadetChannel *ch = cls;
569 GNUNET_assert (NULL != ch->last_control_qe);
570 ch->last_control_qe = NULL;
571 ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
572 LOG (GNUNET_ERROR_TYPE_DEBUG,
573 "Sent CADET_CHANNEL_OPEN on %s, retrying in %s\n",
575 GNUNET_STRINGS_relative_time_to_string (ch->retry_time,
577 ch->retry_control_task
578 = GNUNET_SCHEDULER_add_delayed (ch->retry_time,
585 * Send a channel open message.
587 * @param cls Channel for which to send.
590 send_channel_open (void *cls)
592 struct CadetChannel *ch = cls;
593 struct GNUNET_CADET_ChannelOpenMessage msgcc;
596 ch->retry_control_task = NULL;
597 LOG (GNUNET_ERROR_TYPE_DEBUG,
598 "Sending CHANNEL_OPEN message for %s\n",
602 options |= GNUNET_CADET_OPTION_NOBUFFER;
604 options |= GNUNET_CADET_OPTION_RELIABLE;
605 if (ch->out_of_order)
606 options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
607 msgcc.header.size = htons (sizeof (msgcc));
608 msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN);
609 msgcc.opt = htonl (options);
610 msgcc.h_port = ch->h_port;
612 ch->state = CADET_CHANNEL_OPEN_SENT;
613 if (NULL != ch->last_control_qe)
614 GCT_send_cancel (ch->last_control_qe);
615 ch->last_control_qe = GCT_send (ch->t,
617 &channel_open_sent_cb,
619 GNUNET_assert (NULL == ch->retry_control_task);
624 * Function called once and only once after a channel was bound
625 * to its tunnel via #GCT_add_channel() is ready for transmission.
626 * Note that this is only the case for channels that this peer
627 * initiates, as for incoming channels we assume that they are
628 * ready for transmission immediately upon receiving the open
629 * message. Used to bootstrap the #GCT_send() process.
631 * @param ch the channel for which the tunnel is now ready
634 GCCH_tunnel_up (struct CadetChannel *ch)
636 GNUNET_assert (NULL == ch->retry_control_task);
637 LOG (GNUNET_ERROR_TYPE_DEBUG,
638 "Tunnel up, sending CHANNEL_OPEN on %s now\n",
640 ch->retry_control_task
641 = GNUNET_SCHEDULER_add_now (&send_channel_open,
647 * Create a new channel.
649 * @param owner local client owning the channel
650 * @param ccn local number of this channel at the @a owner
651 * @param destination peer to which we should build the channel
652 * @param port desired port at @a destination
653 * @param options options for the channel
654 * @return handle to the new channel
656 struct CadetChannel *
657 GCCH_channel_local_new (struct CadetClient *owner,
658 struct GNUNET_CADET_ClientChannelNumber ccn,
659 struct CadetPeer *destination,
660 const struct GNUNET_HashCode *port,
663 struct CadetChannel *ch;
664 struct CadetChannelClient *ccco;
666 ccco = GNUNET_new (struct CadetChannelClient);
669 ccco->client_ready = GNUNET_YES;
671 ch = GNUNET_new (struct CadetChannel);
672 ch->mid_recv.mid = htonl (1); /* The OPEN_ACK counts as message 0! */
673 ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
674 ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
675 ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
676 ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
679 GCCH_hash_port (&ch->h_port,
681 GCP_get_id (destination));
682 if (0 == memcmp (&my_full_id,
683 GCP_get_id (destination),
684 sizeof (struct GNUNET_PeerIdentity)))
688 ch->is_loopback = GNUNET_YES;
689 op = GNUNET_CONTAINER_multihashmap_get (open_ports,
693 /* port closed, wait for it to possibly open */
694 ch->state = CADET_CHANNEL_LOOSE;
695 (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
698 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
699 LOG (GNUNET_ERROR_TYPE_DEBUG,
700 "Created loose incoming loopback channel to port %s\n",
701 GNUNET_h2s (&ch->port));
712 ch->t = GCP_get_tunnel (destination,
714 ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
715 ch->ctn = GCT_add_channel (ch->t,
718 GNUNET_STATISTICS_update (stats,
722 LOG (GNUNET_ERROR_TYPE_DEBUG,
723 "Created channel to port %s at peer %s for %s using %s\n",
725 GCP_2s (destination),
727 (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
733 * We had an incoming channel to a port that is closed.
734 * It has not been opened for a while, drop it.
736 * @param cls the channel to drop
739 timeout_closed_cb (void *cls)
741 struct CadetChannel *ch = cls;
743 ch->retry_control_task = NULL;
744 LOG (GNUNET_ERROR_TYPE_DEBUG,
745 "Closing incoming channel to port %s from peer %s due to timeout\n",
746 GNUNET_h2s (&ch->port),
747 GCP_2s (GCT_get_destination (ch->t)));
748 channel_destroy (ch);
753 * Create a new channel based on a request coming in over the network.
755 * @param t tunnel to the remote peer
756 * @param ctn identifier of this channel in the tunnel
757 * @param h_port desired hash of local port
758 * @param options options for the channel
759 * @return handle to the new channel
761 struct CadetChannel *
762 GCCH_channel_incoming_new (struct CadetTunnel *t,
763 struct GNUNET_CADET_ChannelTunnelNumber ctn,
764 const struct GNUNET_HashCode *h_port,
767 struct CadetChannel *ch;
770 ch = GNUNET_new (struct CadetChannel);
771 ch->h_port = *h_port;
774 ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
775 ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
776 ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
777 ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
778 ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
779 GNUNET_STATISTICS_update (stats,
784 op = GNUNET_CONTAINER_multihashmap_get (open_ports,
788 /* port closed, wait for it to possibly open */
789 ch->state = CADET_CHANNEL_LOOSE;
790 (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
793 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
794 GNUNET_assert (NULL == ch->retry_control_task);
795 ch->retry_control_task
796 = GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
799 LOG (GNUNET_ERROR_TYPE_DEBUG,
800 "Created loose incoming channel to port %s from peer %s\n",
801 GNUNET_h2s (&ch->port),
802 GCP_2s (GCT_get_destination (ch->t)));
810 GNUNET_STATISTICS_update (stats,
819 * Function called once the tunnel confirms that we sent the
820 * ACK message. Just remembers it was sent, we do not expect
823 * @param cls our `struct CadetChannel`.
824 * @param cid identifier of the connection within the tunnel, NULL
825 * if transmission failed
828 send_ack_cb (void *cls,
829 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
831 struct CadetChannel *ch = cls;
833 GNUNET_assert (NULL != ch->last_control_qe);
834 ch->last_control_qe = NULL;
839 * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
841 * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
844 send_channel_data_ack (struct CadetChannel *ch)
846 struct GNUNET_CADET_ChannelDataAckMessage msg;
848 if (GNUNET_NO == ch->reliable)
849 return; /* no ACKs */
850 msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK);
851 msg.header.size = htons (sizeof (msg));
853 msg.mid.mid = htonl (ntohl (ch->mid_recv.mid));
854 msg.futures = GNUNET_htonll (ch->mid_futures);
855 LOG (GNUNET_ERROR_TYPE_DEBUG,
856 "Sending DATA_ACK %u:%llX via %s\n",
857 (unsigned int) ntohl (msg.mid.mid),
858 (unsigned long long) ch->mid_futures,
860 if (NULL != ch->last_control_qe)
861 GCT_send_cancel (ch->last_control_qe);
862 ch->last_control_qe = GCT_send (ch->t,
870 * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
873 * @param cls the `struct CadetChannel`
876 send_open_ack (void *cls)
878 struct CadetChannel *ch = cls;
879 struct GNUNET_CADET_ChannelOpenAckMessage msg;
881 ch->retry_control_task = NULL;
882 LOG (GNUNET_ERROR_TYPE_DEBUG,
883 "Sending CHANNEL_OPEN_ACK on %s\n",
885 msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
886 msg.header.size = htons (sizeof (msg));
887 msg.reserved = htonl (0);
890 if (NULL != ch->last_control_qe)
891 GCT_send_cancel (ch->last_control_qe);
892 ch->last_control_qe = GCT_send (ch->t,
900 * We got a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN message again for
901 * this channel. If the binding was successful, (re)transmit the
902 * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
904 * @param ch channel that got the duplicate open
905 * @param cti identifier of the connection that delivered the message
908 GCCH_handle_duplicate_open (struct CadetChannel *ch,
909 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
911 if (NULL == ch->dest)
913 LOG (GNUNET_ERROR_TYPE_DEBUG,
914 "Ignoring duplicate CHANNEL_OPEN on %s: port is closed\n",
918 if (NULL != ch->retry_control_task)
920 LOG (GNUNET_ERROR_TYPE_DEBUG,
921 "Ignoring duplicate CHANNEL_OPEN on %s: control message is pending\n",
925 LOG (GNUNET_ERROR_TYPE_DEBUG,
926 "Retransmitting CHANNEL_OPEN_ACK on %s\n",
928 ch->retry_control_task
929 = GNUNET_SCHEDULER_add_now (&send_open_ack,
935 * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK to the client to solicit more messages.
937 * @param ch channel the ack is for
938 * @param to_owner #GNUNET_YES to send to owner,
939 * #GNUNET_NO to send to dest
942 send_ack_to_client (struct CadetChannel *ch,
945 struct GNUNET_MQ_Envelope *env;
946 struct GNUNET_CADET_LocalAck *ack;
947 struct CadetChannelClient *ccc;
949 ccc = (GNUNET_YES == to_owner) ? ch->owner : ch->dest;
952 /* This can happen if we are just getting ACKs after
953 our local client already disconnected. */
954 GNUNET_assert (GNUNET_YES == ch->destroy);
957 env = GNUNET_MQ_msg (ack,
958 GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
960 LOG (GNUNET_ERROR_TYPE_DEBUG,
961 "Sending CADET_LOCAL_ACK to %s (%s) at ccn %X (%u/%u pending)\n",
963 (GNUNET_YES == to_owner) ? "owner" : "dest",
964 ntohl (ack->ccn.channel_of_client),
965 ch->pending_messages,
966 ch->max_pending_messages);
967 GSC_send_to_client (ccc->c,
973 * A client is bound to the port that we have a channel
974 * open to. Send the acknowledgement for the connection
975 * request and establish the link with the client.
977 * @param ch open incoming channel
978 * @param c client listening on the respective @a port
979 * @param port the port @a is listening on
982 GCCH_bind (struct CadetChannel *ch,
983 struct CadetClient *c,
984 const struct GNUNET_HashCode *port)
987 struct CadetChannelClient *cccd;
989 LOG (GNUNET_ERROR_TYPE_DEBUG,
990 "Binding %s from %s to port %s of %s\n",
993 GNUNET_h2s (&ch->port),
995 if (NULL != ch->retry_control_task)
997 /* there might be a timeout task here */
998 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
999 ch->retry_control_task = NULL;
1003 options |= GNUNET_CADET_OPTION_NOBUFFER;
1005 options |= GNUNET_CADET_OPTION_RELIABLE;
1006 if (ch->out_of_order)
1007 options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
1008 cccd = GNUNET_new (struct CadetChannelClient);
1009 GNUNET_assert (NULL == ch->dest);
1013 cccd->client_ready = GNUNET_YES;
1014 cccd->ccn = GSC_bind (c,
1016 (GNUNET_YES == ch->is_loopback)
1017 ? GCP_get (&my_full_id,
1019 : GCT_get_destination (ch->t),
1022 GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
1023 GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
1024 ch->mid_recv.mid = htonl (1); /* The OPEN counts as message 0! */
1025 if (GNUNET_YES == ch->is_loopback)
1027 ch->state = CADET_CHANNEL_OPEN_SENT;
1028 GCCH_handle_channel_open_ack (ch,
1034 /* notify other peer that we accepted the connection */
1035 ch->state = CADET_CHANNEL_READY;
1036 ch->retry_control_task
1037 = GNUNET_SCHEDULER_add_now (&send_open_ack,
1040 /* give client it's initial supply of ACKs */
1041 GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
1042 GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
1043 for (unsigned int i=0;i<ch->max_pending_messages;i++)
1044 send_ack_to_client (ch,
1050 * One of our clients has disconnected, tell the other one that we
1051 * are finished. Done asynchronously to avoid concurrent modification
1052 * issues if this is the same client.
1054 * @param cls the `struct CadetChannel` where one of the ends is now dead
1057 signal_remote_destroy_cb (void *cls)
1059 struct CadetChannel *ch = cls;
1060 struct CadetChannelClient *ccc;
1062 /* Find which end is left... */
1063 ch->retry_control_task = NULL;
1064 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1065 GSC_handle_remote_channel_destroy (ccc->c,
1068 channel_destroy (ch);
1073 * Destroy locally created channel. Called by the local client, so no
1074 * need to tell the client.
1076 * @param ch channel to destroy
1077 * @param c client that caused the destruction
1078 * @param ccn client number of the client @a c
1081 GCCH_channel_local_destroy (struct CadetChannel *ch,
1082 struct CadetClient *c,
1083 struct GNUNET_CADET_ClientChannelNumber ccn)
1085 LOG (GNUNET_ERROR_TYPE_DEBUG,
1086 "%s asks for destruction of %s\n",
1089 GNUNET_assert (NULL != c);
1090 if ( (NULL != ch->owner) &&
1091 (c == ch->owner->c) &&
1092 (ccn.channel_of_client == ch->owner->ccn.channel_of_client) )
1094 free_channel_client (ch->owner);
1097 else if ( (NULL != ch->dest) &&
1098 (c == ch->dest->c) &&
1099 (ccn.channel_of_client == ch->dest->ccn.channel_of_client) )
1101 free_channel_client (ch->dest);
1109 if (GNUNET_YES == ch->destroy)
1111 /* other end already destroyed, with the local client gone, no need
1112 to finish transmissions, just destroy immediately. */
1113 channel_destroy (ch);
1116 if ( (NULL != ch->head_sent) &&
1117 ( (NULL != ch->owner) ||
1118 (NULL != ch->dest) ) )
1120 /* Wait for other end to destroy us as well,
1121 and otherwise allow send queue to be transmitted first */
1122 ch->destroy = GNUNET_YES;
1125 if ( (GNUNET_YES == ch->is_loopback) &&
1126 ( (NULL != ch->owner) ||
1127 (NULL != ch->dest) ) )
1129 if (NULL != ch->retry_control_task)
1130 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
1131 ch->retry_control_task
1132 = GNUNET_SCHEDULER_add_now (&signal_remote_destroy_cb,
1136 if (GNUNET_NO == ch->is_loopback)
1138 /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
1141 case CADET_CHANNEL_NEW:
1142 /* We gave up on a channel that we created as a client to a remote
1143 target, but that never went anywhere. Nothing to do here. */
1145 case CADET_CHANNEL_LOOSE:
1148 GCT_send_channel_destroy (ch->t,
1152 /* Nothing left to do, just finish destruction */
1153 channel_destroy (ch);
1158 * We got an acknowledgement for the creation of the channel
1159 * (the port is open on the other side). Verify that the
1160 * other end really has the right port, and begin transmissions.
1162 * @param ch channel to destroy
1163 * @param cti identifier of the connection that delivered the message
1164 * @param port port number (needed to verify receiver knows the port)
1167 GCCH_handle_channel_open_ack (struct CadetChannel *ch,
1168 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1169 const struct GNUNET_HashCode *port)
1173 case CADET_CHANNEL_NEW:
1174 /* this should be impossible */
1177 case CADET_CHANNEL_LOOSE:
1178 /* This makes no sense. */
1179 GNUNET_break_op (0);
1181 case CADET_CHANNEL_OPEN_SENT:
1182 if (NULL == ch->owner)
1184 /* We're not the owner, wrong direction! */
1185 GNUNET_break_op (0);
1188 if (0 != memcmp (&ch->port,
1190 sizeof (struct GNUNET_HashCode)))
1192 /* Other peer failed to provide the right port,
1193 refuse connection. */
1194 GNUNET_break_op (0);
1197 LOG (GNUNET_ERROR_TYPE_DEBUG,
1198 "Received CHANNEL_OPEN_ACK for waiting %s, entering READY state\n",
1200 if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
1202 GNUNET_SCHEDULER_cancel (ch->retry_control_task);
1203 ch->retry_control_task = NULL;
1205 ch->state = CADET_CHANNEL_READY;
1206 /* On first connect, send client as many ACKs as we allow messages
1208 for (unsigned int i=0;i<ch->max_pending_messages;i++)
1209 send_ack_to_client (ch,
1212 case CADET_CHANNEL_READY:
1213 /* duplicate ACK, maybe we retried the CREATE. Ignore. */
1214 LOG (GNUNET_ERROR_TYPE_DEBUG,
1215 "Received duplicate channel OPEN_ACK for %s\n",
1217 GNUNET_STATISTICS_update (stats,
1218 "# duplicate CREATE_ACKs",
1227 * Test if element @a e1 comes before element @a e2.
1229 * @param cls closure, to a flag where we indicate duplicate packets
1230 * @param m1 a message of to sort
1231 * @param m2 another message to sort
1232 * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1235 is_before (void *cls,
1236 struct CadetOutOfOrderMessage *m1,
1237 struct CadetOutOfOrderMessage *m2)
1239 int *duplicate = cls;
1240 uint32_t v1 = ntohl (m1->mid.mid);
1241 uint32_t v2 = ntohl (m2->mid.mid);
1246 *duplicate = GNUNET_YES;
1247 if (delta > (uint32_t) INT_MAX)
1249 /* in overflow range, we can safely assume we wrapped around */
1254 /* result is small, thus v2 > v1, thus m1 < m2 */
1261 * We got payload data for a channel. Pass it on to the client
1262 * and send an ACK to the other end (once flow control allows it!)
1264 * @param ch channel that got data
1265 * @param cti identifier of the connection that delivered the message
1266 * @param msg message that was received
1269 GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
1270 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1271 const struct GNUNET_CADET_ChannelAppDataMessage *msg)
1273 struct GNUNET_MQ_Envelope *env;
1274 struct GNUNET_CADET_LocalData *ld;
1275 struct CadetChannelClient *ccc;
1276 size_t payload_size;
1277 struct CadetOutOfOrderMessage *com;
1284 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1285 if ( (NULL == ch->owner) &&
1286 (NULL == ch->dest) )
1288 /* This client is gone, but we still have messages to send to
1289 the other end (which is why @a ch is not yet dead). However,
1290 we cannot pass messages to our client anymore. */
1291 LOG (GNUNET_ERROR_TYPE_DEBUG,
1292 "Dropping incoming payload on %s as this end is already closed\n",
1294 /* send back DESTROY notification to stop further retransmissions! */
1295 if (GNUNET_YES == ch->destroy)
1296 GCT_send_channel_destroy (ch->t,
1300 payload_size = ntohs (msg->header.size) - sizeof (*msg);
1301 env = GNUNET_MQ_msg_extra (ld,
1303 GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1304 ld->ccn = (NULL == ch->dest) ? ch->owner->ccn : ch->dest->ccn;
1305 GNUNET_memcpy (&ld[1],
1308 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1309 if (GNUNET_YES == ccc->client_ready)
1312 * We ad-hoc send the message if
1313 * - The channel is out-of-order
1314 * - The channel is reliable and MID matches next expected MID
1315 * - The channel is unreliable and MID is before lowest seen MID
1317 if ( (GNUNET_YES == ch->out_of_order) ||
1318 ((msg->mid.mid == ch->mid_recv.mid) &&
1319 (GNUNET_YES == ch->reliable)) ||
1320 ((GNUNET_NO == ch->reliable) &&
1321 (ntohl (msg->mid.mid) >= ntohl (ch->mid_recv.mid)) &&
1322 ((NULL == ccc->head_recv) ||
1323 (ntohl (msg->mid.mid) < ntohl (ccc->head_recv->mid.mid)))) )
1325 LOG (GNUNET_ERROR_TYPE_DEBUG,
1326 "Giving %u bytes of payload with MID %u from %s to client %s\n",
1327 (unsigned int) payload_size,
1328 ntohl (msg->mid.mid),
1331 ccc->client_ready = GNUNET_NO;
1332 GSC_send_to_client (ccc->c,
1334 if (GNUNET_NO == ch->out_of_order)
1335 ch->mid_recv.mid = htonl (1 + ntohl (msg->mid.mid));
1337 ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
1338 ch->mid_futures >>= 1;
1339 if ( (GNUNET_YES == ch->out_of_order) &&
1340 (GNUNET_NO == ch->reliable) )
1342 /* possibly shift by more if we skipped messages */
1343 uint64_t delta = htonl (msg->mid.mid) - 1 - ntohl (ch->mid_recv.mid);
1346 ch->mid_futures = 0;
1348 ch->mid_futures >>= delta;
1349 ch->mid_recv.mid = htonl (1 + ntohl (msg->mid.mid));
1351 send_channel_data_ack (ch);
1356 if (GNUNET_YES == ch->reliable)
1358 /* check if message ought to be dropped because it is ancient/too distant/duplicate */
1359 mid_min = ntohl (ch->mid_recv.mid);
1360 mid_max = mid_min + ch->max_pending_messages;
1361 mid_msg = ntohl (msg->mid.mid);
1362 if ( ( (uint32_t) (mid_msg - mid_min) > ch->max_pending_messages) ||
1363 ( (uint32_t) (mid_max - mid_msg) > ch->max_pending_messages) )
1365 LOG (GNUNET_ERROR_TYPE_DEBUG,
1366 "%s at %u drops ancient or far-future message %u\n",
1368 (unsigned int) mid_min,
1369 ntohl (msg->mid.mid));
1371 GNUNET_STATISTICS_update (stats,
1372 "# duplicate DATA (ancient or future)",
1375 GNUNET_MQ_discard (env);
1376 send_channel_data_ack (ch);
1379 /* mark bit for future ACKs */
1380 delta = mid_msg - mid_min - 1; /* overflow/underflow are OK here */
1383 if (0 != (ch->mid_futures & (1LLU << delta)))
1385 /* Duplicate within the queue, drop also */
1386 LOG (GNUNET_ERROR_TYPE_DEBUG,
1387 "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1388 (unsigned int) payload_size,
1390 ntohl (msg->mid.mid));
1391 GNUNET_STATISTICS_update (stats,
1395 GNUNET_MQ_discard (env);
1396 send_channel_data_ack (ch);
1399 ch->mid_futures |= (1LLU << delta);
1400 LOG (GNUNET_ERROR_TYPE_DEBUG,
1401 "Marked bit %llX for mid %u (base: %u); now: %llX\n",
1408 else /* ! ch->reliable */
1410 struct CadetOutOfOrderMessage *next_msg;
1413 * We always send if possible in this case.
1414 * It is guaranteed that the queued MID < received MID
1416 if ((NULL != ccc->head_recv) &&
1417 (GNUNET_YES == ccc->client_ready))
1419 next_msg = ccc->head_recv;
1420 LOG (GNUNET_ERROR_TYPE_DEBUG,
1421 "Giving queued MID %u from %s to client %s\n",
1422 ntohl (next_msg->mid.mid),
1425 ccc->client_ready = GNUNET_NO;
1426 GSC_send_to_client (ccc->c,
1428 ch->mid_recv.mid = htonl (1 + ntohl (next_msg->mid.mid));
1429 ch->mid_futures >>= 1;
1430 send_channel_data_ack (ch);
1431 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1435 /* Do not process duplicate MID */
1436 if (msg->mid.mid == next_msg->mid.mid) /* Duplicate */
1438 /* Duplicate within the queue, drop */
1439 LOG (GNUNET_ERROR_TYPE_DEBUG,
1440 "Message on %s (mid %u) dropped, duplicate\n",
1442 ntohl (msg->mid.mid));
1443 GNUNET_free (next_msg);
1444 GNUNET_MQ_discard (env);
1447 GNUNET_free (next_msg);
1450 if (ntohl (msg->mid.mid) < ntohl (ch->mid_recv.mid)) /* Old */
1452 /* Duplicate within the queue, drop */
1453 LOG (GNUNET_ERROR_TYPE_DEBUG,
1454 "Message on %s (mid %u) dropped, old.\n",
1456 ntohl (msg->mid.mid));
1457 GNUNET_MQ_discard (env);
1461 /* Channel is unreliable, so we do not ACK. But we also cannot
1462 allow buffering everything, so check if we have space... */
1463 if (ccc->num_recv >= ch->max_pending_messages)
1465 struct CadetOutOfOrderMessage *drop;
1467 /* Yep, need to drop. Drop the oldest message in
1469 LOG (GNUNET_ERROR_TYPE_DEBUG,
1470 "Queue full due slow client on %s, dropping oldest message\n",
1472 GNUNET_STATISTICS_update (stats,
1473 "# messages dropped due to slow client",
1476 drop = ccc->head_recv;
1477 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1481 GNUNET_MQ_discard (drop->env);
1486 /* Insert message into sorted out-of-order queue */
1487 com = GNUNET_new (struct CadetOutOfOrderMessage);
1488 com->mid = msg->mid;
1490 duplicate = GNUNET_NO;
1491 GNUNET_CONTAINER_DLL_insert_sorted (struct CadetOutOfOrderMessage,
1498 if (GNUNET_YES == duplicate)
1500 /* Duplicate within the queue, drop also (this is not covered by
1501 the case above if "delta" >= 64, which could be the case if
1502 max_pending_messages is also >= 64 or if our client is unready
1503 and we are seeing retransmissions of the message our client is
1505 LOG (GNUNET_ERROR_TYPE_DEBUG,
1506 "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1507 (unsigned int) payload_size,
1509 ntohl (msg->mid.mid));
1510 GNUNET_STATISTICS_update (stats,
1514 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1518 GNUNET_MQ_discard (com->env);
1520 send_channel_data_ack (ch);
1523 LOG (GNUNET_ERROR_TYPE_DEBUG,
1524 "Queued %s payload of %u bytes on %s-%X(%p) (mid %u, need %u first)\n",
1525 (GNUNET_YES == ccc->client_ready)
1527 : "client-not-ready",
1528 (unsigned int) payload_size,
1530 ntohl (ccc->ccn.channel_of_client),
1532 ntohl (msg->mid.mid),
1533 ntohl (ch->mid_recv.mid));
1534 /* NOTE: this ACK we _could_ skip, as the packet is out-of-order and
1535 the sender may already be transmitting the previous one. Needs
1536 experimental evaluation to see if/when this ACK helps or
1537 hurts. (We might even want another option.) */
1538 send_channel_data_ack (ch);
1543 * Function called once the tunnel has sent one of our messages.
1544 * If the message is unreliable, simply frees the `crm`. If the
1545 * message was reliable, calculate retransmission time and
1546 * wait for ACK (or retransmit).
1548 * @param cls the `struct CadetReliableMessage` that was sent
1549 * @param cid identifier of the connection within the tunnel, NULL
1550 * if transmission failed
1553 data_sent_cb (void *cls,
1554 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid);
1558 * We need to retry a transmission, the last one took too long to
1561 * @param cls the `struct CadetChannel` where we need to retransmit
1564 retry_transmission (void *cls)
1566 struct CadetChannel *ch = cls;
1567 struct CadetReliableMessage *crm = ch->head_sent;
1569 ch->retry_data_task = NULL;
1570 GNUNET_assert (NULL == crm->qe);
1571 LOG (GNUNET_ERROR_TYPE_DEBUG,
1572 "Retrying transmission on %s of message %u\n",
1574 (unsigned int) ntohl (crm->data_message->mid.mid));
1575 crm->qe = GCT_send (ch->t,
1576 &crm->data_message->header,
1579 GNUNET_assert (NULL == ch->retry_data_task);
1584 * We got an PLAINTEXT_DATA_ACK for a message in our queue, remove it from
1585 * the queue and tell our client that it can send more.
1587 * @param ch the channel that got the PLAINTEXT_DATA_ACK
1588 * @param cti identifier of the connection that delivered the message
1589 * @param crm the message that got acknowledged
1592 handle_matching_ack (struct CadetChannel *ch,
1593 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1594 struct CadetReliableMessage *crm)
1596 GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1599 ch->pending_messages--;
1600 GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
1601 LOG (GNUNET_ERROR_TYPE_DEBUG,
1602 "Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
1604 (unsigned int) ntohl (crm->data_message->mid.mid),
1605 ch->pending_messages);
1606 if (NULL != crm->qe)
1608 GCT_send_cancel (crm->qe);
1611 if ( (1 == crm->num_transmissions) &&
1614 GCC_ack_observed (cti);
1615 if (0 == memcmp (cti,
1616 &crm->connection_taken,
1617 sizeof (struct GNUNET_CADET_ConnectionTunnelIdentifier)))
1619 GCC_latency_observed (cti,
1620 GNUNET_TIME_absolute_get_duration (crm->first_transmission_time));
1623 GNUNET_free (crm->data_message);
1625 send_ack_to_client (ch,
1633 * We got an acknowledgement for payload data for a channel.
1634 * Possibly resume transmissions.
1636 * @param ch channel that got the ack
1637 * @param cti identifier of the connection that delivered the message
1638 * @param ack details about what was received
1641 GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
1642 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1643 const struct GNUNET_CADET_ChannelDataAckMessage *ack)
1645 struct CadetReliableMessage *crm;
1646 struct CadetReliableMessage *crmn;
1652 GNUNET_break (GNUNET_NO == ch->is_loopback);
1653 if (GNUNET_NO == ch->reliable)
1655 /* not expecting ACKs on unreliable channel, odd */
1656 GNUNET_break_op (0);
1659 /* mid_base is the MID of the next message that the
1660 other peer expects (i.e. that is missing!), everything
1661 LOWER (but excluding mid_base itself) was received. */
1662 mid_base = ntohl (ack->mid.mid);
1663 mid_mask = GNUNET_htonll (ack->futures);
1665 for (crm = ch->head_sent;
1670 delta = (unsigned int) (ntohl (crm->data_message->mid.mid) - mid_base);
1671 if (delta >= UINT_MAX - ch->max_pending_messages)
1673 /* overflow, means crm was a bit in the past, so this ACK counts for it. */
1674 LOG (GNUNET_ERROR_TYPE_DEBUG,
1675 "Got DATA_ACK with base %u satisfying past message %u on %s\n",
1676 (unsigned int) mid_base,
1677 ntohl (crm->data_message->mid.mid),
1679 handle_matching_ack (ch,
1688 LOG (GNUNET_ERROR_TYPE_DEBUG,
1689 "Testing bit %llX for mid %u (base: %u)\n",
1691 ntohl (crm->data_message->mid.mid),
1693 if (0 != (mid_mask & (1LLU << delta)))
1695 LOG (GNUNET_ERROR_TYPE_DEBUG,
1696 "Got DATA_ACK with mask for %u on %s\n",
1697 ntohl (crm->data_message->mid.mid),
1699 handle_matching_ack (ch,
1705 if (GNUNET_NO == found)
1707 /* ACK for message we already dropped, might have been a
1708 duplicate ACK? Ignore. */
1709 LOG (GNUNET_ERROR_TYPE_DEBUG,
1710 "Duplicate DATA_ACK on %s, ignoring\n",
1712 GNUNET_STATISTICS_update (stats,
1713 "# duplicate DATA_ACKs",
1718 if (NULL != ch->retry_data_task)
1720 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1721 ch->retry_data_task = NULL;
1723 if ( (NULL != ch->head_sent) &&
1724 (NULL == ch->head_sent->qe) )
1726 = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1727 &retry_transmission,
1733 * Destroy channel, based on the other peer closing the
1734 * connection. Also needs to remove this channel from
1737 * @param ch channel to destroy
1738 * @param cti identifier of the connection that delivered the message,
1739 * NULL if we are simulating receiving a destroy due to shutdown
1742 GCCH_handle_remote_destroy (struct CadetChannel *ch,
1743 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
1745 struct CadetChannelClient *ccc;
1747 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1748 LOG (GNUNET_ERROR_TYPE_DEBUG,
1749 "Received remote channel DESTROY for %s\n",
1751 if (GNUNET_YES == ch->destroy)
1753 /* Local client already gone, this is instant-death. */
1754 channel_destroy (ch);
1757 ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1758 if ( (NULL != ccc) &&
1759 (NULL != ccc->head_recv) )
1761 LOG (GNUNET_ERROR_TYPE_WARNING,
1762 "Lost end of transmission due to remote shutdown on %s\n",
1764 /* FIXME: change API to notify client about truncated transmission! */
1766 ch->destroy = GNUNET_YES;
1768 GSC_handle_remote_channel_destroy (ccc->c,
1771 channel_destroy (ch);
1776 * Test if element @a e1 comes before element @a e2.
1778 * @param cls closure, to a flag where we indicate duplicate packets
1779 * @param crm1 an element of to sort
1780 * @param crm2 another element to sort
1781 * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1784 cmp_crm_by_next_retry (void *cls,
1785 struct CadetReliableMessage *crm1,
1786 struct CadetReliableMessage *crm2)
1788 if (crm1->next_retry.abs_value_us <
1789 crm2->next_retry.abs_value_us)
1796 * Function called once the tunnel has sent one of our messages.
1797 * If the message is unreliable, simply frees the `crm`. If the
1798 * message was reliable, calculate retransmission time and
1799 * wait for ACK (or retransmit).
1801 * @param cls the `struct CadetReliableMessage` that was sent
1802 * @param cid identifier of the connection within the tunnel, NULL
1803 * if transmission failed
1806 data_sent_cb (void *cls,
1807 const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
1809 struct CadetReliableMessage *crm = cls;
1810 struct CadetChannel *ch = crm->ch;
1812 GNUNET_assert (GNUNET_NO == ch->is_loopback);
1813 GNUNET_assert (NULL != crm->qe);
1815 GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1818 if (GNUNET_NO == ch->reliable)
1820 GNUNET_free (crm->data_message);
1822 ch->pending_messages--;
1823 send_ack_to_client (ch,
1831 /* There was an error sending. */
1832 crm->num_transmissions = GNUNET_SYSERR;
1834 else if (GNUNET_SYSERR != crm->num_transmissions)
1836 /* Increment transmission counter, and possibly store @a cid
1837 if this was the first transmission. */
1838 crm->num_transmissions++;
1839 if (1 == crm->num_transmissions)
1841 crm->first_transmission_time = GNUNET_TIME_absolute_get ();
1842 crm->connection_taken = *cid;
1843 GCC_ack_expected (cid);
1846 if ( (0 == crm->retry_delay.rel_value_us) &&
1849 struct CadetConnection *cc = GCC_lookup (cid);
1852 crm->retry_delay = GCC_get_metrics (cc)->aged_latency;
1854 crm->retry_delay = ch->retry_time;
1856 crm->retry_delay = GNUNET_TIME_STD_BACKOFF (crm->retry_delay);
1857 crm->retry_delay = GNUNET_TIME_relative_max (crm->retry_delay,
1859 crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
1861 GNUNET_CONTAINER_DLL_insert_sorted (struct CadetReliableMessage,
1862 cmp_crm_by_next_retry,
1867 LOG (GNUNET_ERROR_TYPE_DEBUG,
1868 "Message %u sent, next transmission on %s in %s\n",
1869 (unsigned int) ntohl (crm->data_message->mid.mid),
1871 GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (ch->head_sent->next_retry),
1873 if (NULL == ch->head_sent->qe)
1875 if (NULL != ch->retry_data_task)
1876 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1878 = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1879 &retry_transmission,
1886 * Handle data given by a client.
1888 * Check whether the client is allowed to send in this tunnel, save if
1889 * channel is reliable and send an ACK to the client if there is still
1890 * buffer space in the tunnel.
1892 * @param ch Channel.
1893 * @param sender_ccn ccn of the sender
1894 * @param buf payload to transmit.
1895 * @param buf_len number of bytes in @a buf
1896 * @return #GNUNET_OK if everything goes well,
1897 * #GNUNET_SYSERR in case of an error.
1900 GCCH_handle_local_data (struct CadetChannel *ch,
1901 struct GNUNET_CADET_ClientChannelNumber sender_ccn,
1905 struct CadetReliableMessage *crm;
1907 if (ch->pending_messages >= ch->max_pending_messages)
1909 GNUNET_break (0); /* Fails: #5370 */
1910 return GNUNET_SYSERR;
1912 if (GNUNET_YES == ch->destroy)
1914 /* we are going down, drop messages */
1917 ch->pending_messages++;
1919 if (GNUNET_YES == ch->is_loopback)
1921 struct CadetChannelClient *receiver;
1922 struct GNUNET_MQ_Envelope *env;
1923 struct GNUNET_CADET_LocalData *ld;
1926 env = GNUNET_MQ_msg_extra (ld,
1928 GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1929 if ( (NULL != ch->owner) &&
1930 (sender_ccn.channel_of_client ==
1931 ch->owner->ccn.channel_of_client) )
1933 receiver = ch->dest;
1934 ack_to_owner = GNUNET_YES;
1936 else if ( (NULL != ch->dest) &&
1937 (sender_ccn.channel_of_client ==
1938 ch->dest->ccn.channel_of_client) )
1940 receiver = ch->owner;
1941 ack_to_owner = GNUNET_NO;
1946 return GNUNET_SYSERR;
1948 GNUNET_assert (NULL != receiver);
1949 ld->ccn = receiver->ccn;
1950 GNUNET_memcpy (&ld[1],
1953 if (GNUNET_YES == receiver->client_ready)
1955 ch->pending_messages--;
1956 GSC_send_to_client (receiver->c,
1958 send_ack_to_client (ch,
1963 struct CadetOutOfOrderMessage *oom;
1965 oom = GNUNET_new (struct CadetOutOfOrderMessage);
1967 GNUNET_CONTAINER_DLL_insert_tail (receiver->head_recv,
1968 receiver->tail_recv,
1970 receiver->num_recv++;
1975 /* Everything is correct, send the message. */
1976 crm = GNUNET_malloc (sizeof (*crm));
1978 crm->data_message = GNUNET_malloc (sizeof (struct GNUNET_CADET_ChannelAppDataMessage)
1980 crm->data_message->header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1981 crm->data_message->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
1982 ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
1983 crm->data_message->mid = ch->mid_send;
1984 crm->data_message->ctn = ch->ctn;
1985 GNUNET_memcpy (&crm->data_message[1],
1988 GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent,
1991 LOG (GNUNET_ERROR_TYPE_DEBUG,
1992 "Sending message %u from local client to %s with %u bytes\n",
1993 ntohl (crm->data_message->mid.mid),
1996 if (NULL != ch->retry_data_task)
1998 GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1999 ch->retry_data_task = NULL;
2001 crm->qe = GCT_send (ch->t,
2002 &crm->data_message->header,
2005 GNUNET_assert (NULL == ch->retry_data_task);
2011 * Handle ACK from client on local channel. Means the client is ready
2012 * for more data, see if we have any for it.
2014 * @param ch channel to destroy
2015 * @param client_ccn ccn of the client sending the ack
2018 GCCH_handle_local_ack (struct CadetChannel *ch,
2019 struct GNUNET_CADET_ClientChannelNumber client_ccn)
2021 struct CadetChannelClient *ccc;
2022 struct CadetOutOfOrderMessage *com;
2024 if ( (NULL != ch->owner) &&
2025 (ch->owner->ccn.channel_of_client == client_ccn.channel_of_client) )
2027 else if ( (NULL != ch->dest) &&
2028 (ch->dest->ccn.channel_of_client == client_ccn.channel_of_client) )
2032 ccc->client_ready = GNUNET_YES;
2033 com = ccc->head_recv;
2036 LOG (GNUNET_ERROR_TYPE_DEBUG,
2037 "Got LOCAL_ACK, %s-%X ready to receive more data, but none pending on %s-%X(%p)!\n",
2039 ntohl (client_ccn.channel_of_client),
2041 ntohl (ccc->ccn.channel_of_client),
2043 return; /* none pending */
2045 if (GNUNET_YES == ch->is_loopback)
2049 /* Messages are always in-order, just send */
2050 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
2054 GSC_send_to_client (ccc->c,
2056 /* Notify sender that we can receive more */
2057 if ( (NULL != ch->owner) &&
2058 (ccc->ccn.channel_of_client ==
2059 ch->owner->ccn.channel_of_client) )
2061 to_owner = GNUNET_NO;
2065 GNUNET_assert ( (NULL != ch->dest) &&
2066 (ccc->ccn.channel_of_client ==
2067 ch->dest->ccn.channel_of_client) );
2068 to_owner = GNUNET_YES;
2070 send_ack_to_client (ch,
2076 if ( (com->mid.mid != ch->mid_recv.mid) &&
2077 (GNUNET_NO == ch->out_of_order) &&
2078 (GNUNET_YES == ch->reliable) )
2080 LOG (GNUNET_ERROR_TYPE_DEBUG,
2081 "Got LOCAL_ACK, %s-%X ready to receive more data (but next one is out-of-order %u vs. %u)!\n",
2083 ntohl (ccc->ccn.channel_of_client),
2084 ntohl (com->mid.mid),
2085 ntohl (ch->mid_recv.mid));
2086 return; /* missing next one in-order */
2089 LOG (GNUNET_ERROR_TYPE_DEBUG,
2090 "Got LOCAL_ACK, giving payload message %u to %s-%X on %s\n",
2091 ntohl (com->mid.mid),
2093 ntohl (ccc->ccn.channel_of_client),
2096 /* all good, pass next message to client */
2097 GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
2101 /* FIXME: if unreliable, this is not aggressive
2102 enough, as it would be OK to have lost some! */
2104 ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
2105 ch->mid_futures >>= 1; /* equivalent to division by 2 */
2106 ccc->client_ready = GNUNET_NO;
2107 GSC_send_to_client (ccc->c,
2110 send_channel_data_ack (ch);
2111 if (NULL != ccc->head_recv)
2113 if (GNUNET_NO == ch->destroy)
2115 GCT_send_channel_destroy (ch->t,
2117 channel_destroy (ch);
2121 #define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-chn",__VA_ARGS__)
2127 * @param ch Channel.
2128 * @param level Debug level to use.
2131 GCCH_debug (struct CadetChannel *ch,
2132 enum GNUNET_ErrorType level)
2134 #if !defined(GNUNET_CULL_LOGGING)
2137 do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
2139 __FILE__, __FUNCTION__, __LINE__);
2145 LOG2 (level, "CHN *** DEBUG NULL CHANNEL ***\n");
2153 if (NULL != ch->owner)
2156 "CHN origin %s ready %s local-id: %u\n",
2157 GSC_2s (ch->owner->c),
2158 ch->owner->client_ready ? "YES" : "NO",
2159 ntohl (ch->owner->ccn.channel_of_client));
2161 if (NULL != ch->dest)
2164 "CHN destination %s ready %s local-id: %u\n",
2165 GSC_2s (ch->dest->c),
2166 ch->dest->client_ready ? "YES" : "NO",
2167 ntohl (ch->dest->ccn.channel_of_client));
2170 "CHN Message IDs recv: %d (%LLX), send: %d\n",
2171 ntohl (ch->mid_recv.mid),
2172 (unsigned long long) ch->mid_futures,
2173 ntohl (ch->mid_send.mid));
2179 /* end of gnunet-service-cadet-new_channel.c */