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