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_YES == ch->is_loopback)
 341                    ? "loopback"
 342                    : GNUNET_i2s (GCP_get_id (GCT_get_destination (ch->t))),
 343                    GNUNET_h2s (&ch->port),
 344                    ch->ctn,
 345                    ntohl (ch->ccn_owner.channel_of_client),
 346                    ntohl (ch->ccn_dest.channel_of_client));
 347   return buf;
 348 }
 349
 350
 351 /**
 352  * Get the channel's public ID.
 353  *
 354  * @param ch Channel.
 355  *
 356  * @return ID used to identify the channel with the remote peer.
 357  */
 358 struct GNUNET_CADET_ChannelTunnelNumber
 359 GCCH_get_id (const struct CadetChannel *ch)
 360 {
 361   return ch->ctn;
 362 }
 363
 364
 365 /**
 366  * Destroy the given channel.
 367  *
 368  * @param ch channel to destroy
 369  */
 370 static void
 371 channel_destroy (struct CadetChannel *ch)
 372 {
 373   struct CadetReliableMessage *crm;
 374   struct CadetOutOfOrderMessage *com;
 375
 376   while (NULL != (crm = ch->head_sent))
 377   {
 378     GNUNET_assert (ch == crm->ch);
 379     if (NULL != crm->qe)
 380     {
 381       GCT_send_cancel (crm->qe);
 382       crm->qe = NULL;
 383     }
 384     GNUNET_CONTAINER_DLL_remove (ch->head_sent,
 385                                  ch->tail_sent,
 386                                  crm);
 387     GNUNET_free (crm);
 388   }
 389   while (NULL != (com = ch->head_recv))
 390   {
 391     GNUNET_CONTAINER_DLL_remove (ch->head_recv,
 392                                  ch->tail_recv,
 393                                  com);
 394     GNUNET_MQ_discard (com->env);
 395     GNUNET_free (com);
 396   }
 397   if (NULL != ch->last_control_qe)
 398   {
 399     GCT_send_cancel (ch->last_control_qe);
 400     ch->last_control_qe = NULL;
 401   }
 402   if (NULL != ch->retry_data_task)
 403   {
 404     GNUNET_SCHEDULER_cancel (ch->retry_data_task);
 405     ch->retry_data_task = NULL;
 406   }
 407   if (NULL != ch->retry_control_task)
 408   {
 409     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 410     ch->retry_control_task = NULL;
 411   }
 412   if (GNUNET_NO == ch->is_loopback)
 413   {
 414     GCT_remove_channel (ch->t,
 415                         ch,
 416                         ch->ctn);
 417     ch->t = NULL;
 418   }
 419   GNUNET_free (ch);
 420 }
 421
 422
 423 /**
 424  * Send a channel create message.
 425  *
 426  * @param cls Channel for which to send.
 427  */
 428 static void
 429 send_channel_open (void *cls);
 430
 431
 432 /**
 433  * Function called once the tunnel confirms that we sent the
 434  * create message.  Delays for a bit until we retry.
 435  *
 436  * @param cls our `struct CadetChannel`.
 437  */
 438 static void
 439 channel_open_sent_cb (void *cls)
 440 {
 441   struct CadetChannel *ch = cls;
 442
 443   GNUNET_assert (NULL != ch->last_control_qe);
 444   ch->last_control_qe = NULL;
 445   ch->retry_time = GNUNET_TIME_STD_BACKOFF (ch->retry_time);
 446   LOG (GNUNET_ERROR_TYPE_DEBUG,
 447        "Sent CHANNEL_OPEN on %s, retrying in %s\n",
 448        GCCH_2s (ch),
 449        GNUNET_STRINGS_relative_time_to_string (ch->retry_time,
 450                                                GNUNET_YES));
 451   ch->retry_control_task
 452     = GNUNET_SCHEDULER_add_delayed (ch->retry_time,
 453                                     &send_channel_open,
 454                                     ch);
 455 }
 456
 457
 458 /**
 459  * Send a channel open message.
 460  *
 461  * @param cls Channel for which to send.
 462  */
 463 static void
 464 send_channel_open (void *cls)
 465 {
 466   struct CadetChannel *ch = cls;
 467   struct GNUNET_CADET_ChannelOpenMessage msgcc;
 468   uint32_t options;
 469
 470   ch->retry_control_task = NULL;
 471   LOG (GNUNET_ERROR_TYPE_DEBUG,
 472        "Sending CHANNEL_OPEN message for %s\n",
 473        GCCH_2s (ch));
 474   options = 0;
 475   if (ch->nobuffer)
 476     options |= GNUNET_CADET_OPTION_NOBUFFER;
 477   if (ch->reliable)
 478     options |= GNUNET_CADET_OPTION_RELIABLE;
 479   if (ch->out_of_order)
 480     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
 481   msgcc.header.size = htons (sizeof (msgcc));
 482   msgcc.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN);
 483   msgcc.opt = htonl (options);
 484   msgcc.port = ch->port;
 485   msgcc.ctn = ch->ctn;
 486   ch->state = CADET_CHANNEL_OPEN_SENT;
 487   ch->last_control_qe = GCT_send (ch->t,
 488                                   &msgcc.header,
 489                                   &channel_open_sent_cb,
 490                                   ch);
 491 }
 492
 493
 494 /**
 495  * Function called once and only once after a channel was bound
 496  * to its tunnel via #GCT_add_channel() is ready for transmission.
 497  * Note that this is only the case for channels that this peer
 498  * initiates, as for incoming channels we assume that they are
 499  * ready for transmission immediately upon receiving the open
 500  * message.  Used to bootstrap the #GCT_send() process.
 501  *
 502  * @param ch the channel for which the tunnel is now ready
 503  */
 504 void
 505 GCCH_tunnel_up (struct CadetChannel *ch)
 506 {
 507   GNUNET_assert (NULL == ch->retry_control_task);
 508   LOG (GNUNET_ERROR_TYPE_DEBUG,
 509        "Tunnel up, sending CHANNEL_OPEN on %s now\n",
 510        GCCH_2s (ch));
 511   ch->retry_control_task
 512     = GNUNET_SCHEDULER_add_now (&send_channel_open,
 513                                 ch);
 514 }
 515
 516
 517 /**
 518  * Create a new channel.
 519  *
 520  * @param owner local client owning the channel
 521  * @param ccn local number of this channel at the @a owner
 522  * @param destination peer to which we should build the channel
 523  * @param port desired port at @a destination
 524  * @param options options for the channel
 525  * @return handle to the new channel
 526  */
 527 struct CadetChannel *
 528 GCCH_channel_local_new (struct CadetClient *owner,
 529                         struct GNUNET_CADET_ClientChannelNumber ccn,
 530                         struct CadetPeer *destination,
 531                         const struct GNUNET_HashCode *port,
 532                         uint32_t options)
 533 {
 534   struct CadetChannel *ch;
 535
 536   ch = GNUNET_new (struct CadetChannel);
 537   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
 538   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
 539   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
 540   ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
 541   ch->owner = owner;
 542   ch->ccn_owner = ccn;
 543   ch->port = *port;
 544   if (0 == memcmp (&my_full_id,
 545                    GCP_get_id (destination),
 546                    sizeof (struct GNUNET_PeerIdentity)))
 547   {
 548     ch->is_loopback = GNUNET_YES;
 549     ch->dest = GNUNET_CONTAINER_multihashmap_get (open_ports,
 550                                                   port);
 551     if (NULL == ch->dest)
 552     {
 553       /* port closed, wait for it to possibly open */
 554       (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
 555                                                 port,
 556                                                 ch,
 557                                                 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 558       LOG (GNUNET_ERROR_TYPE_DEBUG,
 559            "Created loose incoming loopback channel to port %s\n",
 560            GNUNET_h2s (&ch->port));
 561     }
 562     else
 563     {
 564       GCCH_bind (ch,
 565                  ch->dest);
 566     }
 567   }
 568   else
 569   {
 570     ch->t = GCP_get_tunnel (destination,
 571                             GNUNET_YES);
 572     ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
 573     ch->ctn = GCT_add_channel (ch->t,
 574                                ch);
 575   }
 576   GNUNET_STATISTICS_update (stats,
 577                             "# channels",
 578                             1,
 579                             GNUNET_NO);
 580   LOG (GNUNET_ERROR_TYPE_DEBUG,
 581        "Created channel to port %s at peer %s for %s using %s\n",
 582        GNUNET_h2s (port),
 583        GCP_2s (destination),
 584        GSC_2s (owner),
 585        (GNUNET_YES == ch->is_loopback) ? "loopback" : GCT_2s (ch->t));
 586   return ch;
 587 }
 588
 589
 590 /**
 591  * We had an incoming channel to a port that is closed.
 592  * It has not been opened for a while, drop it.
 593  *
 594  * @param cls the channel to drop
 595  */
 596 static void
 597 timeout_closed_cb (void *cls)
 598 {
 599   struct CadetChannel *ch = cls;
 600
 601   ch->retry_control_task = NULL;
 602   LOG (GNUNET_ERROR_TYPE_DEBUG,
 603        "Closing incoming channel to port %s from peer %s due to timeout\n",
 604        GNUNET_h2s (&ch->port),
 605        GCP_2s (GCT_get_destination (ch->t)));
 606   channel_destroy (ch);
 607 }
 608
 609
 610 /**
 611  * Create a new channel based on a request coming in over the network.
 612  *
 613  * @param t tunnel to the remote peer
 614  * @param ctn identifier of this channel in the tunnel
 615  * @param port desired local port
 616  * @param options options for the channel
 617  * @return handle to the new channel
 618  */
 619 struct CadetChannel *
 620 GCCH_channel_incoming_new (struct CadetTunnel *t,
 621                            struct GNUNET_CADET_ChannelTunnelNumber ctn,
 622                            const struct GNUNET_HashCode *port,
 623                            uint32_t options)
 624 {
 625   struct CadetChannel *ch;
 626   struct CadetClient *c;
 627
 628   ch = GNUNET_new (struct CadetChannel);
 629   ch->port = *port;
 630   ch->t = t;
 631   ch->ctn = ctn;
 632   ch->retry_time = CADET_INITIAL_RETRANSMIT_TIME;
 633   ch->nobuffer = (0 != (options & GNUNET_CADET_OPTION_NOBUFFER));
 634   ch->reliable = (0 != (options & GNUNET_CADET_OPTION_RELIABLE));
 635   ch->out_of_order = (0 != (options & GNUNET_CADET_OPTION_OUT_OF_ORDER));
 636   ch->max_pending_messages = (ch->nobuffer) ? 1 : 4; /* FIXME: 4!? Do not hardcode! */
 637   GNUNET_STATISTICS_update (stats,
 638                             "# channels",
 639                             1,
 640                             GNUNET_NO);
 641
 642   c = GNUNET_CONTAINER_multihashmap_get (open_ports,
 643                                          port);
 644   if (NULL == c)
 645   {
 646     /* port closed, wait for it to possibly open */
 647     (void) GNUNET_CONTAINER_multihashmap_put (loose_channels,
 648                                               port,
 649                                               ch,
 650                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 651     ch->retry_control_task
 652       = GNUNET_SCHEDULER_add_delayed (TIMEOUT_CLOSED_PORT,
 653                                       &timeout_closed_cb,
 654                                       ch);
 655     LOG (GNUNET_ERROR_TYPE_DEBUG,
 656          "Created loose incoming channel to port %s from peer %s\n",
 657          GNUNET_h2s (&ch->port),
 658          GCP_2s (GCT_get_destination (ch->t)));
 659   }
 660   else
 661   {
 662     GCCH_bind (ch,
 663                c);
 664   }
 665   GNUNET_STATISTICS_update (stats,
 666                             "# channels",
 667                             1,
 668                             GNUNET_NO);
 669   return ch;
 670 }
 671
 672
 673 /**
 674  * Function called once the tunnel confirms that we sent the
 675  * ACK message.  Just remembers it was sent, we do not expect
 676  * ACKs for ACKs ;-).
 677  *
 678  * @param cls our `struct CadetChannel`.
 679  */
 680 static void
 681 send_ack_cb (void *cls)
 682 {
 683   struct CadetChannel *ch = cls;
 684
 685   GNUNET_assert (NULL != ch->last_control_qe);
 686   ch->last_control_qe = NULL;
 687 }
 688
 689
 690 /**
 691  * Compute and send the current #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK to the other peer.
 692  *
 693  * @param ch channel to send the #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK for
 694  */
 695 static void
 696 send_channel_data_ack (struct CadetChannel *ch)
 697 {
 698   struct GNUNET_CADET_ChannelDataAckMessage msg;
 699
 700   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA_ACK);
 701   msg.header.size = htons (sizeof (msg));
 702   msg.ctn = ch->ctn;
 703   msg.mid.mid = htonl (ntohl (ch->mid_recv.mid) - 1);
 704   msg.futures = GNUNET_htonll (ch->mid_futures);
 705   if (NULL != ch->last_control_qe)
 706     GCT_send_cancel (ch->last_control_qe);
 707   ch->last_control_qe = GCT_send (ch->t,
 708                                   &msg.header,
 709                                   &send_ack_cb,
 710                                   ch);
 711 }
 712
 713
 714 /**
 715  * Send our initial #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK to the client confirming that the
 716  * connection is up.
 717  *
 718  * @param cls the `struct CadetChannel`
 719  */
 720 static void
 721 send_open_ack (void *cls)
 722 {
 723   struct CadetChannel *ch = cls;
 724   struct GNUNET_CADET_ChannelManageMessage msg;
 725
 726   LOG (GNUNET_ERROR_TYPE_DEBUG,
 727        "Sending CHANNEL_OPEN_ACK on %s\n",
 728        GCCH_2s (ch));
 729   ch->retry_control_task = NULL;
 730   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK);
 731   msg.header.size = htons (sizeof (msg));
 732   msg.reserved = htonl (0);
 733   msg.ctn = ch->ctn;
 734   if (NULL != ch->last_control_qe)
 735     GCT_send_cancel (ch->last_control_qe);
 736   ch->last_control_qe = GCT_send (ch->t,
 737                                   &msg.header,
 738                                   &send_ack_cb,
 739                                   ch);
 740 }
 741
 742
 743 /**
 744  * We got a #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN message again for
 745  * this channel.  If the binding was successful, (re)transmit the
 746  * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
 747  *
 748  * @param ch channel that got the duplicate open
 749  */
 750 void
 751 GCCH_handle_duplicate_open (struct CadetChannel *ch)
 752 {
 753   if (NULL == ch->dest)
 754   {
 755     LOG (GNUNET_ERROR_TYPE_DEBUG,
 756          "Ignoring duplicate channel OPEN on %s: port is closed\n",
 757          GCCH_2s (ch));
 758     return;
 759   }
 760   if (NULL != ch->retry_control_task)
 761   {
 762     LOG (GNUNET_ERROR_TYPE_DEBUG,
 763          "Ignoring duplicate channel OPEN on %s: control message is pending\n",
 764          GCCH_2s (ch));
 765     return;
 766   }
 767   LOG (GNUNET_ERROR_TYPE_DEBUG,
 768        "Retransmitting OPEN_ACK on %s\n",
 769        GCCH_2s (ch));
 770   ch->retry_control_task
 771     = GNUNET_SCHEDULER_add_now (&send_open_ack,
 772                                 ch);
 773 }
 774
 775
 776 /**
 777  * Send a #GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK to the client to solicit more messages.
 778  *
 779  * @param ch channel the ack is for
 780  * @param to_owner #GNUNET_YES to send to owner,
 781  *                 #GNUNET_NO to send to dest
 782  */
 783 static void
 784 send_ack_to_client (struct CadetChannel *ch,
 785                     int to_owner)
 786 {
 787   struct GNUNET_MQ_Envelope *env;
 788   struct GNUNET_CADET_LocalAck *ack;
 789   struct CadetClient *c;
 790
 791   env = GNUNET_MQ_msg (ack,
 792                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
 793   ack->ccn = (GNUNET_YES == to_owner) ? ch->ccn_owner : ch->ccn_dest;
 794   c = (GNUNET_YES == to_owner)
 795     ? ch->owner
 796     : ch->dest;
 797   LOG (GNUNET_ERROR_TYPE_DEBUG,
 798        "Sending CADET_LOCAL_ACK to %s (%s) at ccn %X\n",
 799        GSC_2s (c),
 800        (GNUNET_YES == to_owner) ? "owner" : "dest",
 801        ntohl (ack->ccn.channel_of_client));
 802   GSC_send_to_client (c,
 803                       env);
 804 }
 805
 806
 807 /**
 808  * A client is bound to the port that we have a channel
 809  * open to.  Send the acknowledgement for the connection
 810  * request and establish the link with the client.
 811  *
 812  * @param ch open incoming channel
 813  * @param c client listening on the respective port
 814  */
 815 void
 816 GCCH_bind (struct CadetChannel *ch,
 817            struct CadetClient *c)
 818 {
 819   uint32_t options;
 820
 821   LOG (GNUNET_ERROR_TYPE_DEBUG,
 822        "Binding %s from %s to port %s of %s\n",
 823        GCCH_2s (ch),
 824        GCT_2s (ch->t),
 825        GNUNET_h2s (&ch->port),
 826        GSC_2s (c));
 827   if (NULL != ch->retry_control_task)
 828   {
 829     /* there might be a timeout task here */
 830     GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 831     ch->retry_control_task = NULL;
 832   }
 833   options = 0;
 834   if (ch->nobuffer)
 835     options |= GNUNET_CADET_OPTION_NOBUFFER;
 836   if (ch->reliable)
 837     options |= GNUNET_CADET_OPTION_RELIABLE;
 838   if (ch->out_of_order)
 839     options |= GNUNET_CADET_OPTION_OUT_OF_ORDER;
 840   ch->dest = c;
 841   ch->ccn_dest = GSC_bind (c,
 842                            ch,
 843                            (GNUNET_YES == ch->is_loopback)
 844                            ? GCP_get (&my_full_id,
 845                                       GNUNET_YES)
 846                            : GCT_get_destination (ch->t),
 847                            &ch->port,
 848                            options);
 849   GNUNET_assert (ntohl (ch->ccn_dest.channel_of_client) <
 850                  GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
 851   ch->mid_recv.mid = htonl (1); /* The CONNECT counts as message 0! */
 852   if (GNUNET_YES == ch->is_loopback)
 853   {
 854     ch->state = CADET_CHANNEL_OPEN_SENT;
 855     GCCH_handle_channel_open_ack (ch);
 856   }
 857   else
 858   {
 859     /* notify other peer that we accepted the connection */
 860     ch->retry_control_task
 861       = GNUNET_SCHEDULER_add_now (&send_open_ack,
 862                                   ch);
 863   }
 864   /* give client it's initial supply of ACKs */
 865   GNUNET_assert (ntohl (ch->ccn_dest.channel_of_client) <
 866                  GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
 867   for (unsigned int i=0;i<ch->max_pending_messages;i++)
 868     send_ack_to_client (ch,
 869                         GNUNET_NO);
 870 }
 871
 872
 873 /**
 874  * Destroy locally created channel.  Called by the local client, so no
 875  * need to tell the client.
 876  *
 877  * @param ch channel to destroy
 878  * @param c client that caused the destruction
 879  */
 880 void
 881 GCCH_channel_local_destroy (struct CadetChannel *ch,
 882                             struct CadetClient *c)
 883 {
 884   LOG (GNUNET_ERROR_TYPE_DEBUG,
 885        "%s asks for destruction of %s\n",
 886        GSC_2s (c),
 887        GCCH_2s (ch));
 888   GNUNET_assert (NULL != c);
 889   if (c == ch->owner)
 890     ch->owner = NULL;
 891   else if (c == ch->dest)
 892     ch->dest = NULL;
 893   else
 894     GNUNET_assert (0);
 895   if (GNUNET_YES == ch->destroy)
 896   {
 897     /* other end already destroyed, with the local client gone, no need
 898        to finish transmissions, just destroy immediately. */
 899     channel_destroy (ch);
 900     return;
 901   }
 902   if ( (NULL != ch->head_sent) ||
 903        (NULL != ch->owner) ||
 904        (NULL != ch->dest) )
 905   {
 906     /* Wait for other end to destroy us as well,
 907        and otherwise allow send queue to be transmitted first */
 908     ch->destroy = GNUNET_YES;
 909     return;
 910   }
 911   /* If the we ever sent the CHANNEL_CREATE, we need to send a destroy message. */
 912   if (CADET_CHANNEL_NEW != ch->state)
 913     GCT_send_channel_destroy (ch->t,
 914                               ch->ctn);
 915   /* Nothing left to do, just finish destruction */
 916   channel_destroy (ch);
 917 }
 918
 919
 920 /**
 921  * We got an acknowledgement for the creation of the channel
 922  * (the port is open on the other side). Begin transmissions.
 923  *
 924  * @param ch channel to destroy
 925  */
 926 void
 927 GCCH_handle_channel_open_ack (struct CadetChannel *ch)
 928 {
 929   switch (ch->state)
 930   {
 931   case CADET_CHANNEL_NEW:
 932     /* this should be impossible */
 933     GNUNET_break (0);
 934     break;
 935   case CADET_CHANNEL_OPEN_SENT:
 936     if (NULL == ch->owner)
 937     {
 938       /* We're not the owner, wrong direction! */
 939       GNUNET_break_op (0);
 940       return;
 941     }
 942     LOG (GNUNET_ERROR_TYPE_DEBUG,
 943          "Received CHANNEL_OPEN_ACK for waiting %s, entering READY state\n",
 944          GCCH_2s (ch));
 945     if (NULL != ch->retry_control_task) /* can be NULL if ch->is_loopback */
 946     {
 947       GNUNET_SCHEDULER_cancel (ch->retry_control_task);
 948       ch->retry_control_task = NULL;
 949     }
 950     ch->state = CADET_CHANNEL_READY;
 951     /* On first connect, send client as many ACKs as we allow messages
 952        to be buffered! */
 953     for (unsigned int i=0;i<ch->max_pending_messages;i++)
 954       send_ack_to_client (ch,
 955                           GNUNET_YES);
 956     break;
 957   case CADET_CHANNEL_READY:
 958     /* duplicate ACK, maybe we retried the CREATE. Ignore. */
 959     LOG (GNUNET_ERROR_TYPE_DEBUG,
 960          "Received duplicate channel OPEN_ACK for %s\n",
 961          GCCH_2s (ch));
 962     GNUNET_STATISTICS_update (stats,
 963                               "# duplicate CREATE_ACKs",
 964                               1,
 965                               GNUNET_NO);
 966     break;
 967   }
 968 }
 969
 970
 971 /**
 972  * Test if element @a e1 comes before element @a e2.
 973  *
 974  * TODO: use opportunity to create generic list insertion sort
 975  * logic in container!
 976  *
 977  * @param cls closure, our `struct CadetChannel`
 978  * @param e1 an element of to sort
 979  * @param e2 another element to sort
 980  * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
 981  */
 982 static int
 983 is_before (void *cls,
 984            void *e1,
 985            void *e2)
 986 {
 987   struct CadetOutOfOrderMessage *m1 = e1;
 988   struct CadetOutOfOrderMessage *m2 = e2;
 989   uint32_t v1 = ntohl (m1->mid.mid);
 990   uint32_t v2 = ntohl (m2->mid.mid);
 991   uint32_t delta;
 992
 993   delta = v1 - v2;
 994   if (delta > (uint32_t) INT_MAX)
 995   {
 996     /* in overflow range, we can safely assume we wrapped around */
 997     return GNUNET_NO;
 998   }
 999   else
1000   {
1001     return GNUNET_YES;
1002   }
1003 }
1004
1005
1006 /**
1007  * We got payload data for a channel.  Pass it on to the client
1008  * and send an ACK to the other end (once flow control allows it!)
1009  *
1010  * @param ch channel that got data
1011  */
1012 void
1013 GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
1014                                     const struct GNUNET_CADET_ChannelAppDataMessage *msg)
1015 {
1016   struct GNUNET_MQ_Envelope *env;
1017   struct GNUNET_CADET_LocalData *ld;
1018   struct CadetOutOfOrderMessage *com;
1019   size_t payload_size;
1020
1021   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1022   payload_size = ntohs (msg->header.size) - sizeof (*msg);
1023   env = GNUNET_MQ_msg_extra (ld,
1024                              payload_size,
1025                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1026   ld->ccn = (NULL == ch->dest) ? ch->ccn_owner : ch->ccn_dest;
1027   GNUNET_memcpy (&ld[1],
1028                  &msg[1],
1029                  payload_size);
1030   if ( (GNUNET_YES == ch->client_ready) &&
1031        ( (GNUNET_YES == ch->out_of_order) ||
1032          (msg->mid.mid == ch->mid_recv.mid) ) )
1033   {
1034     LOG (GNUNET_ERROR_TYPE_DEBUG,
1035          "Giving %u bytes of payload from %s to client %s\n",
1036          (unsigned int) payload_size,
1037          GCCH_2s (ch),
1038          GSC_2s (ch->owner ? ch->owner : ch->dest));
1039     GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
1040                         env);
1041     ch->mid_recv.mid = htonl (1 + ntohl (ch->mid_recv.mid));
1042     ch->mid_futures >>= 1;
1043   }
1044   else
1045   {
1046     /* FIXME-SECURITY: if the element is WAY too far ahead,
1047        drop it (can't buffer too much!) */
1048     LOG (GNUNET_ERROR_TYPE_DEBUG,
1049          "Queuing %s payload of %u bytes on %s (mid %u, need %u first)\n",
1050          (GNUNET_YES == ch->client_ready)
1051          ? "out-of-order"
1052          : "client-not-ready",
1053          (unsigned int) payload_size,
1054          GCCH_2s (ch),
1055          ntohl (msg->mid.mid),
1056          ntohl (ch->mid_recv.mid));
1057
1058     com = GNUNET_new (struct CadetOutOfOrderMessage);
1059     com->mid = msg->mid;
1060     com->env = env;
1061     /* sort into list ordered by "is_before" */
1062     if ( (NULL == ch->head_recv) ||
1063          (GNUNET_YES == is_before (ch,
1064                                    com,
1065                                    ch->head_recv)) )
1066     {
1067       GNUNET_CONTAINER_DLL_insert (ch->head_recv,
1068                                    ch->tail_recv,
1069                                    com);
1070     }
1071     else
1072     {
1073       struct CadetOutOfOrderMessage *pos;
1074
1075       for (pos = ch->head_recv;
1076            NULL != pos;
1077            pos = pos->next)
1078       {
1079         if (GNUNET_YES !=
1080             is_before (ch,
1081                        pos,
1082                        com))
1083           break;
1084       }
1085       if (NULL == pos)
1086         GNUNET_CONTAINER_DLL_insert_tail (ch->head_recv,
1087                                           ch->tail_recv,
1088                                           com);
1089       else
1090         GNUNET_CONTAINER_DLL_insert_after (ch->head_recv,
1091                                            ch->tail_recv,
1092                                            com,
1093                                            pos->prev);
1094     }
1095   }
1096 }
1097
1098
1099 /**
1100  * We got an acknowledgement for payload data for a channel.
1101  * Possibly resume transmissions.
1102  *
1103  * @param ch channel that got the ack
1104  * @param ack details about what was received
1105  */
1106 void
1107 GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
1108                                         const struct GNUNET_CADET_ChannelDataAckMessage *ack)
1109 {
1110   struct CadetReliableMessage *crm;
1111
1112   GNUNET_break (GNUNET_NO == ch->is_loopback);
1113   if (GNUNET_NO == ch->reliable)
1114   {
1115     /* not expecting ACKs on unreliable channel, odd */
1116     GNUNET_break_op (0);
1117     return;
1118   }
1119   for (crm = ch->head_sent;
1120         NULL != crm;
1121        crm = crm->next)
1122     if (ack->mid.mid == crm->data_message.mid.mid)
1123       break;
1124   if (NULL == crm)
1125   {
1126     /* ACK for message we already dropped, might have been a
1127        duplicate ACK? Ignore. */
1128     LOG (GNUNET_ERROR_TYPE_DEBUG,
1129          "Duplicate DATA_ACK on %s, ignoring\n",
1130          GCCH_2s (ch));
1131     GNUNET_STATISTICS_update (stats,
1132                               "# duplicate DATA_ACKs",
1133                               1,
1134                               GNUNET_NO);
1135     return;
1136   }
1137   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1138                                ch->tail_sent,
1139                                crm);
1140   ch->pending_messages--;
1141   send_ack_to_client (ch,
1142                       (NULL == ch->owner)
1143                       ? GNUNET_NO
1144                       : GNUNET_YES);
1145   GNUNET_free (crm);
1146   GNUNET_assert (ch->pending_messages < ch->max_pending_messages);
1147   LOG (GNUNET_ERROR_TYPE_DEBUG,
1148        "Received DATA_ACK on %s for message %u (%u ACKs pending)\n",
1149        GCCH_2s (ch),
1150        (unsigned int) ntohl (ack->mid.mid),
1151        ch->pending_messages);
1152   send_ack_to_client (ch,
1153                       (NULL == ch->owner)
1154                       ? GNUNET_NO
1155                       : GNUNET_YES);
1156 }
1157
1158
1159 /**
1160  * Destroy channel, based on the other peer closing the
1161  * connection.  Also needs to remove this channel from
1162  * the tunnel.
1163  *
1164  * @param ch channel to destroy
1165  */
1166 void
1167 GCCH_handle_remote_destroy (struct CadetChannel *ch)
1168 {
1169   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1170   LOG (GNUNET_ERROR_TYPE_DEBUG,
1171        "Received remote channel DESTROY for %s\n",
1172        GCCH_2s (ch));
1173   if (GNUNET_YES == ch->destroy)
1174   {
1175     /* Local client already gone, this is instant-death. */
1176     channel_destroy (ch);
1177     return;
1178   }
1179   if (NULL != ch->head_recv)
1180   {
1181     LOG (GNUNET_ERROR_TYPE_WARNING,
1182          "Lost end of transmission due to remote shutdown on %s\n",
1183          GCCH_2s (ch));
1184     /* FIXME: change API to notify client about truncated transmission! */
1185   }
1186   ch->destroy = GNUNET_YES;
1187   GSC_handle_remote_channel_destroy ((NULL != ch->owner) ? ch->owner : ch->dest,
1188                                      (NULL != ch->owner) ? ch->ccn_owner : ch->ccn_dest,
1189                                      ch);
1190   channel_destroy (ch);
1191 }
1192
1193
1194 /**
1195  * Function called once the tunnel has sent one of our messages.
1196  * If the message is unreliable, simply frees the `crm`. If the
1197  * message was reliable, calculate retransmission time and
1198  * wait for ACK (or retransmit).
1199  *
1200  * @param cls the `struct CadetReliableMessage` that was sent
1201  */
1202 static void
1203 data_sent_cb (void *cls);
1204
1205
1206 /**
1207  * We need to retry a transmission, the last one took too long to
1208  * be acknowledged.
1209  *
1210  * @param cls the `struct CadetChannel` where we need to retransmit
1211  */
1212 static void
1213 retry_transmission (void *cls)
1214 {
1215   struct CadetChannel *ch = cls;
1216   struct CadetReliableMessage *crm = ch->head_sent;
1217
1218   ch->retry_data_task = NULL;
1219   GNUNET_assert (NULL == crm->qe);
1220   crm->qe = GCT_send (ch->t,
1221                       &crm->data_message.header,
1222                       &data_sent_cb,
1223                       crm);
1224 }
1225
1226
1227 /**
1228  * Function called once the tunnel has sent one of our messages.
1229  * If the message is unreliable, simply frees the `crm`. If the
1230  * message was reliable, calculate retransmission time and
1231  * wait for ACK (or retransmit).
1232  *
1233  * @param cls the `struct CadetReliableMessage` that was sent
1234  */
1235 static void
1236 data_sent_cb (void *cls)
1237 {
1238   struct CadetReliableMessage *crm = cls;
1239   struct CadetChannel *ch = crm->ch;
1240   struct CadetReliableMessage *off;
1241
1242   GNUNET_assert (GNUNET_NO == ch->is_loopback);
1243   crm->qe = NULL;
1244   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
1245                                ch->tail_sent,
1246                                crm);
1247   if (GNUNET_NO == ch->reliable)
1248   {
1249     GNUNET_free (crm);
1250     ch->pending_messages--;
1251     send_ack_to_client (ch,
1252                         (NULL == ch->owner)
1253                         ? GNUNET_NO
1254                         : GNUNET_YES);
1255     return;
1256   }
1257   if (0 == crm->retry_delay.rel_value_us)
1258     crm->retry_delay = ch->expected_delay;
1259   crm->next_retry = GNUNET_TIME_relative_to_absolute (crm->retry_delay);
1260
1261   /* find position for re-insertion into the DLL */
1262   if ( (NULL == ch->head_sent) ||
1263        (crm->next_retry.abs_value_us < ch->head_sent->next_retry.abs_value_us) )
1264   {
1265     /* insert at HEAD, also (re)schedule retry task! */
1266     GNUNET_CONTAINER_DLL_insert (ch->head_sent,
1267                                  ch->tail_sent,
1268                                  crm);
1269     if (NULL != ch->retry_data_task)
1270       GNUNET_SCHEDULER_cancel (ch->retry_data_task);
1271     ch->retry_data_task
1272       = GNUNET_SCHEDULER_add_delayed (crm->retry_delay,
1273                                       &retry_transmission,
1274                                       ch);
1275     return;
1276   }
1277   for (off = ch->head_sent; NULL != off; off = off->next)
1278     if (crm->next_retry.abs_value_us < off->next_retry.abs_value_us)
1279       break;
1280   if (NULL == off)
1281   {
1282     /* insert at tail */
1283     GNUNET_CONTAINER_DLL_insert_tail (ch->head_sent,
1284                                       ch->tail_sent,
1285                                       crm);
1286   }
1287   else
1288   {
1289     /* insert before off */
1290     GNUNET_CONTAINER_DLL_insert_after (ch->head_sent,
1291                                        ch->tail_sent,
1292                                        off->prev,
1293                                        crm);
1294   }
1295 }
1296
1297
1298 /**
1299  * Handle data given by a client.
1300  *
1301  * Check whether the client is allowed to send in this tunnel, save if
1302  * channel is reliable and send an ACK to the client if there is still
1303  * buffer space in the tunnel.
1304  *
1305  * @param ch Channel.
1306  * @param sender_ccn ccn of the sender
1307  * @param buf payload to transmit.
1308  * @param buf_len number of bytes in @a buf
1309  * @return #GNUNET_OK if everything goes well,
1310  *         #GNUNET_SYSERR in case of an error.
1311  */
1312 int
1313 GCCH_handle_local_data (struct CadetChannel *ch,
1314                         struct GNUNET_CADET_ClientChannelNumber sender_ccn,
1315                         const char *buf,
1316                         size_t buf_len)
1317 {
1318   struct CadetReliableMessage *crm;
1319
1320   if (ch->pending_messages > ch->max_pending_messages)
1321   {
1322     GNUNET_break (0);
1323     return GNUNET_SYSERR;
1324   }
1325   ch->pending_messages++;
1326
1327   if (GNUNET_YES == ch->is_loopback)
1328   {
1329     struct CadetClient *receiver;
1330     struct GNUNET_MQ_Envelope *env;
1331     struct GNUNET_CADET_LocalData *ld;
1332     int to_owner;
1333
1334     env = GNUNET_MQ_msg_extra (ld,
1335                                buf_len,
1336                                GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
1337     if (sender_ccn.channel_of_client ==
1338         ch->ccn_owner.channel_of_client)
1339     {
1340       receiver = ch->dest;
1341       ld->ccn = ch->ccn_dest;
1342       to_owner = GNUNET_NO;
1343     }
1344     else
1345     {
1346       GNUNET_assert (sender_ccn.channel_of_client ==
1347                      ch->ccn_dest.channel_of_client);
1348       receiver = ch->owner;
1349       ld->ccn = ch->ccn_owner;
1350       to_owner = GNUNET_YES;
1351     }
1352     GNUNET_memcpy (&ld[1],
1353                    buf,
1354                    buf_len);
1355     /* FIXME: this does not provide for flow control! */
1356     GSC_send_to_client (receiver,
1357                         env);
1358     send_ack_to_client (ch,
1359                         to_owner);
1360     return GNUNET_OK;
1361   }
1362
1363   /* Everything is correct, send the message. */
1364   crm = GNUNET_malloc (sizeof (*crm) + buf_len);
1365   crm->ch = ch;
1366   crm->data_message.header.size = htons (sizeof (struct GNUNET_CADET_ChannelAppDataMessage) + buf_len);
1367   crm->data_message.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CHANNEL_APP_DATA);
1368   ch->mid_send.mid = htonl (ntohl (ch->mid_send.mid) + 1);
1369   crm->data_message.mid = ch->mid_send;
1370   crm->data_message.ctn = ch->ctn;
1371   GNUNET_memcpy (&crm[1],
1372                  buf,
1373                  buf_len);
1374   GNUNET_CONTAINER_DLL_insert (ch->head_sent,
1375                                ch->tail_sent,
1376                                crm);
1377   LOG (GNUNET_ERROR_TYPE_DEBUG,
1378        "Sending %u bytes from local client to %s\n",
1379        buf_len,
1380        GCCH_2s (ch));
1381   crm->qe = GCT_send (ch->t,
1382                       &crm->data_message.header,
1383                       &data_sent_cb,
1384                       crm);
1385   return GNUNET_OK;
1386 }
1387
1388
1389 /**
1390  * Try to deliver messages to the local client, if it is ready for more.
1391  *
1392  * @param ch channel to process
1393  */
1394 static void
1395 send_client_buffered_data (struct CadetChannel *ch)
1396 {
1397   struct CadetOutOfOrderMessage *com;
1398
1399   if (GNUNET_NO == ch->client_ready)
1400     return; /* client not ready */
1401   com = ch->head_recv;
1402   if (NULL == com)
1403     return; /* none pending */
1404   if ( (com->mid.mid != ch->mid_recv.mid) &&
1405        (GNUNET_NO == ch->out_of_order) )
1406     return; /* missing next one in-order */
1407
1408   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1409               "Passing payload message to client on %s\n",
1410               GCCH_2s (ch));
1411
1412   /* all good, pass next message to client */
1413   GNUNET_CONTAINER_DLL_remove (ch->head_recv,
1414                                ch->tail_recv,
1415                                com);
1416   /* FIXME: if unreliable, this is not aggressive
1417      enough, as it would be OK to have lost some! */
1418   ch->mid_recv.mid = htonl (1 + ntohl (com->mid.mid));
1419   ch->mid_futures >>= 1; /* equivalent to division by 2 */
1420   GSC_send_to_client (ch->owner ? ch->owner : ch->dest,
1421                       com->env);
1422   GNUNET_free (com);
1423   if ( (0xFFULL == (ch->mid_futures & 0xFFULL)) &&
1424        (GNUNET_YES == ch->reliable) )
1425   {
1426     /* The next 15 messages were also already received (0xFF), this
1427        suggests that the sender may be blocked on flow control
1428        urgently waiting for an ACK from us. (As we have an inherent
1429        maximum of 64 bits, and 15 is getting too close for comfort.)
1430        So we should send one now. */
1431     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1432                 "Sender on %s likely blocked on flow-control, sending ACK now.\n",
1433                 GCCH_2s (ch));
1434     if (GNUNET_YES == ch->reliable)
1435       send_channel_data_ack (ch);
1436   }
1437
1438   if (NULL != ch->head_recv)
1439     return;
1440   if (GNUNET_NO == ch->destroy)
1441     return;
1442   GCT_send_channel_destroy (ch->t,
1443                             ch->ctn);
1444   channel_destroy (ch);
1445 }
1446
1447
1448 /**
1449  * Handle ACK from client on local channel.
1450  *
1451  * @param ch channel to destroy
1452  * @param client_ccn ccn of the client sending the ack
1453  */
1454 void
1455 GCCH_handle_local_ack (struct CadetChannel *ch,
1456                        struct GNUNET_CADET_ClientChannelNumber client_ccn)
1457 {
1458   ch->client_ready = GNUNET_YES;
1459   LOG (GNUNET_ERROR_TYPE_DEBUG,
1460        "Got LOCAL_ACK, client ready to receive more data!\n");
1461   send_client_buffered_data (ch);
1462 }
1463
1464
1465 #define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-chn",__VA_ARGS__)
1466
1467
1468 /**
1469  * Log channel info.
1470  *
1471  * @param ch Channel.
1472  * @param level Debug level to use.
1473  */
1474 void
1475 GCCH_debug (struct CadetChannel *ch,
1476             enum GNUNET_ErrorType level)
1477 {
1478   int do_log;
1479
1480   do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
1481                                        "cadet-chn",
1482                                        __FILE__, __FUNCTION__, __LINE__);
1483   if (0 == do_log)
1484     return;
1485
1486   if (NULL == ch)
1487   {
1488     LOG2 (level, "CHN *** DEBUG NULL CHANNEL ***\n");
1489     return;
1490   }
1491   LOG2 (level,
1492         "CHN %s:%X (%p)\n",
1493         GCT_2s (ch->t),
1494         ch->ctn,
1495         ch);
1496   if (NULL != ch->owner)
1497   {
1498     LOG2 (level,
1499           "CHN origin %s ready %s local-id: %u\n",
1500           GSC_2s (ch->owner),
1501           ch->client_ready ? "YES" : "NO",
1502           ntohl (ch->ccn_owner.channel_of_client));
1503   }
1504   if (NULL != ch->dest)
1505   {
1506     LOG2 (level,
1507           "CHN destination %s ready %s local-id: %u\n",
1508           GSC_2s (ch->dest),
1509           ch->client_ready ? "YES" : "NO",
1510           ntohl (ch->ccn_dest.channel_of_client));
1511   }
1512   LOG2 (level,
1513         "CHN  Message IDs recv: %d (%LLX), send: %d\n",
1514         ntohl (ch->mid_recv.mid),
1515         (unsigned long long) ch->mid_futures,
1516         ntohl (ch->mid_send.mid));
1517 }
1518
1519
1520
1521 /* end of gnunet-service-cadet-new_channel.c */