src/cadet/gnunet-service-cadet-new_channel.c

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