fix loopback disconnect signalling and a few memory leaks
[oweals/gnunet.git] / src / cadet / gnunet-service-cadet-new_channel.c
1 /*
2      This file is part of GNUnet.
3      Copyright (C) 2001-2017 GNUnet e.V.
4
5      GNUnet is free software; you can redistribute it and/or modify
6      it under the terms of the GNU General Public License as published
7      by the Free Software Foundation; either version 3, or (at your
8      option) any later version.
9
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      General Public License for more details.
14
15      You should have received a copy of the GNU General Public License
16      along with GNUnet; see the file COPYING.  If not, write to the
17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
18      Boston, MA 02110-1301, USA.
19 */
20 /**
21  * @file cadet/gnunet-service-cadet-new_channel.c
22  * @brief logical links between CADET clients
23  * @author Bartlomiej Polot
24  * @author Christian Grothoff
25  *
26  * TODO:
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?
35  */
36 #include "platform.h"
37 #include "gnunet_util_lib.h"
38 #include "cadet.h"
39 #include "gnunet_statistics_service.h"
40 #include "gnunet-service-cadet-new.h"
41 #include "gnunet-service-cadet-new_channel.h"
42 #include "gnunet-service-cadet-new_connection.h"
43 #include "gnunet-service-cadet-new_tunnels.h"
44 #include "gnunet-service-cadet-new_peer.h"
45 #include "gnunet-service-cadet-new_paths.h"
46
47 #define LOG(level,...) GNUNET_log_from (level,"cadet-chn",__VA_ARGS__)
48
49 /**
50  * How long do we initially wait before retransmitting?
51  */
52 #define CADET_INITIAL_RETRANSMIT_TIME GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, 250)
53
54 /**
55  * How long do we wait before dropping state about incoming
56  * connection to closed port?
57  */
58 #define TIMEOUT_CLOSED_PORT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 30)
59
60 /**
61  * How long do we wait at least before retransmitting ever?
62  */
63 #define MIN_RTT_DELAY GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, 75)
64
65 /**
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.
70  *
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...
74  */
75 #define MAX_OUT_OF_ORDER_DISTANCE 1024
76
77
78 /**
79  * All the states a channel can be in.
80  */
81 enum CadetChannelState
82 {
83   /**
84    * Uninitialized status, should never appear in operation.
85    */
86   CADET_CHANNEL_NEW,
87
88   /**
89    * Channel is to a port that is not open, we're waiting for the
90    * port to be opened.
91    */
92   CADET_CHANNEL_LOOSE,
93
94   /**
95    * CHANNEL_OPEN message sent, waiting for CHANNEL_OPEN_ACK.
96    */
97   CADET_CHANNEL_OPEN_SENT,
98
99   /**
100    * Connection confirmed, ready to carry traffic.
101    */
102   CADET_CHANNEL_READY
103 };
104
105
106 /**
107  * Info needed to retry a message in case it gets lost.
108  * Note that we DO use this structure also for unreliable
109  * messages.
110  */
111 struct CadetReliableMessage
112 {
113   /**
114    * Double linked list, FIFO style
115    */
116   struct CadetReliableMessage *next;
117
118   /**
119    * Double linked list, FIFO style
120    */
121   struct CadetReliableMessage *prev;
122
123   /**
124    * Which channel is this message in?
125    */
126   struct CadetChannel *ch;
127
128   /**
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
131    * ACK in time.
132    */
133   struct CadetTunnelQueueEntry *qe;
134
135   /**
136    * Data message we are trying to send.
137    */
138   struct GNUNET_CADET_ChannelAppDataMessage *data_message;
139
140   /**
141    * How soon should we retry if we fail to get an ACK?
142    * Messages in the queue are sorted by this value.
143    */
144   struct GNUNET_TIME_Absolute next_retry;
145
146   /**
147    * How long do we wait for an ACK after transmission?
148    * Use for the back-off calculation.
149    */
150   struct GNUNET_TIME_Relative retry_delay;
151
152   /**
153    * Time when we first successfully transmitted the message
154    * (that is, set @e num_transmissions to 1).
155    */
156   struct GNUNET_TIME_Absolute first_transmission_time;
157
158   /**
159    * Identifier of the connection that this message took when it
160    * was first transmitted.  Only useful if @e num_transmissions is 1.
161    */
162   struct GNUNET_CADET_ConnectionTunnelIdentifier connection_taken;
163
164   /**
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.
168    */
169   int num_transmissions;
170
171 };
172
173
174 /**
175  * List of received out-of-order data messages.
176  */
177 struct CadetOutOfOrderMessage
178 {
179   /**
180    * Double linked list, FIFO style
181    */
182   struct CadetOutOfOrderMessage *next;
183
184   /**
185    * Double linked list, FIFO style
186    */
187   struct CadetOutOfOrderMessage *prev;
188
189   /**
190    * ID of the message (messages up to this point needed
191    * before we give this one to the client).
192    */
193   struct ChannelMessageIdentifier mid;
194
195   /**
196    * The envelope with the payload of the out-of-order message
197    */
198   struct GNUNET_MQ_Envelope *env;
199
200 };
201
202
203 /**
204  * Client endpoint of a `struct CadetChannel`.  A channel may be a
205  * loopback channel, in which case it has two of these endpoints.
206  * Note that flow control also is required in both directions.
207  */
208 struct CadetChannelClient
209 {
210   /**
211    * Client handle.  Not by itself sufficient to designate
212    * the client endpoint, as the same client handle may
213    * be used for both the owner and the destination, and
214    * we thus also need the channel ID to identify the client.
215    */
216   struct CadetClient *c;
217
218   /**
219    * Head of DLL of messages received out of order or while client was unready.
220    */
221   struct CadetOutOfOrderMessage *head_recv;
222
223   /**
224    * Tail DLL of messages received out of order or while client was unready.
225    */
226   struct CadetOutOfOrderMessage *tail_recv;
227
228   /**
229    * Local tunnel number for this client.
230    * (if owner >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI,
231    *  otherwise < #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
232    */
233   struct GNUNET_CADET_ClientChannelNumber ccn;
234
235   /**
236    * Number of entries currently in @a head_recv DLL.
237    */
238   unsigned int num_recv;
239
240   /**
241    * Can we send data to the client?
242    */
243   int client_ready;
244
245 };
246
247
248 /**
249  * Struct containing all information regarding a channel to a remote client.
250  */
251 struct CadetChannel
252 {
253   /**
254    * Tunnel this channel is in.
255    */
256   struct CadetTunnel *t;
257
258   /**
259    * Client owner of the tunnel, if any.
260    * (Used if this channel represends the initiating end of the tunnel.)
261    */
262   struct CadetChannelClient *owner;
263
264   /**
265    * Client destination of the tunnel, if any.
266    * (Used if this channel represents the listening end of the tunnel.)
267    */
268   struct CadetChannelClient *dest;
269
270   /**
271    * Last entry in the tunnel's queue relating to control messages
272    * (#GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN or
273    * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK).  Used to cancel
274    * transmission in case we receive updated information.
275    */
276   struct CadetTunnelQueueEntry *last_control_qe;
277
278   /**
279    * Head of DLL of messages sent and not yet ACK'd.
280    */
281   struct CadetReliableMessage *head_sent;
282
283   /**
284    * Tail of DLL of messages sent and not yet ACK'd.
285    */
286   struct CadetReliableMessage *tail_sent;
287
288   /**
289    * Task to resend/poll in case no ACK is received.
290    */
291   struct GNUNET_SCHEDULER_Task *retry_control_task;
292
293   /**
294    * Task to resend/poll in case no ACK is received.
295    */
296   struct GNUNET_SCHEDULER_Task *retry_data_task;
297
298   /**
299    * Last time the channel was used
300    */
301   struct GNUNET_TIME_Absolute timestamp;
302
303   /**
304    * Destination port of the channel.
305    */
306   struct GNUNET_HashCode port;
307
308   /**
309    * Counter for exponential backoff.
310    */
311   struct GNUNET_TIME_Relative retry_time;
312
313   /**
314    * Bitfield of already-received messages past @e mid_recv.
315    */
316   uint64_t mid_futures;
317
318   /**
319    * Next MID expected for incoming traffic.
320    */
321   struct ChannelMessageIdentifier mid_recv;
322
323   /**
324    * Next MID to use for outgoing traffic.
325    */
326   struct ChannelMessageIdentifier mid_send;
327
328   /**
329    * Total (reliable) messages pending ACK for this channel.
330    */
331   unsigned int pending_messages;
332
333   /**
334    * Maximum (reliable) messages pending ACK for this channel
335    * before we throttle the client.
336    */
337   unsigned int max_pending_messages;
338
339   /**
340    * Number identifying this channel in its tunnel.
341    */
342   struct GNUNET_CADET_ChannelTunnelNumber ctn;
343
344   /**
345    * Channel state.
346    */
347   enum CadetChannelState state;
348
349   /**
350    * Count how many ACKs we skipped, used to prevent long
351    * sequences of ACK skipping.
352    */
353   unsigned int skip_ack_series;
354
355   /**
356    * Is the tunnel bufferless (minimum latency)?
357    */
358   int nobuffer;
359
360   /**
361    * Is the tunnel reliable?
362    */
363   int reliable;
364
365   /**
366    * Is the tunnel out-of-order?
367    */
368   int out_of_order;
369
370   /**
371    * Is this channel a loopback channel, where the destination is us again?
372    */
373   int is_loopback;
374
375   /**
376    * Flag to signal the destruction of the channel.  If this is set to
377    * #GNUNET_YES the channel will be destroyed once the queue is
378    * empty.
379    */
380   int destroy;
381
382 };
383
384
385 /**
386  * Get the static string for identification of the channel.
387  *
388  * @param ch Channel.
389  *
390  * @return Static string with the channel IDs.
391  */
392 const char *
393 GCCH_2s (const struct CadetChannel *ch)
394 {
395   static char buf[128];
396
397   GNUNET_snprintf (buf,
398                    sizeof (buf),
399                    "Channel %s:%s ctn:%X(%X/%X)",
400                    (GNUNET_YES == ch->is_loopback)
401                    ? "loopback"
402                    : GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
403                    GNUNET_h2s (&ch->port),
404                    ch->ctn,
405                    (NULL == ch->owner) ? 0 : ntohl (ch->owner->ccn.channel_of_client),
406                    (NULL == ch->dest) ? 0 : ntohl (ch->dest->ccn.channel_of_client));
407   return buf;
408 }
409
410
411 /**
412  * Get the channel's public ID.
413  *
414  * @param ch Channel.
415  *
416  * @return ID used to identify the channel with the remote peer.
417  */
418 struct GNUNET_CADET_ChannelTunnelNumber
419 GCCH_get_id (const struct CadetChannel *ch)
420 {
421   return ch->ctn;
422 }
423
424
425 /**
426  * Release memory associated with @a ccc
427  *
428  * @param ccc data structure to clean up
429  */
430 static void
431 free_channel_client (struct CadetChannelClient *ccc)
432 {
433   struct CadetOutOfOrderMessage *com;
434
435   while (NULL != (com = ccc->head_recv))
436   {
437     GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
438                                  ccc->tail_recv,
439                                  com);
440     ccc->num_recv--;
441     GNUNET_MQ_discard (com->env);
442     GNUNET_free (com);
443   }
444   GNUNET_free (ccc);
445 }
446
447
448 /**
449  * Destroy the given channel.
450  *
451  * @param ch channel to destroy
452  */
453 static void
454 channel_destroy (struct CadetChannel *ch)
455 {
456   struct CadetReliableMessage *crm;
457
458   while (NULL != (crm = ch->head_sent))
459   {
460     GNUNET_assert (ch == crm->ch);
461     if (NULL != crm->qe)
462     {
463       GCT_send_cancel (crm->qe);
464       crm->qe = NULL;
465     }
466     GNUNET_CONTAINER_DLL_remove (ch->head_sent,
467                                  ch->tail_sent,
468                                  crm);
469     GNUNET_free (crm->data_message);
470     GNUNET_free (crm);
471   }
472   if (NULL != ch->owner)
473   {
474     free_channel_client (ch->owner);
475     ch->owner = NULL;
476   }
477   if (NULL != ch->dest)
478   {
479     free_channel_client (ch->dest);
480     ch->dest = NULL;
481   }
482   if (NULL != ch->last_control_qe)
483   {
484     GCT_send_cancel (ch->last_control_qe);
485     ch->last_control_qe = NULL;
486   }
487   if (NULL != ch->retry_data_task)
488   {
489     GNUNET_SCHEDULER_cancel (ch->retry_data_task);
490     ch->retry_data_task = NULL;
491   }
492   if (NULL != ch->retry_control_task)
493   {
494     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
495     ch->retry_control_task = NULL;
496   }
497   if (GNUNET_NO == ch->is_loopback)
498   {
499     GCT_remove_channel (ch->t,
500                         ch,
501                         ch->ctn);
502     ch->t = NULL;
503   }
504   GNUNET_free (ch);
505 }
506
507
508 /**
509  * Send a channel create message.
510  *
511  * @param cls Channel for which to send.
512  */
513 static void
514 send_channel_open (void *cls);
515
516
517 /**
518  * Function called once the tunnel confirms that we sent the
519  * create message.  Delays for a bit until we retry.
520  *
521  * @param cls our `struct CadetChannel`.
522  * @param cid identifier of the connection within the tunnel, NULL
523  *            if transmission failed
524  */
525 static void
526 channel_open_sent_cb (void *cls,
527                       const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
528 {
529   struct CadetChannel *ch = cls;
530
531   GNUNET_assert (NULL != ch->last_control_qe);
532   ch->last_control_qe = NULL;
533   ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
534   LOG (GNUNET_ERROR_TYPE_DEBUG,
535        "Sent CADET_CHANNEL_OPEN on %s, retrying in %s\n",
536        GCCH_2s (ch),
537        GNUNET_STRINGS_relative_time_to_string (ch->retry_time,
538                                                GNUNET_YES));
539   ch->retry_control_task
540     = GNUNET_SCHEDULER_add_delayed (ch->retry_time,
541                                     &send_channel_open,
542                                     ch);
543 }
544
545
546 /**
547  * Send a channel open message.
548  *
549  * @param cls Channel for which to send.
550  */
551 static void
552 send_channel_open (void *cls)
553 {
554   struct CadetChannel *ch = cls;
555   struct GNUNET_CADET_ChannelOpenMessage msgcc;
556   uint32_t options;
557
558   ch->retry_control_task = NULL;
559   LOG (GNUNET_ERROR_TYPE_DEBUG,
560        "Sending CHANNEL_OPEN message for %s\n",
561        GCCH_2s (ch));
562   options = 0;
563   if (ch->nobuffer)
564     options |= GNUNET_CADET_OPTION_NOBUFFER;
565   if (ch->reliable)
566     options |= GNUNET_CADET_OPTION_RELIABLE;
567   if (ch->out_of_order)
568     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
569   msgcc.header.size = htons (sizeof (msgcc));
570   msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN);
571   msgcc.opt = htonl (options);
572   msgcc.port = ch->port;
573   msgcc.ctn = ch->ctn;
574   ch->state = CADET_CHANNEL_OPEN_SENT;
575   if (NULL != ch->last_control_qe)
576     GCT_send_cancel (ch->last_control_qe);
577   ch->last_control_qe = GCT_send (ch->t,
578                                   &msgcc.header,
579                                   &channel_open_sent_cb,
580                                   ch);
581   GNUNET_assert (NULL == ch->retry_control_task);
582 }
583
584
585 /**
586  * Function called once and only once after a channel was bound
587  * to its tunnel via #GCT_add_channel() is ready for transmission.
588  * Note that this is only the case for channels that this peer
589  * initiates, as for incoming channels we assume that they are
590  * ready for transmission immediately upon receiving the open
591  * message.  Used to bootstrap the #GCT_send() process.
592  *
593  * @param ch the channel for which the tunnel is now ready
594  */
595 void
596 GCCH_tunnel_up (struct CadetChannel *ch)
597 {
598   GNUNET_assert (NULL == ch->retry_control_task);
599   LOG (GNUNET_ERROR_TYPE_DEBUG,
600        "Tunnel up, sending CHANNEL_OPEN on %s now\n",
601        GCCH_2s (ch));
602   ch->retry_control_task
603     = GNUNET_SCHEDULER_add_now (&send_channel_open,
604                                 ch);
605 }
606
607
608 /**
609  * Create a new channel.
610  *
611  * @param owner local client owning the channel
612  * @param ccn local number of this channel at the @a owner
613  * @param destination peer to which we should build the channel
614  * @param port desired port at @a destination
615  * @param options options for the channel
616  * @return handle to the new channel
617  */
618 struct CadetChannel *
619 GCCH_channel_local_new (struct CadetClient *owner,
620                         struct GNUNET_CADET_ClientChannelNumber ccn,
621                         struct CadetPeer *destination,
622                         const struct GNUNET_HashCode *port,
623                         uint32_t options)
624 {
625   struct CadetChannel *ch;
626   struct CadetChannelClient *ccco;
627
628   ccco = GNUNET_new (struct CadetChannelClient);
629   ccco->c = owner;
630   ccco->ccn = ccn;
631   ccco->client_ready = GNUNET_YES;
632
633   ch = GNUNET_new (struct CadetChannel);
634   ch->mid_recv.mid = htonl (1); /* The OPEN_ACK counts as message 0! */
635   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
636   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
637   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
638   ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
639   ch->owner = ccco;
640   ch->port = *port;
641   if (0 == memcmp (&my_full_id,
642                    GCP_get_id (destination),
643                    sizeof (struct GNUNET_PeerIdentity)))
644   {
645     struct CadetClient *c;
646
647     ch->is_loopback = GNUNET_YES;
648     c = GNUNET_CONTAINER_multihashmap_get (open_ports,
649                                            port);
650     if (NULL == c)
651     {
652       /* port closed, wait for it to possibly open */
653       ch->state = CADET_CHANNEL_LOOSE;
654       (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
655                                                 port,
656                                                 ch,
657                                                 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
658       LOG (GNUNET_ERROR_TYPE_DEBUG,
659            "Created loose incoming loopback channel to port %s\n",
660            GNUNET_h2s (&ch->port));
661     }
662     else
663     {
664       ch->dest = GNUNET_new (struct CadetChannelClient);
665       ch->dest->c = c;
666       ch->dest->client_ready = GNUNET_YES;
667       GCCH_bind (ch,
668                  ch->dest->c);
669     }
670   }
671   else
672   {
673     ch->t = GCP_get_tunnel (destination,
674                             GNUNET_YES);
675     ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
676     ch->ctn = GCT_add_channel (ch->t,
677                                ch);
678   }
679   GNUNET_STATISTICS_update (stats,
680                             "# channels",
681                             1,
682                             GNUNET_NO);
683   LOG (GNUNET_ERROR_TYPE_DEBUG,
684        "Created channel to port %s at peer %s for %s using %s\n",
685        GNUNET_h2s (port),
686        GCP_2s (destination),
687        GSC_2s (owner),
688        (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
689   return ch;
690 }
691
692
693 /**
694  * We had an incoming channel to a port that is closed.
695  * It has not been opened for a while, drop it.
696  *
697  * @param cls the channel to drop
698  */
699 static void
700 timeout_closed_cb (void *cls)
701 {
702   struct CadetChannel *ch = cls;
703
704   ch->retry_control_task = NULL;
705   LOG (GNUNET_ERROR_TYPE_DEBUG,
706        "Closing incoming channel to port %s from peer %s due to timeout\n",
707        GNUNET_h2s (&ch->port),
708        GCP_2s (GCT_get_destination (ch->t)));
709   channel_destroy (ch);
710 }
711
712
713 /**
714  * Create a new channel based on a request coming in over the network.
715  *
716  * @param t tunnel to the remote peer
717  * @param ctn identifier of this channel in the tunnel
718  * @param port desired local port
719  * @param options options for the channel
720  * @return handle to the new channel
721  */
722 struct CadetChannel *
723 GCCH_channel_incoming_new (struct CadetTunnel *t,
724                            struct GNUNET_CADET_ChannelTunnelNumber ctn,
725                            const struct GNUNET_HashCode *port,
726                            uint32_t options)
727 {
728   struct CadetChannel *ch;
729   struct CadetClient *c;
730
731   ch = GNUNET_new (struct CadetChannel);
732   ch->port = *port;
733   ch->t = t;
734   ch->ctn = ctn;
735   ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
736   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
737   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
738   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
739   ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
740   GNUNET_STATISTICS_update (stats,
741                             "# channels",
742                             1,
743                             GNUNET_NO);
744
745   c = GNUNET_CONTAINER_multihashmap_get (open_ports,
746                                          port);
747   if (NULL == c)
748   {
749     /* port closed, wait for it to possibly open */
750     ch->state = CADET_CHANNEL_LOOSE;
751     (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
752                                               port,
753                                               ch,
754                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
755     ch->retry_control_task
756       = GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
757                                       &timeout_closed_cb,
758                                       ch);
759     LOG (GNUNET_ERROR_TYPE_DEBUG,
760          "Created loose incoming channel to port %s from peer %s\n",
761          GNUNET_h2s (&ch->port),
762          GCP_2s (GCT_get_destination (ch->t)));
763   }
764   else
765   {
766     GCCH_bind (ch,
767                c);
768   }
769   GNUNET_STATISTICS_update (stats,
770                             "# channels",
771                             1,
772                             GNUNET_NO);
773   return ch;
774 }
775
776
777 /**
778  * Function called once the tunnel confirms that we sent the
779  * ACK message.  Just remembers it was sent, we do not expect
780  * ACKs for ACKs ;-).
781  *
782  * @param cls our `struct CadetChannel`.
783  * @param cid identifier of the connection within the tunnel, NULL
784  *            if transmission failed
785  */
786 static void
787 send_ack_cb (void *cls,
788              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
789 {
790   struct CadetChannel *ch = cls;
791
792   GNUNET_assert (NULL != ch->last_control_qe);
793   ch->last_control_qe = NULL;
794 }
795
796
797 /**
798  * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
799  *
800  * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
801  */
802 static void
803 send_channel_data_ack (struct CadetChannel *ch)
804 {
805   struct GNUNET_CADET_ChannelDataAckMessage msg;
806
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));
811   msg.ctn = ch->ctn;
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,
818        GCCH_2s (ch));
819   if (NULL != ch->last_control_qe)
820     GCT_send_cancel (ch->last_control_qe);
821   ch->last_control_qe = GCT_send (ch->t,
822                                   &msg.header,
823                                   &send_ack_cb,
824                                   ch);
825 }
826
827
828 /**
829  * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
830  * connection is up.
831  *
832  * @param cls the `struct CadetChannel`
833  */
834 static void
835 send_open_ack (void *cls)
836 {
837   struct CadetChannel *ch = cls;
838   struct GNUNET_CADET_ChannelManageMessage msg;
839
840   LOG (GNUNET_ERROR_TYPE_DEBUG,
841        "Sending CHANNEL_OPEN_ACK on %s\n",
842        GCCH_2s (ch));
843   ch->retry_control_task = NULL;
844   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
845   msg.header.size = htons (sizeof (msg));
846   msg.reserved = htonl (0);
847   msg.ctn = ch->ctn;
848   if (NULL != ch->last_control_qe)
849     GCT_send_cancel (ch->last_control_qe);
850   ch->last_control_qe = GCT_send (ch->t,
851                                   &msg.header,
852                                   &send_ack_cb,
853                                   ch);
854 }
855
856
857 /**
858  * We got a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN message again for
859  * this channel.  If the binding was successful, (re)transmit the
860  * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
861  *
862  * @param ch channel that got the duplicate open
863  * @param cti identifier of the connection that delivered the message
864  */
865 void
866 GCCH_handle_duplicate_open (struct CadetChannel *ch,
867                             const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
868 {
869   if (NULL == ch->dest)
870   {
871     LOG (GNUNET_ERROR_TYPE_DEBUG,
872          "Ignoring duplicate CHANNEL_OPEN on %s: port is closed\n",
873          GCCH_2s (ch));
874     return;
875   }
876   if (NULL != ch->retry_control_task)
877   {
878     LOG (GNUNET_ERROR_TYPE_DEBUG,
879          "Ignoring duplicate CHANNEL_OPEN on %s: control message is pending\n",
880          GCCH_2s (ch));
881     return;
882   }
883   LOG (GNUNET_ERROR_TYPE_DEBUG,
884        "Retransmitting CHANNEL_OPEN_ACK on %s\n",
885        GCCH_2s (ch));
886   ch->retry_control_task
887     = GNUNET_SCHEDULER_add_now (&send_open_ack,
888                                 ch);
889 }
890
891
892 /**
893  * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK to the client to solicit more messages.
894  *
895  * @param ch channel the ack is for
896  * @param to_owner #GNUNET_YES to send to owner,
897  *                 #GNUNET_NO to send to dest
898  */
899 static void
900 send_ack_to_client (struct CadetChannel *ch,
901                     int to_owner)
902 {
903   struct GNUNET_MQ_Envelope *env;
904   struct GNUNET_CADET_LocalAck *ack;
905   struct CadetChannelClient *ccc;
906
907   ccc = (GNUNET_YES == to_owner) ? ch->owner : ch->dest;
908   if (NULL == ccc)
909   {
910     /* This can happen if we are just getting ACKs after
911        our local client already disconnected. */
912     GNUNET_assert (GNUNET_YES == ch->destroy);
913     return;
914   }
915   env = GNUNET_MQ_msg (ack,
916                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
917   ack->ccn = ccc->ccn;
918   LOG (GNUNET_ERROR_TYPE_DEBUG,
919        "Sending CADET_LOCAL_ACK to %s (%s) at ccn %X (%u/%u pending)\n",
920        GSC_2s (ccc->c),
921        (GNUNET_YES == to_owner) ? "owner" : "dest",
922        ntohl (ack->ccn.channel_of_client),
923        ch->pending_messages,
924        ch->max_pending_messages);
925   GSC_send_to_client (ccc->c,
926                       env);
927 }
928
929
930 /**
931  * A client is bound to the port that we have a channel
932  * open to.  Send the acknowledgement for the connection
933  * request and establish the link with the client.
934  *
935  * @param ch open incoming channel
936  * @param c client listening on the respective port
937  */
938 void
939 GCCH_bind (struct CadetChannel *ch,
940            struct CadetClient *c)
941 {
942   uint32_t options;
943   struct CadetChannelClient *cccd;
944
945   LOG (GNUNET_ERROR_TYPE_DEBUG,
946        "Binding %s from %s to port %s of %s\n",
947        GCCH_2s (ch),
948        GCT_2s (ch->t),
949        GNUNET_h2s (&ch->port),
950        GSC_2s (c));
951   if (NULL != ch->retry_control_task)
952   {
953     /* there might be a timeout task here */
954     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
955     ch->retry_control_task = NULL;
956   }
957   options = 0;
958   if (ch->nobuffer)
959     options |= GNUNET_CADET_OPTION_NOBUFFER;
960   if (ch->reliable)
961     options |= GNUNET_CADET_OPTION_RELIABLE;
962   if (ch->out_of_order)
963     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
964   cccd = GNUNET_new (struct CadetChannelClient);
965   ch->dest = cccd;
966   cccd->c = c;
967   cccd->client_ready = GNUNET_YES;
968   cccd->ccn = GSC_bind (c,
969                         ch,
970                         (GNUNET_YES == ch->is_loopback)
971                         ? GCP_get (&my_full_id,
972                                    GNUNET_YES)
973                         : GCT_get_destination (ch->t),
974                         &ch->port,
975                         options);
976   GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
977                  GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
978   ch->mid_recv.mid = htonl (1); /* The OPEN counts as message 0! */
979   if (GNUNET_YES == ch->is_loopback)
980   {
981     ch->state = CADET_CHANNEL_OPEN_SENT;
982     GCCH_handle_channel_open_ack (ch,
983                                   NULL);
984   }
985   else
986   {
987     /* notify other peer that we accepted the connection */
988     ch->state = CADET_CHANNEL_READY;
989     ch->retry_control_task
990       = GNUNET_SCHEDULER_add_now (&send_open_ack,
991                                   ch);
992   }
993   /* give client it's initial supply of ACKs */
994   GNUNET_assert (ntohl (cccd->ccn.channel_of_client) <
995                  GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
996   for (unsigned int i=0;i<ch->max_pending_messages;i++)
997     send_ack_to_client (ch,
998                         GNUNET_NO);
999 }
1000
1001
1002 /**
1003  * Destroy locally created channel.  Called by the local client, so no
1004  * need to tell the client.
1005  *
1006  * @param ch channel to destroy
1007  * @param c client that caused the destruction
1008  * @param ccn client number of the client @a c
1009  */
1010 void
1011 GCCH_channel_local_destroy (struct CadetChannel *ch,
1012                             struct CadetClient *c,
1013                             struct GNUNET_CADET_ClientChannelNumber ccn)
1014 {
1015   LOG (GNUNET_ERROR_TYPE_DEBUG,
1016        "%s asks for destruction of %s\n",
1017        GSC_2s (c),
1018        GCCH_2s (ch));
1019   GNUNET_assert (NULL != c);
1020   if ( (NULL != ch->owner) &&
1021        (c == ch->owner->c) &&
1022        (ccn.channel_of_client == ch->owner->ccn.channel_of_client) )
1023   {
1024     free_channel_client (ch->owner);
1025     ch->owner = NULL;
1026   }
1027   else if ( (NULL != ch->dest) &&
1028             (c == ch->dest->c) &&
1029             (ccn.channel_of_client == ch->dest->ccn.channel_of_client) )
1030   {
1031     free_channel_client (ch->dest);
1032     ch->dest = NULL;
1033   }
1034   else
1035   {
1036     GNUNET_assert (0);
1037   }
1038
1039   if (GNUNET_YES == ch->destroy)
1040   {
1041     /* other end already destroyed, with the local client gone, no need
1042        to finish transmissions, just destroy immediately. */
1043     channel_destroy (ch);
1044     return;
1045   }
1046   if ( (NULL != ch->head_sent) &&
1047        ( (NULL != ch->owner) ||
1048          (NULL != ch->dest) ) )
1049   {
1050     /* Wait for other end to destroy us as well,
1051        and otherwise allow send queue to be transmitted first */
1052     ch->destroy = GNUNET_YES;
1053     return;
1054   }
1055   if (GNUNET_YES == ch->is_loopback)
1056   {
1057     struct CadetChannelClient *ccc;
1058
1059     /* Find which end is left... */
1060     ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1061     GSC_handle_remote_channel_destroy (ccc->c,
1062                                        ccc->ccn,
1063                                        ch);
1064     channel_destroy (ch);
1065     return;
1066   }
1067   /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
1068   switch (ch->state)
1069   {
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. */
1073     break;
1074   case CADET_CHANNEL_LOOSE:
1075     GSC_drop_loose_channel (&ch->port,
1076                             ch);
1077     break;
1078   default:
1079     GCT_send_channel_destroy (ch->t,
1080                               ch->ctn);
1081   }
1082   /* Nothing left to do, just finish destruction */
1083   channel_destroy (ch);
1084 }
1085
1086
1087 /**
1088  * We got an acknowledgement for the creation of the channel
1089  * (the port is open on the other side). Begin transmissions.
1090  *
1091  * @param ch channel to destroy
1092  * @param cti identifier of the connection that delivered the message
1093  */
1094 void
1095 GCCH_handle_channel_open_ack (struct CadetChannel *ch,
1096                               const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
1097 {
1098   switch (ch->state)
1099   {
1100   case CADET_CHANNEL_NEW:
1101     /* this should be impossible */
1102     GNUNET_break (0);
1103     break;
1104   case CADET_CHANNEL_LOOSE:
1105     /* This makes no sense. */
1106     GNUNET_break_op (0);
1107     break;
1108   case CADET_CHANNEL_OPEN_SENT:
1109     if (NULL == ch->owner)
1110     {
1111       /* We're not the owner, wrong direction! */
1112       GNUNET_break_op (0);
1113       return;
1114     }
1115     LOG (GNUNET_ERROR_TYPE_DEBUG,
1116          "Received CHANNEL_OPEN_ACK for waiting %s, entering READY state\n",
1117          GCCH_2s (ch));
1118     if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
1119     {
1120       GNUNET_SCHEDULER_cancel (ch->retry_control_task);
1121       ch->retry_control_task = NULL;
1122     }
1123     ch->state = CADET_CHANNEL_READY;
1124     /* On first connect, send client as many ACKs as we allow messages
1125        to be buffered! */
1126     for (unsigned int i=0;i<ch->max_pending_messages;i++)
1127       send_ack_to_client (ch,
1128                           GNUNET_YES);
1129     break;
1130   case CADET_CHANNEL_READY:
1131     /* duplicate ACK, maybe we retried the CREATE. Ignore. */
1132     LOG (GNUNET_ERROR_TYPE_DEBUG,
1133          "Received duplicate channel OPEN_ACK for %s\n",
1134          GCCH_2s (ch));
1135     GNUNET_STATISTICS_update (stats,
1136                               "# duplicate CREATE_ACKs",
1137                               1,
1138                               GNUNET_NO);
1139     break;
1140   }
1141 }
1142
1143
1144 /**
1145  * Test if element @a e1 comes before element @a e2.
1146  *
1147  * @param cls closure, to a flag where we indicate duplicate packets
1148  * @param m1 a message of to sort
1149  * @param m2 another message to sort
1150  * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1151  */
1152 static int
1153 is_before (void *cls,
1154            struct CadetOutOfOrderMessage *m1,
1155            struct CadetOutOfOrderMessage *m2)
1156 {
1157   int *duplicate = cls;
1158   uint32_t v1 = ntohl (m1->mid.mid);
1159   uint32_t v2 = ntohl (m2->mid.mid);
1160   uint32_t delta;
1161
1162   delta = v2 - v1;
1163   if (0 == delta)
1164     *duplicate = GNUNET_YES;
1165   if (delta > (uint32_t) INT_MAX)
1166   {
1167     /* in overflow range, we can safely assume we wrapped around */
1168     return GNUNET_NO;
1169   }
1170   else
1171   {
1172     /* result is small, thus v2 > v1, thus m1 < m2 */
1173     return GNUNET_YES;
1174   }
1175 }
1176
1177
1178 /**
1179  * We got payload data for a channel.  Pass it on to the client
1180  * and send an ACK to the other end (once flow control allows it!)
1181  *
1182  * @param ch channel that got data
1183  * @param cti identifier of the connection that delivered the message
1184  * @param msg message that was received
1185  */
1186 void
1187 GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
1188                                     const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1189                                     const struct GNUNET_CADET_ChannelAppDataMessage *msg)
1190 {
1191   struct GNUNET_MQ_Envelope *env;
1192   struct GNUNET_CADET_LocalData *ld;
1193   struct CadetChannelClient *ccc;
1194   size_t payload_size;
1195   struct CadetOutOfOrderMessage *com;
1196   int duplicate;
1197   uint32_t mid_min;
1198   uint32_t mid_max;
1199   uint32_t mid_msg;
1200   uint32_t delta;
1201
1202   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1203   if ( (GNUNET_YES == ch->destroy) &&
1204        (NULL == ch->owner) &&
1205        (NULL == ch->dest) )
1206   {
1207     /* This client is gone, but we still have messages to send to
1208        the other end (which is why @a ch is not yet dead).  However,
1209        we cannot pass messages to our client anymore. */
1210     LOG (GNUNET_ERROR_TYPE_DEBUG,
1211          "Dropping incoming payload on %s as this end is already closed\n",
1212          GCCH_2s (ch));
1213     /* send back DESTROY notification to stop further retransmissions! */
1214     GCT_send_channel_destroy (ch->t,
1215                               ch->ctn);
1216     return;
1217   }
1218   payload_size = ntohs (msg->header.size) - sizeof (*msg);
1219   env = GNUNET_MQ_msg_extra (ld,
1220                              payload_size,
1221                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1222   ld->ccn = (NULL == ch->dest) ? ch->owner->ccn : ch->dest->ccn;
1223   GNUNET_memcpy (&ld[1],
1224                  &msg[1],
1225                  payload_size);
1226   ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1227   if ( (GNUNET_YES == ccc->client_ready) &&
1228        ( (GNUNET_YES == ch->out_of_order) ||
1229          (msg->mid.mid == ch->mid_recv.mid) ) )
1230   {
1231     LOG (GNUNET_ERROR_TYPE_DEBUG,
1232          "Giving %u bytes of payload with MID %u from %s to client %s\n",
1233          (unsigned int) payload_size,
1234          ntohl (msg->mid.mid),
1235          GCCH_2s (ch),
1236          GSC_2s (ccc->c));
1237     ccc->client_ready = GNUNET_NO;
1238     GSC_send_to_client (ccc->c,
1239                         env);
1240     ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
1241     ch->mid_futures >>= 1;
1242     send_channel_data_ack (ch);
1243     return;
1244   }
1245
1246   if (GNUNET_YES == ch->reliable)
1247   {
1248     /* check if message ought to be dropped because it is ancient/too distant/duplicate */
1249     mid_min = ntohl (ch->mid_recv.mid);
1250     mid_max = mid_min + ch->max_pending_messages;
1251     mid_msg = ntohl (msg->mid.mid);
1252     if ( ( (uint32_t) (mid_msg - mid_min) > ch->max_pending_messages) ||
1253          ( (uint32_t) (mid_max - mid_msg) > ch->max_pending_messages) )
1254     {
1255       LOG (GNUNET_ERROR_TYPE_DEBUG,
1256            "%s at %u drops ancient or far-future message %u\n",
1257            GCCH_2s (ch),
1258            (unsigned int) mid_min,
1259            ntohl (msg->mid.mid));
1260
1261       GNUNET_STATISTICS_update (stats,
1262                                 "# duplicate DATA (ancient or future)",
1263                                 1,
1264                                 GNUNET_NO);
1265       GNUNET_MQ_discard (env);
1266       send_channel_data_ack (ch);
1267       return;
1268     }
1269     /* mark bit for future ACKs */
1270     delta = mid_msg - mid_min - 1; /* overflow/underflow are OK here */
1271     if (delta < 64)
1272     {
1273       if (0 != (ch->mid_futures & (1LLU << delta)))
1274       {
1275         /* Duplicate within the queue, drop also */
1276         LOG (GNUNET_ERROR_TYPE_DEBUG,
1277              "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1278              (unsigned int) payload_size,
1279              GCCH_2s (ch),
1280              ntohl (msg->mid.mid));
1281         GNUNET_STATISTICS_update (stats,
1282                                   "# duplicate DATA",
1283                                   1,
1284                                   GNUNET_NO);
1285         GNUNET_MQ_discard (env);
1286         send_channel_data_ack (ch);
1287         return;
1288       }
1289       ch->mid_futures |= (1LLU << delta);
1290       LOG (GNUNET_ERROR_TYPE_DEBUG,
1291            "Marked bit %llX for mid %u (base: %u); now: %llX\n",
1292            (1LLU << delta),
1293            mid_msg,
1294            mid_min,
1295            ch->mid_futures);
1296     }
1297   }
1298   else /* ! ch->reliable */
1299   {
1300     /* Channel is unreliable, so we do not ACK. But we also cannot
1301        allow buffering everything, so check if we have space... */
1302     if (ccc->num_recv >= ch->max_pending_messages)
1303     {
1304       struct CadetOutOfOrderMessage *drop;
1305
1306       /* Yep, need to drop. Drop the oldest message in
1307          the buffer. */
1308       LOG (GNUNET_ERROR_TYPE_DEBUG,
1309            "Queue full due slow client on %s, dropping oldest message\n",
1310            GCCH_2s (ch));
1311       GNUNET_STATISTICS_update (stats,
1312                                 "# messages dropped due to slow client",
1313                                 1,
1314                                 GNUNET_NO);
1315       drop = ccc->head_recv;
1316       GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1317                                    ccc->tail_recv,
1318                                    drop);
1319       ccc->num_recv--;
1320       GNUNET_MQ_discard (drop->env);
1321       GNUNET_free (drop);
1322     }
1323   }
1324
1325   /* Insert message into sorted out-of-order queue */
1326   com = GNUNET_new (struct CadetOutOfOrderMessage);
1327   com->mid = msg->mid;
1328   com->env = env;
1329   duplicate = GNUNET_NO;
1330   GNUNET_CONTAINER_DLL_insert_sorted (struct CadetOutOfOrderMessage,
1331                                       is_before,
1332                                       &duplicate,
1333                                       ccc->head_recv,
1334                                       ccc->tail_recv,
1335                                       com);
1336   ccc->num_recv++;
1337   if (GNUNET_YES == duplicate)
1338   {
1339     /* Duplicate within the queue, drop also (this is not covered by
1340        the case above if "delta" >= 64, which could be the case if
1341        max_pending_messages is also >= 64 or if our client is unready
1342        and we are seeing retransmissions of the message our client is
1343        blocked on. */
1344     LOG (GNUNET_ERROR_TYPE_DEBUG,
1345          "Duplicate payload of %u bytes on %s (mid %u) dropped\n",
1346          (unsigned int) payload_size,
1347          GCCH_2s (ch),
1348          ntohl (msg->mid.mid));
1349     GNUNET_STATISTICS_update (stats,
1350                               "# duplicate DATA",
1351                               1,
1352                               GNUNET_NO);
1353     GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1354                                  ccc->tail_recv,
1355                                  com);
1356     ccc->num_recv--;
1357     GNUNET_MQ_discard (com->env);
1358     GNUNET_free (com);
1359     send_channel_data_ack (ch);
1360     return;
1361   }
1362   LOG (GNUNET_ERROR_TYPE_DEBUG,
1363        "Queued %s payload of %u bytes on %s-%X(%p) (mid %u, need %u first)\n",
1364        (GNUNET_YES == ccc->client_ready)
1365        ? "out-of-order"
1366        : "client-not-ready",
1367        (unsigned int) payload_size,
1368        GCCH_2s (ch),
1369        ntohl (ccc->ccn.channel_of_client),
1370        ccc,
1371        ntohl (msg->mid.mid),
1372        ntohl (ch->mid_recv.mid));
1373   /* NOTE: this ACK we _could_ skip, as the packet is out-of-order and
1374      the sender may already be transmitting the previous one.  Needs
1375      experimental evaluation to see if/when this ACK helps or
1376      hurts. (We might even want another option.) */
1377   send_channel_data_ack (ch);
1378 }
1379
1380
1381 /**
1382  * Function called once the tunnel has sent one of our messages.
1383  * If the message is unreliable, simply frees the `crm`. If the
1384  * message was reliable, calculate retransmission time and
1385  * wait for ACK (or retransmit).
1386  *
1387  * @param cls the `struct CadetReliableMessage` that was sent
1388  * @param cid identifier of the connection within the tunnel, NULL
1389  *            if transmission failed
1390  */
1391 static void
1392 data_sent_cb (void *cls,
1393               const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid);
1394
1395
1396 /**
1397  * We need to retry a transmission, the last one took too long to
1398  * be acknowledged.
1399  *
1400  * @param cls the `struct CadetChannel` where we need to retransmit
1401  */
1402 static void
1403 retry_transmission (void *cls)
1404 {
1405   struct CadetChannel *ch = cls;
1406   struct CadetReliableMessage *crm = ch->head_sent;
1407
1408   ch->retry_data_task = NULL;
1409   GNUNET_assert (NULL == crm->qe);
1410   LOG (GNUNET_ERROR_TYPE_DEBUG,
1411        "Retrying transmission on %s of message %u\n",
1412        GCCH_2s (ch),
1413        (unsigned int) ntohl (crm->data_message->mid.mid));
1414   crm->qe = GCT_send (ch->t,
1415                       &crm->data_message->header,
1416                       &data_sent_cb,
1417                       crm);
1418   GNUNET_assert (NULL == ch->retry_data_task);
1419 }
1420
1421
1422 /**
1423  * We got an PLAINTEXT_DATA_ACK for a message in our queue, remove it from
1424  * the queue and tell our client that it can send more.
1425  *
1426  * @param ch the channel that got the PLAINTEXT_DATA_ACK
1427  * @param cti identifier of the connection that delivered the message
1428  * @param crm the message that got acknowledged
1429  */
1430 static void
1431 handle_matching_ack (struct CadetChannel *ch,
1432                      const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1433                      struct CadetReliableMessage *crm)
1434 {
1435   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1436                                ch->tail_sent,
1437                                crm);
1438   ch->pending_messages--;
1439   GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
1440   LOG (GNUNET_ERROR_TYPE_DEBUG,
1441        "Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
1442        GCCH_2s (ch),
1443        (unsigned int) ntohl (crm->data_message->mid.mid),
1444        ch->pending_messages);
1445   if (NULL != crm->qe)
1446   {
1447     GCT_send_cancel (crm->qe);
1448     crm->qe = NULL;
1449   }
1450   if ( (1 == crm->num_transmissions) &&
1451        (NULL != cti) )
1452   {
1453     GCC_ack_observed (cti);
1454     if (0 == memcmp (cti,
1455                      &crm->connection_taken,
1456                      sizeof (struct GNUNET_CADET_ConnectionTunnelIdentifier)))
1457     {
1458       GCC_latency_observed (cti,
1459                             GNUNET_TIME_absolute_get_duration (crm->first_transmission_time));
1460     }
1461   }
1462   GNUNET_free (crm->data_message);
1463   GNUNET_free (crm);
1464   send_ack_to_client (ch,
1465                       (NULL == ch->owner)
1466                       ? GNUNET_NO
1467                       : GNUNET_YES);
1468 }
1469
1470
1471 /**
1472  * We got an acknowledgement for payload data for a channel.
1473  * Possibly resume transmissions.
1474  *
1475  * @param ch channel that got the ack
1476  * @param cti identifier of the connection that delivered the message
1477  * @param ack details about what was received
1478  */
1479 void
1480 GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
1481                                         const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
1482                                         const struct GNUNET_CADET_ChannelDataAckMessage *ack)
1483 {
1484   struct CadetReliableMessage *crm;
1485   struct CadetReliableMessage *crmn;
1486   int found;
1487   uint32_t mid_base;
1488   uint64_t mid_mask;
1489   unsigned int delta;
1490
1491   GNUNET_break (GNUNET_NO == ch->is_loopback);
1492   if (GNUNET_NO == ch->reliable)
1493   {
1494     /* not expecting ACKs on unreliable channel, odd */
1495     GNUNET_break_op (0);
1496     return;
1497   }
1498   /* mid_base is the MID of the next message that the
1499      other peer expects (i.e. that is missing!), everything
1500      LOWER (but excluding mid_base itself) was received. */
1501   mid_base = ntohl (ack->mid.mid);
1502   mid_mask = GNUNET_htonll (ack->futures);
1503   found = GNUNET_NO;
1504   for (crm = ch->head_sent;
1505         NULL != crm;
1506        crm = crmn)
1507   {
1508     crmn = crm->next;
1509     delta = (unsigned int) (ntohl (crm->data_message->mid.mid) - mid_base);
1510     if (delta >= UINT_MAX - ch->max_pending_messages)
1511     {
1512       /* overflow, means crm was a bit in the past, so this ACK counts for it. */
1513       LOG (GNUNET_ERROR_TYPE_DEBUG,
1514            "Got DATA_ACK with base %u satisfying past message %u on %s\n",
1515            (unsigned int) mid_base,
1516            ntohl (crm->data_message->mid.mid),
1517            GCCH_2s (ch));
1518       handle_matching_ack (ch,
1519                            cti,
1520                            crm);
1521       found = GNUNET_YES;
1522       continue;
1523     }
1524     delta--;
1525     if (delta >= 64)
1526       continue;
1527     LOG (GNUNET_ERROR_TYPE_DEBUG,
1528          "Testing bit %llX for mid %u (base: %u)\n",
1529          (1LLU << delta),
1530          ntohl (crm->data_message->mid.mid),
1531          mid_base);
1532     if (0 != (mid_mask & (1LLU << delta)))
1533     {
1534       LOG (GNUNET_ERROR_TYPE_DEBUG,
1535            "Got DATA_ACK with mask for %u on %s\n",
1536            ntohl (crm->data_message->mid.mid),
1537            GCCH_2s (ch));
1538       handle_matching_ack (ch,
1539                            cti,
1540                            crm);
1541       found = GNUNET_YES;
1542     }
1543   }
1544   if (GNUNET_NO == found)
1545   {
1546     /* ACK for message we already dropped, might have been a
1547        duplicate ACK? Ignore. */
1548     LOG (GNUNET_ERROR_TYPE_DEBUG,
1549          "Duplicate DATA_ACK on %s, ignoring\n",
1550          GCCH_2s (ch));
1551     GNUNET_STATISTICS_update (stats,
1552                               "# duplicate DATA_ACKs",
1553                               1,
1554                               GNUNET_NO);
1555     return;
1556   }
1557   if (NULL != ch->retry_data_task)
1558   {
1559     GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1560     ch->retry_data_task = NULL;
1561   }
1562   if ( (NULL != ch->head_sent) &&
1563        (NULL == ch->head_sent->qe) )
1564     ch->retry_data_task
1565       = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1566                                  &retry_transmission,
1567                                  ch);
1568 }
1569
1570
1571 /**
1572  * Destroy channel, based on the other peer closing the
1573  * connection.  Also needs to remove this channel from
1574  * the tunnel.
1575  *
1576  * @param ch channel to destroy
1577  * @param cti identifier of the connection that delivered the message,
1578  *            NULL if we are simulating receiving a destroy due to shutdown
1579  */
1580 void
1581 GCCH_handle_remote_destroy (struct CadetChannel *ch,
1582                             const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
1583 {
1584   struct CadetChannelClient *ccc;
1585
1586   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1587   LOG (GNUNET_ERROR_TYPE_DEBUG,
1588        "Received remote channel DESTROY for %s\n",
1589        GCCH_2s (ch));
1590   if (GNUNET_YES == ch->destroy)
1591   {
1592     /* Local client already gone, this is instant-death. */
1593     channel_destroy (ch);
1594     return;
1595   }
1596   ccc = (NULL != ch->owner) ? ch->owner : ch->dest;
1597   if (NULL != ccc->head_recv)
1598   {
1599     LOG (GNUNET_ERROR_TYPE_WARNING,
1600          "Lost end of transmission due to remote shutdown on %s\n",
1601          GCCH_2s (ch));
1602     /* FIXME: change API to notify client about truncated transmission! */
1603   }
1604   ch->destroy = GNUNET_YES;
1605   GSC_handle_remote_channel_destroy (ccc->c,
1606                                      ccc->ccn,
1607                                      ch);
1608   channel_destroy (ch);
1609 }
1610
1611
1612 /**
1613  * Test if element @a e1 comes before element @a e2.
1614  *
1615  * @param cls closure, to a flag where we indicate duplicate packets
1616  * @param crm1 an element of to sort
1617  * @param crm2 another element to sort
1618  * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
1619  */
1620 static int
1621 cmp_crm_by_next_retry (void *cls,
1622                        struct CadetReliableMessage *crm1,
1623                        struct CadetReliableMessage *crm2)
1624 {
1625   if (crm1->next_retry.abs_value_us <
1626       crm2->next_retry.abs_value_us)
1627     return GNUNET_YES;
1628   return GNUNET_NO;
1629 }
1630
1631
1632 /**
1633  * Function called once the tunnel has sent one of our messages.
1634  * If the message is unreliable, simply frees the `crm`. If the
1635  * message was reliable, calculate retransmission time and
1636  * wait for ACK (or retransmit).
1637  *
1638  * @param cls the `struct CadetReliableMessage` that was sent
1639  * @param cid identifier of the connection within the tunnel, NULL
1640  *            if transmission failed
1641  */
1642 static void
1643 data_sent_cb (void *cls,
1644               const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
1645 {
1646   struct CadetReliableMessage *crm = cls;
1647   struct CadetChannel *ch = crm->ch;
1648
1649   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1650   GNUNET_assert (NULL != crm->qe);
1651   crm->qe = NULL;
1652   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1653                                ch->tail_sent,
1654                                crm);
1655   if (GNUNET_NO == ch->reliable)
1656   {
1657     GNUNET_free (crm->data_message);
1658     GNUNET_free (crm);
1659     ch->pending_messages--;
1660     send_ack_to_client (ch,
1661                         (NULL == ch->owner)
1662                         ? GNUNET_NO
1663                         : GNUNET_YES);
1664     return;
1665   }
1666   if (NULL == cid)
1667   {
1668     /* There was an error sending. */
1669     crm->num_transmissions = GNUNET_SYSERR;
1670   }
1671   else if (GNUNET_SYSERR != crm->num_transmissions)
1672   {
1673     /* Increment transmission counter, and possibly store @a cid
1674        if this was the first transmission. */
1675     crm->num_transmissions++;
1676     if (1 == crm->num_transmissions)
1677     {
1678       crm->first_transmission_time = GNUNET_TIME_absolute_get ();
1679       crm->connection_taken = *cid;
1680       GCC_ack_expected (cid);
1681     }
1682   }
1683   if ( (0 == crm->retry_delay.rel_value_us) &&
1684        (NULL != cid) )
1685   {
1686     struct CadetConnection *cc = GCC_lookup (cid);
1687
1688     if (NULL != cc)
1689       crm->retry_delay = GCC_get_metrics (cc)->aged_latency;
1690     else
1691       crm->retry_delay = ch->retry_time;
1692   }
1693   crm->retry_delay = GNUNET_TIME_STD_BACKOFF (crm->retry_delay);
1694   crm->retry_delay = GNUNET_TIME_relative_max (crm->retry_delay,
1695                                                MIN_RTT_DELAY);
1696   crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
1697
1698   GNUNET_CONTAINER_DLL_insert_sorted (struct CadetReliableMessage,
1699                                       cmp_crm_by_next_retry,
1700                                       NULL,
1701                                       ch->head_sent,
1702                                       ch->tail_sent,
1703                                       crm);
1704   LOG (GNUNET_ERROR_TYPE_DEBUG,
1705        "Message %u sent, next transmission on %s in %s\n",
1706        (unsigned int) ntohl (crm->data_message->mid.mid),
1707        GCCH_2s (ch),
1708        GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (ch->head_sent->next_retry),
1709                                                GNUNET_YES));
1710   if (NULL == ch->head_sent->qe)
1711   {
1712     if (NULL != ch->retry_data_task)
1713       GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1714     ch->retry_data_task
1715       = GNUNET_SCHEDULER_add_at (ch->head_sent->next_retry,
1716                                  &retry_transmission,
1717                                  ch);
1718   }
1719 }
1720
1721
1722 /**
1723  * Handle data given by a client.
1724  *
1725  * Check whether the client is allowed to send in this tunnel, save if
1726  * channel is reliable and send an ACK to the client if there is still
1727  * buffer space in the tunnel.
1728  *
1729  * @param ch Channel.
1730  * @param sender_ccn ccn of the sender
1731  * @param buf payload to transmit.
1732  * @param buf_len number of bytes in @a buf
1733  * @return #GNUNET_OK if everything goes well,
1734  *         #GNUNET_SYSERR in case of an error.
1735  */
1736 int
1737 GCCH_handle_local_data (struct CadetChannel *ch,
1738                         struct GNUNET_CADET_ClientChannelNumber sender_ccn,
1739                         const char *buf,
1740                         size_t buf_len)
1741 {
1742   struct CadetReliableMessage *crm;
1743
1744   if (ch->pending_messages > ch->max_pending_messages)
1745   {
1746     GNUNET_break (0);
1747     return GNUNET_SYSERR;
1748   }
1749   if (GNUNET_YES == ch->destroy)
1750   {
1751     /* we are going down, drop messages */
1752     return GNUNET_OK;
1753   }
1754   ch->pending_messages++;
1755
1756   if (GNUNET_YES == ch->is_loopback)
1757   {
1758     struct CadetChannelClient *receiver;
1759     struct GNUNET_MQ_Envelope *env;
1760     struct GNUNET_CADET_LocalData *ld;
1761     int to_owner;
1762
1763     env = GNUNET_MQ_msg_extra (ld,
1764                                buf_len,
1765                                GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1766     if ( (NULL != ch->owner) &&
1767          (sender_ccn.channel_of_client ==
1768           ch->owner->ccn.channel_of_client) )
1769     {
1770       receiver = ch->dest;
1771       to_owner = GNUNET_NO;
1772     }
1773     else if ( (NULL != ch->dest) &&
1774               (sender_ccn.channel_of_client ==
1775                ch->dest->ccn.channel_of_client) )
1776     {
1777       receiver = ch->owner;
1778       to_owner = GNUNET_YES;
1779     }
1780     else
1781     {
1782       GNUNET_break (0);
1783       return GNUNET_SYSERR;
1784     }
1785     ld->ccn = receiver->ccn;
1786     GNUNET_memcpy (&ld[1],
1787                    buf,
1788                    buf_len);
1789     if (GNUNET_YES == receiver->client_ready)
1790     {
1791       GSC_send_to_client (receiver->c,
1792                           env);
1793       send_ack_to_client (ch,
1794                           to_owner);
1795     }
1796     else
1797     {
1798       struct CadetOutOfOrderMessage *oom;
1799
1800       oom = GNUNET_new (struct CadetOutOfOrderMessage);
1801       oom->env = env;
1802       GNUNET_CONTAINER_DLL_insert_tail (receiver->head_recv,
1803                                         receiver->tail_recv,
1804                                         oom);
1805       receiver->num_recv++;
1806     }
1807     return GNUNET_OK;
1808   }
1809
1810   /* Everything is correct, send the message. */
1811   crm = GNUNET_malloc (sizeof (*crm));
1812   crm->ch = ch;
1813   crm->data_message = GNUNET_malloc (sizeof (struct GNUNET_CADET_ChannelAppDataMessage)
1814                                      + buf_len);
1815   crm->data_message->header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1816   crm->data_message->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
1817   ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
1818   crm->data_message->mid = ch->mid_send;
1819   crm->data_message->ctn = ch->ctn;
1820   GNUNET_memcpy (&crm->data_message[1],
1821                  buf,
1822                  buf_len);
1823   GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent,
1824                                     ch->tail_sent,
1825                                     crm);
1826   LOG (GNUNET_ERROR_TYPE_DEBUG,
1827        "Sending message %u from local client to %s with %u bytes\n",
1828        ntohl (crm->data_message->mid.mid),
1829        GCCH_2s (ch),
1830        buf_len);
1831   if (NULL != ch->retry_data_task)
1832   {
1833     GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1834     ch->retry_data_task = NULL;
1835   }
1836   crm->qe = GCT_send (ch->t,
1837                       &crm->data_message->header,
1838                       &data_sent_cb,
1839                       crm);
1840   GNUNET_assert (NULL == ch->retry_data_task);
1841   return GNUNET_OK;
1842 }
1843
1844
1845 /**
1846  * Handle ACK from client on local channel.  Means the client is ready
1847  * for more data, see if we have any for it.
1848  *
1849  * @param ch channel to destroy
1850  * @param client_ccn ccn of the client sending the ack
1851  */
1852 void
1853 GCCH_handle_local_ack (struct CadetChannel *ch,
1854                        struct GNUNET_CADET_ClientChannelNumber client_ccn)
1855 {
1856   struct CadetChannelClient *ccc;
1857   struct CadetOutOfOrderMessage *com;
1858
1859   if ( (NULL != ch->owner) &&
1860        (ch->owner->ccn.channel_of_client == client_ccn.channel_of_client) )
1861     ccc = ch->owner;
1862   else if ( (NULL != ch->dest) &&
1863             (ch->dest->ccn.channel_of_client == client_ccn.channel_of_client) )
1864     ccc = ch->dest;
1865   else
1866     GNUNET_assert (0);
1867   ccc->client_ready = GNUNET_YES;
1868   com = ccc->head_recv;
1869   if (NULL == com)
1870   {
1871     LOG (GNUNET_ERROR_TYPE_DEBUG,
1872          "Got LOCAL_ACK, %s-%X ready to receive more data, but none pending on %s-%X(%p)!\n",
1873          GSC_2s (ccc->c),
1874          ntohl (client_ccn.channel_of_client),
1875          GCCH_2s (ch),
1876          ntohl (ccc->ccn.channel_of_client),
1877          ccc);
1878     return; /* none pending */
1879   }
1880   if (GNUNET_YES == ch->is_loopback)
1881   {
1882     int to_owner;
1883
1884     /* Messages are always in-order, just send */
1885     GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1886                                  ccc->tail_recv,
1887                                  com);
1888     ccc->num_recv--;
1889     GSC_send_to_client (ccc->c,
1890                         com->env);
1891     /* Notify sender that we can receive more */
1892     if (ccc->ccn.channel_of_client ==
1893         ch->owner->ccn.channel_of_client)
1894     {
1895       to_owner = GNUNET_NO;
1896     }
1897     else
1898     {
1899       GNUNET_assert (ccc->ccn.channel_of_client ==
1900                      ch->dest->ccn.channel_of_client);
1901       to_owner = GNUNET_YES;
1902     }
1903     send_ack_to_client (ch,
1904                         to_owner);
1905     GNUNET_free (com);
1906     return;
1907   }
1908
1909   if ( (com->mid.mid != ch->mid_recv.mid) &&
1910        (GNUNET_NO == ch->out_of_order) &&
1911        (GNUNET_YES == ch->reliable) )
1912   {
1913     LOG (GNUNET_ERROR_TYPE_DEBUG,
1914          "Got LOCAL_ACK, %s-%X ready to receive more data (but next one is out-of-order %u vs. %u)!\n",
1915          GSC_2s (ccc->c),
1916          ntohl (ccc->ccn.channel_of_client),
1917          ntohl (com->mid.mid),
1918          ntohl (ch->mid_recv.mid));
1919     return; /* missing next one in-order */
1920   }
1921
1922   LOG (GNUNET_ERROR_TYPE_DEBUG,
1923        "Got LOCAL_ACK, giving payload message %u to %s-%X on %s\n",
1924        ntohl (com->mid.mid),
1925        GSC_2s (ccc->c),
1926        ntohl (ccc->ccn.channel_of_client),
1927        GCCH_2s (ch));
1928
1929   /* all good, pass next message to client */
1930   GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
1931                                ccc->tail_recv,
1932                                com);
1933   ccc->num_recv--;
1934   /* FIXME: if unreliable, this is not aggressive
1935      enough, as it would be OK to have lost some! */
1936
1937   ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
1938   ch->mid_futures >>= 1; /* equivalent to division by 2 */
1939   ccc->client_ready = GNUNET_NO;
1940   GSC_send_to_client (ccc->c,
1941                       com->env);
1942   GNUNET_free (com);
1943   send_channel_data_ack (ch);
1944   if (NULL != ccc->head_recv)
1945     return;
1946   if (GNUNET_NO == ch->destroy)
1947     return;
1948   GCT_send_channel_destroy (ch->t,
1949                             ch->ctn);
1950   channel_destroy (ch);
1951 }
1952
1953
1954 #define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-chn",__VA_ARGS__)
1955
1956
1957 /**
1958  * Log channel info.
1959  *
1960  * @param ch Channel.
1961  * @param level Debug level to use.
1962  */
1963 void
1964 GCCH_debug (struct CadetChannel *ch,
1965             enum GNUNET_ErrorType level)
1966 {
1967   int do_log;
1968
1969   do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
1970                                        "cadet-chn",
1971                                        __FILE__, __FUNCTION__, __LINE__);
1972   if (0 == do_log)
1973     return;
1974
1975   if (NULL == ch)
1976   {
1977     LOG2 (level, "CHN *** DEBUG NULL CHANNEL ***\n");
1978     return;
1979   }
1980   LOG2 (level,
1981         "CHN %s:%X (%p)\n",
1982         GCT_2s (ch->t),
1983         ch->ctn,
1984         ch);
1985   if (NULL != ch->owner)
1986   {
1987     LOG2 (level,
1988           "CHN origin %s ready %s local-id: %u\n",
1989           GSC_2s (ch->owner->c),
1990           ch->owner->client_ready ? "YES" : "NO",
1991           ntohl (ch->owner->ccn.channel_of_client));
1992   }
1993   if (NULL != ch->dest)
1994   {
1995     LOG2 (level,
1996           "CHN destination %s ready %s local-id: %u\n",
1997           GSC_2s (ch->dest->c),
1998           ch->dest->client_ready ? "YES" : "NO",
1999           ntohl (ch->dest->ccn.channel_of_client));
2000   }
2001   LOG2 (level,
2002         "CHN  Message IDs recv: %d (%LLX), send: %d\n",
2003         ntohl (ch->mid_recv.mid),
2004         (unsigned long long) ch->mid_futures,
2005         ntohl (ch->mid_send.mid));
2006 }
2007
2008
2009
2010 /* end of gnunet-service-cadet-new_channel.c */