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  * - introduce shutdown so we can have half-closed channels, modify
  29  *   destroy to include MID to have FIN-ACK equivalents, etc.
  30  * - estimate max bandwidth using bursts and use to for CONGESTION CONTROL!
  31  * - check that '0xFFULL' really is sufficient for flow control!
  32  * - revisit handling of 'unreliable' traffic!
  33  * - revisit handling of 'out-of-order' option, especially in combination with/without 'reliable'.
  34  * - figure out flow control without ACKs (unreliable traffic!)
  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 /**
  62  * All the states a connection can be in.
  63  */
  64 enum CadetChannelState
  65 {
  66   /**
  67    * Uninitialized status, should never appear in operation.
  68    */
  69   CADET_CHANNEL_NEW,
  70
  71   /**
  72    * Connection create message sent, waiting for ACK.
  73    */
  74   CADET_CHANNEL_OPEN_SENT,
  75
  76   /**
  77    * Connection confirmed, ready to carry traffic.
  78    */
  79   CADET_CHANNEL_READY
  80 };
  81
  82
  83 /**
  84  * Info needed to retry a message in case it gets lost.
  85  * Note that we DO use this structure also for unreliable
  86  * messages.
  87  */
  88 struct CadetReliableMessage
  89 {
  90   /**
  91    * Double linked list, FIFO style
  92    */
  93   struct CadetReliableMessage *next;
  94
  95   /**
  96    * Double linked list, FIFO style
  97    */
  98   struct CadetReliableMessage *prev;
  99
 100   /**
 101    * Which channel is this message in?
 102    */
 103   struct CadetChannel *ch;
 104
 105   /**
 106    * Entry in the tunnels queue for this message, NULL if it has left
 107    * the tunnel.  Used to cancel transmission in case we receive an
 108    * ACK in time.
 109    */
 110   struct CadetTunnelQueueEntry *qe;
 111
 112   /**
 113    * How soon should we retry if we fail to get an ACK?
 114    * Messages in the queue are sorted by this value.
 115    */
 116   struct GNUNET_TIME_Absolute next_retry;
 117
 118   /**
 119    * How long do we wait for an ACK after transmission?
 120    * Use for the back-off calculation.
 121    */
 122   struct GNUNET_TIME_Relative retry_delay;
 123
 124   /**
 125    * Data message we are trying to send.
 126    */
 127   struct GNUNET_CADET_ChannelAppDataMessage data_message;
 128
 129   /* followed by variable-size payload */
 130 };
 131
 132
 133 /**
 134  * List of received out-of-order data messages.
 135  */
 136 struct CadetOutOfOrderMessage
 137 {
 138   /**
 139    * Double linked list, FIFO style
 140    */
 141   struct CadetOutOfOrderMessage *next;
 142
 143   /**
 144    * Double linked list, FIFO style
 145    */
 146   struct CadetOutOfOrderMessage *prev;
 147
 148   /**
 149    * ID of the message (messages up to this point needed
 150    * before we give this one to the client).
 151    */
 152   struct ChannelMessageIdentifier mid;
 153
 154   /**
 155    * The envelope with the payload of the out-of-order message
 156    */
 157   struct GNUNET_MQ_Envelope *env;
 158
 159 };
 160
 161
 162 /**
 163  * Struct containing all information regarding a channel to a remote client.
 164  */
 165 struct CadetChannel
 166 {
 167   /**
 168    * Tunnel this channel is in.
 169    */
 170   struct CadetTunnel *t;
 171
 172   /**
 173    * Client owner of the tunnel, if any.
 174    * (Used if this channel represends the initiating end of the tunnel.)
 175    */
 176   struct CadetClient *owner;
 177
 178   /**
 179    * Client destination of the tunnel, if any.
 180    * (Used if this channel represents the listening end of the tunnel.)
 181    */
 182   struct CadetClient *dest;
 183
 184   /**
 185    * Last entry in the tunnel's queue relating to control messages
 186    * (#GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN or
 187    * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK).  Used to cancel
 188    * transmission in case we receive updated information.
 189    */
 190   struct CadetTunnelQueueEntry *last_control_qe;
 191
 192   /**
 193    * Head of DLL of messages sent and not yet ACK'd.
 194    */
 195   struct CadetReliableMessage *head_sent;
 196
 197   /**
 198    * Tail of DLL of messages sent and not yet ACK'd.
 199    */
 200   struct CadetReliableMessage *tail_sent;
 201
 202   /**
 203    * Head of DLL of messages received out of order or while client was unready.
 204    */
 205   struct CadetOutOfOrderMessage *head_recv;
 206
 207   /**
 208    * Tail DLL of messages received out of order or while client was unready.
 209    */
 210   struct CadetOutOfOrderMessage *tail_recv;
 211
 212   /**
 213    * Task to resend/poll in case no ACK is received.
 214    */
 215   struct GNUNET_SCHEDULER_Task *retry_control_task;
 216
 217   /**
 218    * Task to resend/poll in case no ACK is received.
 219    */
 220   struct GNUNET_SCHEDULER_Task *retry_data_task;
 221
 222   /**
 223    * Last time the channel was used
 224    */
 225   struct GNUNET_TIME_Absolute timestamp;
 226
 227   /**
 228    * Destination port of the channel.
 229    */
 230   struct GNUNET_HashCode port;
 231
 232   /**
 233    * Counter for exponential backoff.
 234    */
 235   struct GNUNET_TIME_Relative retry_time;
 236
 237   /**
 238    * How long does it usually take to get an ACK.
 239    */
 240   struct GNUNET_TIME_Relative expected_delay;
 241
 242   /**
 243    * Bitfield of already-received messages past @e mid_recv.
 244    */
 245   uint64_t mid_futures;
 246
 247   /**
 248    * Next MID expected for incoming traffic.
 249    */
 250   struct ChannelMessageIdentifier mid_recv;
 251
 252   /**
 253    * Next MID to use for outgoing traffic.
 254    */
 255   struct ChannelMessageIdentifier mid_send;
 256
 257   /**
 258    * Total (reliable) messages pending ACK for this channel.
 259    */
 260   unsigned int pending_messages;
 261
 262   /**
 263    * Maximum (reliable) messages pending ACK for this channel
 264    * before we throttle the client.
 265    */
 266   unsigned int max_pending_messages;
 267
 268   /**
 269    * Number identifying this channel in its tunnel.
 270    */
 271   struct GNUNET_CADET_ChannelTunnelNumber ctn;
 272
 273   /**
 274    * Local tunnel number for local client @e owner owning the channel.
 275    * ( >= #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
 276    */
 277   struct GNUNET_CADET_ClientChannelNumber ccn_owner;
 278
 279   /**
 280    * Local tunnel number for local client @e dest owning the channel.
 281    * (< #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
 282    */
 283   struct GNUNET_CADET_ClientChannelNumber ccn_dest;
 284
 285   /**
 286    * Channel state.
 287    */
 288   enum CadetChannelState state;
 289
 290   /**
 291    * Can we send data to the client?
 292    */
 293   int client_ready;
 294
 295   /**
 296    * Is the tunnel bufferless (minimum latency)?
 297    */
 298   int nobuffer;
 299
 300   /**
 301    * Is the tunnel reliable?
 302    */
 303   int reliable;
 304
 305   /**
 306    * Is the tunnel out-of-order?
 307    */
 308   int out_of_order;
 309
 310   /**
 311    * Is this channel a loopback channel, where the destination is us again?
 312    */
 313   int is_loopback;
 314
 315   /**
 316    * Flag to signal the destruction of the channel.  If this is set to
 317    * #GNUNET_YES the channel will be destroyed once the queue is
 318    * empty.
 319    */
 320   int destroy;
 321
 322 };
 323
 324
 325 /**
 326  * Get the static string for identification of the channel.
 327  *
 328  * @param ch Channel.
 329  *
 330  * @return Static string with the channel IDs.
 331  */
 332 const char *
 333 GCCH_2s (const struct CadetChannel *ch)
 334 {
 335   static char buf[128];
 336
 337   GNUNET_snprintf (buf,
 338                    sizeof (buf),
 339                    "Channel %s:%s ctn:%X(%X/%X)",
 340                    GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
 341                    GNUNET_h2s (&ch->port),
 342                    ch->ctn,
 343                    ntohl (ch->ccn_owner.channel_of_client),
 344                    ntohl (ch->ccn_dest.channel_of_client));
 345   return buf;
 346 }
 347
 348
 349 /**
 350  * Get the channel's public ID.
 351  *
 352  * @param ch Channel.
 353  *
 354  * @return ID used to identify the channel with the remote peer.
 355  */
 356 struct GNUNET_CADET_ChannelTunnelNumber
 357 GCCH_get_id (const struct CadetChannel *ch)
 358 {
 359   return ch->ctn;
 360 }
 361
 362
 363 /**
 364  * Destroy the given channel.
 365  *
 366  * @param ch channel to destroy
 367  */
 368 static void
 369 channel_destroy (struct CadetChannel *ch)
 370 {
 371   struct CadetReliableMessage *crm;
 372   struct CadetOutOfOrderMessage *com;
 373
 374   while (NULL != (crm = ch->head_sent))
 375   {
 376     GNUNET_assert (ch == crm->ch);
 377     if (NULL != crm->qe)
 378     {
 379       GCT_send_cancel (crm->qe);
 380       crm->qe = NULL;
 381     }
 382     GNUNET_CONTAINER_DLL_remove (ch->head_sent,
 383                                  ch->tail_sent,
 384                                  crm);
 385     GNUNET_free (crm);
 386   }
 387   while (NULL != (com = ch->head_recv))
 388   {
 389     GNUNET_CONTAINER_DLL_remove (ch->head_recv,
 390                                  ch->tail_recv,
 391                                  com);
 392     GNUNET_MQ_discard (com->env);
 393     GNUNET_free (com);
 394   }
 395   if (NULL != ch->last_control_qe)
 396   {
 397     GCT_send_cancel (ch->last_control_qe);
 398     ch->last_control_qe = NULL;
 399   }
 400   if (NULL != ch->retry_data_task)
 401   {
 402     GNUNET_SCHEDULER_cancel (ch->retry_data_task);
 403     ch->retry_data_task = NULL;
 404   }
 405   if (NULL != ch->retry_control_task)
 406   {
 407     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 408     ch->retry_control_task = NULL;
 409   }
 410   if (GNUNET_NO == ch->is_loopback)
 411   {
 412     GCT_remove_channel (ch->t,
 413                         ch,
 414                         ch->ctn);
 415     ch->t = NULL;
 416   }
 417   GNUNET_free (ch);
 418 }
 419
 420
 421 /**
 422  * Send a channel create message.
 423  *
 424  * @param cls Channel for which to send.
 425  */
 426 static void
 427 send_channel_open (void *cls);
 428
 429
 430 /**
 431  * Function called once the tunnel confirms that we sent the
 432  * create message.  Delays for a bit until we retry.
 433  *
 434  * @param cls our `struct CadetChannel`.
 435  */
 436 static void
 437 channel_open_sent_cb (void *cls)
 438 {
 439   struct CadetChannel *ch = cls;
 440
 441   GNUNET_assert (NULL != ch->last_control_qe);
 442   ch->last_control_qe = NULL;
 443   ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
 444   LOG (GNUNET_ERROR_TYPE_DEBUG,
 445        "Sent CHANNEL_OPEN on %s, retrying in %s\n",
 446        GCCH_2s (ch),
 447        GNUNET_STRINGS_relative_time_to_string (ch->retry_time,
 448                                                GNUNET_YES));
 449   ch->retry_control_task
 450     = GNUNET_SCHEDULER_add_delayed (ch->retry_time,
 451                                     &send_channel_open,
 452                                     ch);
 453 }
 454
 455
 456 /**
 457  * Send a channel open message.
 458  *
 459  * @param cls Channel for which to send.
 460  */
 461 static void
 462 send_channel_open (void *cls)
 463 {
 464   struct CadetChannel *ch = cls;
 465   struct GNUNET_CADET_ChannelOpenMessage msgcc;
 466   uint32_t options;
 467
 468   ch->retry_control_task = NULL;
 469   LOG (GNUNET_ERROR_TYPE_DEBUG,
 470        "Sending CHANNEL_OPEN message for %s\n",
 471        GCCH_2s (ch));
 472   options = 0;
 473   if (ch->nobuffer)
 474     options |= GNUNET_CADET_OPTION_NOBUFFER;
 475   if (ch->reliable)
 476     options |= GNUNET_CADET_OPTION_RELIABLE;
 477   if (ch->out_of_order)
 478     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
 479   msgcc.header.size = htons (sizeof (msgcc));
 480   msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN);
 481   msgcc.opt = htonl (options);
 482   msgcc.port = ch->port;
 483   msgcc.ctn = ch->ctn;
 484   ch->state = CADET_CHANNEL_OPEN_SENT;
 485   ch->last_control_qe = GCT_send (ch->t,
 486                                   &msgcc.header,
 487                                   &channel_open_sent_cb,
 488                                   ch);
 489 }
 490
 491
 492 /**
 493  * Function called once and only once after a channel was bound
 494  * to its tunnel via #GCT_add_channel() is ready for transmission.
 495  * Note that this is only the case for channels that this peer
 496  * initiates, as for incoming channels we assume that they are
 497  * ready for transmission immediately upon receiving the open
 498  * message.  Used to bootstrap the #GCT_send() process.
 499  *
 500  * @param ch the channel for which the tunnel is now ready
 501  */
 502 void
 503 GCCH_tunnel_up (struct CadetChannel *ch)
 504 {
 505   GNUNET_assert (NULL == ch->retry_control_task);
 506   LOG (GNUNET_ERROR_TYPE_DEBUG,
 507        "Tunnel up, sending CHANNEL_OPEN on %s now\n",
 508        GCCH_2s (ch));
 509   ch->retry_control_task
 510     = GNUNET_SCHEDULER_add_now (&send_channel_open,
 511                                 ch);
 512 }
 513
 514
 515 /**
 516  * Create a new channel.
 517  *
 518  * @param owner local client owning the channel
 519  * @param ccn local number of this channel at the @a owner
 520  * @param destination peer to which we should build the channel
 521  * @param port desired port at @a destination
 522  * @param options options for the channel
 523  * @return handle to the new channel
 524  */
 525 struct CadetChannel *
 526 GCCH_channel_local_new (struct CadetClient *owner,
 527                         struct GNUNET_CADET_ClientChannelNumber ccn,
 528                         struct CadetPeer *destination,
 529                         const struct GNUNET_HashCode *port,
 530                         uint32_t options)
 531 {
 532   struct CadetChannel *ch;
 533
 534   ch = GNUNET_new (struct CadetChannel);
 535   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
 536   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
 537   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
 538   ch->max_pending_messages = (ch->nobuffer) ? 1 : 32; /* FIXME: 32!? Do not hardcode! */
 539   ch->owner = owner;
 540   ch->ccn_owner = ccn;
 541   ch->port = *port;
 542   if (0 == memcmp (&my_full_id,
 543                    GCP_get_id (destination),
 544                    sizeof (struct GNUNET_PeerIdentity)))
 545   {
 546     ch->is_loopback = GNUNET_YES;
 547     ch->dest = GNUNET_CONTAINER_multihashmap_get (open_ports,
 548                                                   port);
 549     if (NULL == ch->dest)
 550     {
 551       /* port closed, wait for it to possibly open */
 552       (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
 553                                                 port,
 554                                                 ch,
 555                                                 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 556       LOG (GNUNET_ERROR_TYPE_DEBUG,
 557            "Created loose incoming loopback channel to port %s\n",
 558            GNUNET_h2s (&ch->port));
 559     }
 560     else
 561     {
 562       GCCH_bind (ch,
 563                  ch->dest);
 564     }
 565   }
 566   else
 567   {
 568     ch->t = GCP_get_tunnel (destination,
 569                             GNUNET_YES);
 570     ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
 571     ch->ctn = GCT_add_channel (ch->t,
 572                                ch);
 573   }
 574   GNUNET_STATISTICS_update (stats,
 575                             "# channels",
 576                             1,
 577                             GNUNET_NO);
 578   LOG (GNUNET_ERROR_TYPE_DEBUG,
 579        "Created channel to port %s at peer %s for %s using %s\n",
 580        GNUNET_h2s (port),
 581        GCP_2s (destination),
 582        GSC_2s (owner),
 583        (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
 584   return ch;
 585 }
 586
 587
 588 /**
 589  * We had an incoming channel to a port that is closed.
 590  * It has not been opened for a while, drop it.
 591  *
 592  * @param cls the channel to drop
 593  */
 594 static void
 595 timeout_closed_cb (void *cls)
 596 {
 597   struct CadetChannel *ch = cls;
 598
 599   ch->retry_control_task = NULL;
 600   LOG (GNUNET_ERROR_TYPE_DEBUG,
 601        "Closing incoming channel to port %s from peer %s due to timeout\n",
 602        GNUNET_h2s (&ch->port),
 603        GCP_2s (GCT_get_destination (ch->t)));
 604   channel_destroy (ch);
 605 }
 606
 607
 608 /**
 609  * Create a new channel based on a request coming in over the network.
 610  *
 611  * @param t tunnel to the remote peer
 612  * @param ctn identifier of this channel in the tunnel
 613  * @param port desired local port
 614  * @param options options for the channel
 615  * @return handle to the new channel
 616  */
 617 struct CadetChannel *
 618 GCCH_channel_incoming_new (struct CadetTunnel *t,
 619                            struct GNUNET_CADET_ChannelTunnelNumber ctn,
 620                            const struct GNUNET_HashCode *port,
 621                            uint32_t options)
 622 {
 623   struct CadetChannel *ch;
 624   struct CadetClient *c;
 625
 626   ch = GNUNET_new (struct CadetChannel);
 627   ch->port = *port;
 628   ch->t = t;
 629   ch->ctn = ctn;
 630   ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
 631   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
 632   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
 633   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
 634   ch->max_pending_messages = (ch->nobuffer) ? 1 : 32; /* FIXME: 32!? Do not hardcode! */
 635   GNUNET_STATISTICS_update (stats,
 636                             "# channels",
 637                             1,
 638                             GNUNET_NO);
 639
 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     ch->retry_control_task
 650       = GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
 651                                       &timeout_closed_cb,
 652                                       ch);
 653     LOG (GNUNET_ERROR_TYPE_DEBUG,
 654          "Created loose incoming channel to port %s from peer %s\n",
 655          GNUNET_h2s (&ch->port),
 656          GCP_2s (GCT_get_destination (ch->t)));
 657   }
 658   else
 659   {
 660     GCCH_bind (ch,
 661                c);
 662   }
 663   GNUNET_STATISTICS_update (stats,
 664                             "# channels",
 665                             1,
 666                             GNUNET_NO);
 667   return ch;
 668 }
 669
 670
 671 /**
 672  * Function called once the tunnel confirms that we sent the
 673  * ACK message.  Just remembers it was sent, we do not expect
 674  * ACKs for ACKs ;-).
 675  *
 676  * @param cls our `struct CadetChannel`.
 677  */
 678 static void
 679 send_ack_cb (void *cls)
 680 {
 681   struct CadetChannel *ch = cls;
 682
 683   GNUNET_assert (NULL != ch->last_control_qe);
 684   ch->last_control_qe = NULL;
 685 }
 686
 687
 688 /**
 689  * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
 690  *
 691  * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
 692  */
 693 static void
 694 send_channel_data_ack (struct CadetChannel *ch)
 695 {
 696   struct GNUNET_CADET_ChannelDataAckMessage msg;
 697
 698   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK);
 699   msg.header.size = htons (sizeof (msg));
 700   msg.ctn = ch->ctn;
 701   msg.mid.mid = htonl (ntohl (ch->mid_recv.mid) - 1);
 702   msg.futures = GNUNET_htonll (ch->mid_futures);
 703   if (NULL != ch->last_control_qe)
 704     GCT_send_cancel (ch->last_control_qe);
 705   ch->last_control_qe = GCT_send (ch->t,
 706                                   &msg.header,
 707                                   &send_ack_cb,
 708                                   ch);
 709 }
 710
 711
 712 /**
 713  * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
 714  * connection is up.
 715  *
 716  * @param cls the `struct CadetChannel`
 717  */
 718 static void
 719 send_open_ack (void *cls)
 720 {
 721   struct CadetChannel *ch = cls;
 722   struct GNUNET_CADET_ChannelManageMessage msg;
 723
 724   LOG (GNUNET_ERROR_TYPE_DEBUG,
 725        "Sending CHANNEL_OPEN_ACK on channel %s\n",
 726        GCCH_2s (ch));
 727   ch->retry_control_task = NULL;
 728   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
 729   msg.header.size = htons (sizeof (msg));
 730   msg.reserved = htonl (0);
 731   msg.ctn = ch->ctn;
 732   if (NULL != ch->last_control_qe)
 733     GCT_send_cancel (ch->last_control_qe);
 734   ch->last_control_qe = GCT_send (ch->t,
 735                                   &msg.header,
 736                                   &send_ack_cb,
 737                                   ch);
 738 }
 739
 740
 741 /**
 742  * We got a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN message again for
 743  * this channel.  If the binding was successful, (re)transmit the
 744  * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
 745  *
 746  * @param ch channel that got the duplicate open
 747  */
 748 void
 749 GCCH_handle_duplicate_open (struct CadetChannel *ch)
 750 {
 751   if (NULL == ch->dest)
 752   {
 753     LOG (GNUNET_ERROR_TYPE_DEBUG,
 754          "Ignoring duplicate channel OPEN on %s: port is closed\n",
 755          GCCH_2s (ch));
 756     return;
 757   }
 758   if (NULL != ch->retry_control_task)
 759   {
 760     LOG (GNUNET_ERROR_TYPE_DEBUG,
 761          "Ignoring duplicate channel OPEN on %s: control message is pending\n",
 762          GCCH_2s (ch));
 763     return;
 764   }
 765   LOG (GNUNET_ERROR_TYPE_DEBUG,
 766        "Retransmitting OPEN_ACK on channel %s\n",
 767        GCCH_2s (ch));
 768   ch->retry_control_task
 769     = GNUNET_SCHEDULER_add_now (&send_open_ack,
 770                                 ch);
 771 }
 772
 773
 774 /**
 775  * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL ACK to the client to solicit more messages.
 776  *
 777  * @param ch channel the ack is for
 778  * @param c client to send the ACK to
 779  */
 780 static void
 781 send_ack_to_client (struct CadetChannel *ch,
 782                     struct CadetClient *c)
 783 {
 784   struct GNUNET_MQ_Envelope *env;
 785   struct GNUNET_CADET_LocalAck *ack;
 786
 787   env = GNUNET_MQ_msg (ack,
 788                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
 789   ack->ccn = (c == ch->owner) ? ch->ccn_owner : ch->ccn_dest;
 790   GSC_send_to_client (c,
 791                       env);
 792 }
 793
 794
 795 /**
 796  * A client is bound to the port that we have a channel
 797  * open to.  Send the acknowledgement for the connection
 798  * request and establish the link with the client.
 799  *
 800  * @param ch open incoming channel
 801  * @param c client listening on the respective port
 802  */
 803 void
 804 GCCH_bind (struct CadetChannel *ch,
 805            struct CadetClient *c)
 806 {
 807   struct GNUNET_MQ_Envelope *env;
 808   struct GNUNET_CADET_LocalChannelCreateMessage *tcm;
 809   uint32_t options;
 810
 811   LOG (GNUNET_ERROR_TYPE_DEBUG,
 812        "Binding %s from %s to port %s of %s\n",
 813        GCCH_2s (ch),
 814        GCT_2s (ch->t),
 815        GNUNET_h2s (&ch->port),
 816        GSC_2s (c));
 817   if (NULL != ch->retry_control_task)
 818   {
 819     /* there might be a timeout task here */
 820     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 821     ch->retry_control_task = NULL;
 822   }
 823   options = 0;
 824   if (ch->nobuffer)
 825     options |= GNUNET_CADET_OPTION_NOBUFFER;
 826   if (ch->reliable)
 827     options |= GNUNET_CADET_OPTION_RELIABLE;
 828   if (ch->out_of_order)
 829     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
 830   ch->dest = c;
 831   ch->ccn_dest = GSC_bind (c,
 832                            ch,
 833                            (GNUNET_YES == ch->is_loopback)
 834                            ? GCP_get (&my_full_id,
 835                                       GNUNET_YES)
 836                            : GCT_get_destination (ch->t),
 837                            &ch->port,
 838                            options);
 839   ch->mid_recv.mid = htonl (1); /* The CONNECT counts as message 0! */
 840   if (GNUNET_YES == ch->is_loopback)
 841   {
 842     ch->state = CADET_CHANNEL_OPEN_SENT;
 843     GCCH_handle_channel_open_ack (ch);
 844   }
 845   else
 846   {
 847     /* notify other peer that we accepted the connection */
 848     ch->retry_control_task
 849       = GNUNET_SCHEDULER_add_now (&send_open_ack,
 850                                   ch);
 851   }
 852   /* give client it's initial supply of ACKs */
 853   env = GNUNET_MQ_msg (tcm,
 854                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE);
 855   tcm->ccn = ch->ccn_dest;
 856   if (GNUNET_YES == ch->is_loopback)
 857     tcm->peer = my_full_id;
 858   else
 859     tcm->peer = *GCP_get_id (GCT_get_destination (ch->t));
 860   tcm->port = ch->port;
 861   tcm->opt = htonl (options);
 862   GSC_send_to_client (ch->dest,
 863                       env);
 864   for (unsigned int i=0;i<ch->max_pending_messages;i++)
 865     send_ack_to_client (ch,
 866                         ch->dest);
 867 }
 868
 869
 870 /**
 871  * Destroy locally created channel.  Called by the local client, so no
 872  * need to tell the client.
 873  *
 874  * @param ch channel to destroy
 875  * @param c client that caused the destruction
 876  */
 877 void
 878 GCCH_channel_local_destroy (struct CadetChannel *ch,
 879                             struct CadetClient *c)
 880 {
 881   LOG (GNUNET_ERROR_TYPE_DEBUG,
 882        "%s asks for destruction of %s\n",
 883        GSC_2s (c),
 884        GCCH_2s (ch));
 885   GNUNET_assert (NULL != c);
 886   if (c == ch->owner)
 887     ch->owner = NULL;
 888   else if (c == ch->dest)
 889     ch->dest = NULL;
 890   else
 891     GNUNET_assert (0);
 892   if (GNUNET_YES == ch->destroy)
 893   {
 894     /* other end already destroyed, with the local client gone, no need
 895        to finish transmissions, just destroy immediately. */
 896     channel_destroy (ch);
 897     return;
 898   }
 899   if ( (NULL != ch->head_sent) ||
 900        (NULL != ch->owner) ||
 901        (NULL != ch->dest) )
 902   {
 903     /* Wait for other end to destroy us as well,
 904        and otherwise allow send queue to be transmitted first */
 905     ch->destroy = GNUNET_YES;
 906     return;
 907   }
 908   /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
 909   if (CADET_CHANNEL_NEW != ch->state)
 910     GCT_send_channel_destroy (ch->t,
 911                               ch->ctn);
 912   /* Nothing left to do, just finish destruction */
 913   channel_destroy (ch);
 914 }
 915
 916
 917 /**
 918  * We got an acknowledgement for the creation of the channel
 919  * (the port is open on the other side). Begin transmissions.
 920  *
 921  * @param ch channel to destroy
 922  */
 923 void
 924 GCCH_handle_channel_open_ack (struct CadetChannel *ch)
 925 {
 926   switch (ch->state)
 927   {
 928   case CADET_CHANNEL_NEW:
 929     /* this should be impossible */
 930     GNUNET_break (0);
 931     break;
 932   case CADET_CHANNEL_OPEN_SENT:
 933     if (NULL == ch->owner)
 934     {
 935       /* We're not the owner, wrong direction! */
 936       GNUNET_break_op (0);
 937       return;
 938     }
 939     LOG (GNUNET_ERROR_TYPE_DEBUG,
 940          "Received channel OPEN_ACK for waiting %s, entering READY state\n",
 941          GCCH_2s (ch));
 942     if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
 943     {
 944       GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 945       ch->retry_control_task = NULL;
 946     }
 947     ch->state = CADET_CHANNEL_READY;
 948     /* On first connect, send client as many ACKs as we allow messages
 949        to be buffered! */
 950     for (unsigned int i=0;i<ch->max_pending_messages;i++)
 951       send_ack_to_client (ch,
 952                           ch->owner);
 953     break;
 954   case CADET_CHANNEL_READY:
 955     /* duplicate ACK, maybe we retried the CREATE. Ignore. */
 956     LOG (GNUNET_ERROR_TYPE_DEBUG,
 957          "Received duplicate channel OPEN_ACK for %s\n",
 958          GCCH_2s (ch));
 959     GNUNET_STATISTICS_update (stats,
 960                               "# duplicate CREATE_ACKs",
 961                               1,
 962                               GNUNET_NO);
 963     break;
 964   }
 965 }
 966
 967
 968 /**
 969  * Test if element @a e1 comes before element @a e2.
 970  *
 971  * TODO: use opportunity to create generic list insertion sort
 972  * logic in container!
 973  *
 974  * @param cls closure, our `struct CadetChannel`
 975  * @param e1 an element of to sort
 976  * @param e2 another element to sort
 977  * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
 978  */
 979 static int
 980 is_before (void *cls,
 981            void *e1,
 982            void *e2)
 983 {
 984   struct CadetOutOfOrderMessage *m1 = e1;
 985   struct CadetOutOfOrderMessage *m2 = e2;
 986   uint32_t v1 = ntohl (m1->mid.mid);
 987   uint32_t v2 = ntohl (m2->mid.mid);
 988   uint32_t delta;
 989
 990   delta = v1 - v2;
 991   if (delta > (uint32_t) INT_MAX)
 992   {
 993     /* in overflow range, we can safely assume we wrapped around */
 994     return GNUNET_NO;
 995   }
 996   else
 997   {
 998     return GNUNET_YES;
 999   }
1000 }
1001
1002
1003 /**
1004  * We got payload data for a channel.  Pass it on to the client
1005  * and send an ACK to the other end (once flow control allows it!)
1006  *
1007  * @param ch channel that got data
1008  */
1009 void
1010 GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
1011                                     const struct GNUNET_CADET_ChannelAppDataMessage *msg)
1012 {
1013   struct GNUNET_MQ_Envelope *env;
1014   struct GNUNET_CADET_LocalData *ld;
1015   struct CadetOutOfOrderMessage *com;
1016   size_t payload_size;
1017
1018   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1019   payload_size = ntohs (msg->header.size) - sizeof (*msg);
1020   LOG (GNUNET_ERROR_TYPE_DEBUG,
1021        "Receicved %u bytes of application data on %s\n",
1022        (unsigned int) payload_size,
1023        GCCH_2s (ch));
1024   env = GNUNET_MQ_msg_extra (ld,
1025                              payload_size,
1026                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1027   ld->ccn = (NULL == ch->dest) ? ch->ccn_owner : ch->ccn_dest;
1028   GNUNET_memcpy (&ld[1],
1029                  &msg[1],
1030                  payload_size);
1031   if ( (GNUNET_YES == ch->client_ready) &&
1032        ( (GNUNET_YES == ch->out_of_order) ||
1033          (msg->mid.mid == ch->mid_recv.mid) ) )
1034   {
1035     GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
1036                         env);
1037     ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
1038     ch->mid_futures >>= 1;
1039   }
1040   else
1041   {
1042     /* FIXME-SECURITY: if the element is WAY too far ahead,
1043        drop it (can't buffer too much!) */
1044     com = GNUNET_new (struct CadetOutOfOrderMessage);
1045     com->mid = msg->mid;
1046     com->env = env;
1047     /* sort into list ordered by "is_before" */
1048     if ( (NULL == ch->head_recv) ||
1049          (GNUNET_YES == is_before (ch,
1050                                    com,
1051                                    ch->head_recv)) )
1052     {
1053       GNUNET_CONTAINER_DLL_insert (ch->head_recv,
1054                                    ch->tail_recv,
1055                                    com);
1056     }
1057     else
1058     {
1059       struct CadetOutOfOrderMessage *pos;
1060
1061       for (pos = ch->head_recv;
1062            NULL != pos;
1063            pos = pos->next)
1064       {
1065         if (GNUNET_YES !=
1066             is_before (ch,
1067                        pos,
1068                        com))
1069           break;
1070       }
1071       if (NULL == pos)
1072         GNUNET_CONTAINER_DLL_insert_tail (ch->head_recv,
1073                                           ch->tail_recv,
1074                                           com);
1075       else
1076         GNUNET_CONTAINER_DLL_insert_after (ch->head_recv,
1077                                            ch->tail_recv,
1078                                            com,
1079                                            pos->prev);
1080     }
1081   }
1082 }
1083
1084
1085 /**
1086  * We got an acknowledgement for payload data for a channel.
1087  * Possibly resume transmissions.
1088  *
1089  * @param ch channel that got the ack
1090  * @param ack details about what was received
1091  */
1092 void
1093 GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
1094                                         const struct GNUNET_CADET_ChannelDataAckMessage *ack)
1095 {
1096   struct CadetReliableMessage *crm;
1097
1098   GNUNET_break (GNUNET_NO == ch->is_loopback);
1099   if (GNUNET_NO == ch->reliable)
1100   {
1101     /* not expecting ACKs on unreliable channel, odd */
1102     GNUNET_break_op (0);
1103     return;
1104   }
1105   for (crm = ch->head_sent;
1106         NULL != crm;
1107        crm = crm->next)
1108     if (ack->mid.mid == crm->data_message.mid.mid)
1109       break;
1110   if (NULL == crm)
1111   {
1112     /* ACK for message we already dropped, might have been a
1113        duplicate ACK? Ignore. */
1114     LOG (GNUNET_ERROR_TYPE_DEBUG,
1115          "Duplicate DATA_ACK on %s, ignoring\n",
1116          GCCH_2s (ch));
1117     GNUNET_STATISTICS_update (stats,
1118                               "# duplicate DATA_ACKs",
1119                               1,
1120                               GNUNET_NO);
1121     return;
1122   }
1123   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1124                                ch->tail_sent,
1125                                crm);
1126   ch->pending_messages--;
1127   GNUNET_free (crm);
1128   GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
1129   LOG (GNUNET_ERROR_TYPE_DEBUG,
1130        "Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
1131        GCCH_2s (ch),
1132        (unsigned int) ntohl (ack->mid.mid),
1133        ch->pending_messages);
1134   send_ack_to_client (ch,
1135                       (NULL == ch->owner) ? ch->dest : ch->owner);
1136 }
1137
1138
1139 /**
1140  * Destroy channel, based on the other peer closing the
1141  * connection.  Also needs to remove this channel from
1142  * the tunnel.
1143  *
1144  * @param ch channel to destroy
1145  */
1146 void
1147 GCCH_handle_remote_destroy (struct CadetChannel *ch)
1148 {
1149   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1150   LOG (GNUNET_ERROR_TYPE_DEBUG,
1151        "Received remote channel DESTROY for %s\n",
1152        GCCH_2s (ch));
1153   if (GNUNET_YES == ch->destroy)
1154   {
1155     /* Local client already gone, this is instant-death. */
1156     channel_destroy (ch);
1157     return;
1158   }
1159   if (NULL != ch->head_recv)
1160   {
1161     LOG (GNUNET_ERROR_TYPE_WARNING,
1162          "Lost end of transmission due to remote shutdown on channel %s\n",
1163          GCCH_2s (ch));
1164     /* FIXME: change API to notify client about truncated transmission! */
1165   }
1166   ch->destroy = GNUNET_YES;
1167   GSC_handle_remote_channel_destroy ((NULL != ch->owner) ? ch->owner : ch->dest,
1168                                      (NULL != ch->owner) ? ch->ccn_owner : ch->ccn_dest,
1169                                      ch);
1170   channel_destroy (ch);
1171 }
1172
1173
1174 /**
1175  * Function called once the tunnel has sent one of our messages.
1176  * If the message is unreliable, simply frees the `crm`. If the
1177  * message was reliable, calculate retransmission time and
1178  * wait for ACK (or retransmit).
1179  *
1180  * @param cls the `struct CadetReliableMessage` that was sent
1181  */
1182 static void
1183 data_sent_cb (void *cls);
1184
1185
1186 /**
1187  * We need to retry a transmission, the last one took too long to
1188  * be acknowledged.
1189  *
1190  * @param cls the `struct CadetChannel` where we need to retransmit
1191  */
1192 static void
1193 retry_transmission (void *cls)
1194 {
1195   struct CadetChannel *ch = cls;
1196   struct CadetReliableMessage *crm = ch->head_sent;
1197
1198   ch->retry_data_task = NULL;
1199   GNUNET_assert (NULL == crm->qe);
1200   crm->qe = GCT_send (ch->t,
1201                       &crm->data_message.header,
1202                       &data_sent_cb,
1203                       crm);
1204 }
1205
1206
1207 /**
1208  * Check if we can now allow the client to transmit, and if so,
1209  * let the client know about it.
1210  *
1211  * @param ch channel to check
1212  */
1213 static void
1214 GCCH_check_allow_client (struct CadetChannel *ch)
1215 {
1216   struct GNUNET_MQ_Envelope *env;
1217   struct GNUNET_CADET_LocalAck *msg;
1218
1219   if (CADET_CHANNEL_READY != ch->state)
1220   {
1221     /* destination did not yet ACK our CREATE! */
1222     LOG (GNUNET_ERROR_TYPE_DEBUG,
1223          "%s not yet ready, throttling client until ACK.\n",
1224          GCCH_2s (ch));
1225     return;
1226   }
1227   if (ch->pending_messages > ch->max_pending_messages)
1228   {
1229     /* Too many messages in queue. */
1230     LOG (GNUNET_ERROR_TYPE_DEBUG,
1231          "Message queue still too long on %s, throttling client until ACK.\n",
1232          GCCH_2s (ch));
1233     return;
1234   }
1235   if ( (NULL != ch->head_sent) &&
1236        (64 <= ntohl (ch->mid_send.mid) - ntohl (ch->head_sent->data_message.mid.mid)) )
1237   {
1238     LOG (GNUNET_ERROR_TYPE_DEBUG,
1239          "Gap in ACKs too big on %s, throttling client until ACK.\n",
1240          GCCH_2s (ch));
1241     return;
1242   }
1243
1244   LOG (GNUNET_ERROR_TYPE_DEBUG,
1245        "Sending local ack to %s client\n",
1246        GCCH_2s (ch));
1247   env = GNUNET_MQ_msg (msg,
1248                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
1249   msg->ccn = (NULL != ch->owner) ? ch->ccn_owner : ch->ccn_dest;
1250   GSC_send_to_client ((NULL != ch->owner) ? ch->owner : ch->dest,
1251                       env);
1252 }
1253
1254
1255 /**
1256  * Function called once the tunnel has sent one of our messages.
1257  * If the message is unreliable, simply frees the `crm`. If the
1258  * message was reliable, calculate retransmission time and
1259  * wait for ACK (or retransmit).
1260  *
1261  * @param cls the `struct CadetReliableMessage` that was sent
1262  */
1263 static void
1264 data_sent_cb (void *cls)
1265 {
1266   struct CadetReliableMessage *crm = cls;
1267   struct CadetChannel *ch = crm->ch;
1268   struct CadetReliableMessage *off;
1269
1270   crm->qe = NULL;
1271   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1272                                ch->tail_sent,
1273                                crm);
1274   if (GNUNET_NO == ch->reliable)
1275   {
1276     GNUNET_free (crm);
1277     ch->pending_messages--;
1278     GCCH_check_allow_client (ch);
1279     return;
1280   }
1281   if (0 == crm->retry_delay.rel_value_us)
1282     crm->retry_delay = ch->expected_delay;
1283   crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
1284
1285   /* find position for re-insertion into the DLL */
1286   if ( (NULL == ch->head_sent) ||
1287        (crm->next_retry.abs_value_us < ch->head_sent->next_retry.abs_value_us) )
1288   {
1289     /* insert at HEAD, also (re)schedule retry task! */
1290     GNUNET_CONTAINER_DLL_insert (ch->head_sent,
1291                                  ch->tail_sent,
1292                                  crm);
1293     if (NULL != ch->retry_data_task)
1294       GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1295     ch->retry_data_task
1296       = GNUNET_SCHEDULER_add_delayed (crm->retry_delay,
1297                                       &retry_transmission,
1298                                       ch);
1299     return;
1300   }
1301   for (off = ch->head_sent; NULL != off; off = off->next)
1302     if (crm->next_retry.abs_value_us < off->next_retry.abs_value_us)
1303       break;
1304   if (NULL == off)
1305   {
1306     /* insert at tail */
1307     GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent,
1308                                       ch->tail_sent,
1309                                       crm);
1310   }
1311   else
1312   {
1313     /* insert before off */
1314     GNUNET_CONTAINER_DLL_insert_after (ch->head_sent,
1315                                        ch->tail_sent,
1316                                        off->prev,
1317                                        crm);
1318   }
1319 }
1320
1321
1322 /**
1323  * Handle data given by a client.
1324  *
1325  * Check whether the client is allowed to send in this tunnel, save if
1326  * channel is reliable and send an ACK to the client if there is still
1327  * buffer space in the tunnel.
1328  *
1329  * @param ch Channel.
1330  * @param buf payload to transmit.
1331  * @param buf_len number of bytes in @a buf
1332  * @return #GNUNET_OK if everything goes well,
1333  *         #GNUNET_SYSERR in case of an error.
1334  */
1335 int
1336 GCCH_handle_local_data (struct CadetChannel *ch,
1337                         const char *buf,
1338                         size_t buf_len)
1339 {
1340   struct CadetReliableMessage *crm;
1341
1342   if (ch->pending_messages > ch->max_pending_messages)
1343   {
1344     GNUNET_break (0);
1345     return GNUNET_SYSERR;
1346   }
1347   ch->pending_messages++;
1348
1349   if (GNUNET_YES == ch->is_loopback)
1350   {
1351     GNUNET_break (0); // fIXME: not implemented
1352     return GNUNET_SYSERR;
1353   }
1354
1355   /* Everything is correct, send the message. */
1356   crm = GNUNET_malloc (sizeof (*crm) + buf_len);
1357   crm->ch = ch;
1358   crm->data_message.header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1359   crm->data_message.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
1360   ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
1361   crm->data_message.mid = ch->mid_send;
1362   crm->data_message.ctn = ch->ctn;
1363   GNUNET_memcpy (&crm[1],
1364                  buf,
1365                  buf_len);
1366   GNUNET_CONTAINER_DLL_insert (ch->head_sent,
1367                                ch->tail_sent,
1368                                crm);
1369   LOG (GNUNET_ERROR_TYPE_DEBUG,
1370        "Sending %u bytes from local client to %s\n",
1371        buf_len,
1372        GCCH_2s (ch));
1373   crm->qe = GCT_send (ch->t,
1374                       &crm->data_message.header,
1375                       &data_sent_cb,
1376                       crm);
1377   GCCH_check_allow_client (ch);
1378   return GNUNET_OK;
1379 }
1380
1381
1382 /**
1383  * Try to deliver messages to the local client, if it is ready for more.
1384  *
1385  * @param ch channel to process
1386  */
1387 static void
1388 send_client_buffered_data (struct CadetChannel *ch)
1389 {
1390   struct CadetOutOfOrderMessage *com;
1391
1392   if (GNUNET_NO == ch->client_ready)
1393     return; /* client not ready */
1394   com = ch->head_recv;
1395   if (NULL == com)
1396     return; /* none pending */
1397   if ( (com->mid.mid != ch->mid_recv.mid) &&
1398        (GNUNET_NO == ch->out_of_order) )
1399     return; /* missing next one in-order */
1400
1401   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1402               "Passing payload message to client on %s\n",
1403               GCCH_2s (ch));
1404
1405   /* all good, pass next message to client */
1406   GNUNET_CONTAINER_DLL_remove (ch->head_recv,
1407                                ch->tail_recv,
1408                                com);
1409   /* FIXME: if unreliable, this is not aggressive
1410      enough, as it would be OK to have lost some! */
1411   ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
1412   ch->mid_futures >>= 1; /* equivalent to division by 2 */
1413   GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
1414                       com->env);
1415   GNUNET_free (com);
1416   if ( (0xFFULL == (ch->mid_futures & 0xFFULL)) &&
1417        (GNUNET_YES == ch->reliable) )
1418   {
1419     /* The next 15 messages were also already received (0xFF), this
1420        suggests that the sender may be blocked on flow control
1421        urgently waiting for an ACK from us. (As we have an inherent
1422        maximum of 64 bits, and 15 is getting too close for comfort.)
1423        So we should send one now. */
1424     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1425                 "Sender on %s likely blocked on flow-control, sending ACK now.\n",
1426                 GCCH_2s (ch));
1427     if (GNUNET_YES == ch->reliable)
1428       send_channel_data_ack (ch);
1429   }
1430
1431   if (NULL != ch->head_recv)
1432     return;
1433   if (GNUNET_NO == ch->destroy)
1434     return;
1435   GCT_send_channel_destroy (ch->t,
1436                             ch->ctn);
1437   channel_destroy (ch);
1438 }
1439
1440
1441 /**
1442  * Handle ACK from client on local channel.
1443  *
1444  * @param ch channel to destroy
1445  */
1446 void
1447 GCCH_handle_local_ack (struct CadetChannel *ch)
1448 {
1449   ch->client_ready = GNUNET_YES;
1450   send_client_buffered_data (ch);
1451 }
1452
1453
1454 #define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-chn",__VA_ARGS__)
1455
1456
1457 /**
1458  * Log channel info.
1459  *
1460  * @param ch Channel.
1461  * @param level Debug level to use.
1462  */
1463 void
1464 GCCH_debug (struct CadetChannel *ch,
1465             enum GNUNET_ErrorType level)
1466 {
1467   int do_log;
1468
1469   do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
1470                                        "cadet-chn",
1471                                        __FILE__, __FUNCTION__, __LINE__);
1472   if (0 == do_log)
1473     return;
1474
1475   if (NULL == ch)
1476   {
1477     LOG2 (level, "CHN *** DEBUG NULL CHANNEL ***\n");
1478     return;
1479   }
1480   LOG2 (level,
1481         "CHN %s:%X (%p)\n",
1482         GCT_2s (ch->t),
1483         ch->ctn,
1484         ch);
1485   if (NULL != ch->owner)
1486   {
1487     LOG2 (level,
1488           "CHN origin %s ready %s local-id: %u\n",
1489           GSC_2s (ch->owner),
1490           ch->client_ready ? "YES" : "NO",
1491           ntohl (ch->ccn_owner.channel_of_client));
1492   }
1493   if (NULL != ch->dest)
1494   {
1495     LOG2 (level,
1496           "CHN destination %s ready %s local-id: %u\n",
1497           GSC_2s (ch->dest),
1498           ch->client_ready ? "YES" : "NO",
1499           ntohl (ch->ccn_dest.channel_of_client));
1500   }
1501   LOG2 (level,
1502         "CHN  Message IDs recv: %d (%LLX), send: %d\n",
1503         ntohl (ch->mid_recv.mid),
1504         (unsigned long long) ch->mid_futures,
1505         ntohl (ch->mid_send.mid));
1506 }
1507
1508
1509
1510 /* end of gnunet-service-cadet-new_channel.c */