src/cadet/gnunet-service-cadet_connection.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2001-2015 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20 /**
  21  * @file cadet/gnunet-service-cadet_connection.c
  22  * @brief GNUnet CADET service connection handling
  23  * @author Bartlomiej Polot
  24  */
  25 #include "platform.h"
  26 #include "gnunet_util_lib.h"
  27 #include "gnunet_statistics_service.h"
  28 #include "cadet_path.h"
  29 #include "cadet_protocol.h"
  30 #include "cadet.h"
  31 #include "gnunet-service-cadet_connection.h"
  32 #include "gnunet-service-cadet_peer.h"
  33 #include "gnunet-service-cadet_tunnel.h"
  34
  35
  36 #define LOG(level, ...) GNUNET_log_from (level,"cadet-con",__VA_ARGS__)
  37 #define LOG2(level, ...) GNUNET_log_from_nocheck(level,"cadet-con",__VA_ARGS__)
  38
  39
  40 #define CADET_MAX_POLL_TIME      GNUNET_TIME_relative_multiply (\
  41                                   GNUNET_TIME_UNIT_MINUTES,\
  42                                   10)
  43 #define AVG_MSGS                32
  44
  45
  46 /******************************************************************************/
  47 /********************************   STRUCTS  **********************************/
  48 /******************************************************************************/
  49
  50 /**
  51  * Struct to encapsulate all the Flow Control information to a peer to which
  52  * we are directly connected (on a core level).
  53  */
  54 struct CadetFlowControl
  55 {
  56   /**
  57    * Connection this controls.
  58    */
  59   struct CadetConnection *c;
  60
  61   /**
  62    * How many messages are in the queue on this connection.
  63    */
  64   unsigned int queue_n;
  65
  66   /**
  67    * How many messages do we accept in the queue.
  68    */
  69   unsigned int queue_max;
  70
  71   /**
  72    * ID of the last packet sent towards the peer.
  73    */
  74   uint32_t last_pid_sent;
  75
  76   /**
  77    * ID of the last packet received from the peer.
  78    */
  79   uint32_t last_pid_recv;
  80
  81   /**
  82    * Bitmap of past 32 messages received:
  83    * - LSB being @c last_pid_recv.
  84    * - MSB being @c last_pid_recv - 31 (mod UINTMAX).
  85    */
  86   uint32_t recv_bitmap;
  87
  88   /**
  89    * Last ACK sent to the peer (peer can't send more than this PID).
  90    */
  91   uint32_t last_ack_sent;
  92
  93   /**
  94    * Last ACK sent towards the origin (for traffic towards leaf node).
  95    */
  96   uint32_t last_ack_recv;
  97
  98   /**
  99    * Task to poll the peer in case of a lost ACK causes stall.
 100    */
 101   struct GNUNET_SCHEDULER_Task *poll_task;
 102
 103   /**
 104    * How frequently to poll for ACKs.
 105    */
 106   struct GNUNET_TIME_Relative poll_time;
 107
 108   /**
 109    * Queued poll message, to cancel if not necessary anymore (got ACK).
 110    */
 111   struct CadetConnectionQueue *poll_msg;
 112
 113   /**
 114    * Queued poll message, to cancel if not necessary anymore (got ACK).
 115    */
 116   struct CadetConnectionQueue *ack_msg;
 117 };
 118
 119 /**
 120  * Keep a record of the last messages sent on this connection.
 121  */
 122 struct CadetConnectionPerformance
 123 {
 124   /**
 125    * Circular buffer for storing measurements.
 126    */
 127   double usecsperbyte[AVG_MSGS];
 128
 129   /**
 130    * Running average of @c usecsperbyte.
 131    */
 132   double avg;
 133
 134   /**
 135    * How many values of @c usecsperbyte are valid.
 136    */
 137   uint16_t size;
 138
 139   /**
 140    * Index of the next "free" position in @c usecsperbyte.
 141    */
 142   uint16_t idx;
 143 };
 144
 145
 146 /**
 147  * Struct containing all information regarding a connection to a peer.
 148  */
 149 struct CadetConnection
 150 {
 151   /**
 152    * Tunnel this connection is part of.
 153    */
 154   struct CadetTunnel *t;
 155
 156   /**
 157    * Flow control information for traffic fwd.
 158    */
 159   struct CadetFlowControl fwd_fc;
 160
 161   /**
 162    * Flow control information for traffic bck.
 163    */
 164   struct CadetFlowControl bck_fc;
 165
 166   /**
 167    * Measure connection performance on the endpoint.
 168    */
 169   struct CadetConnectionPerformance *perf;
 170
 171   /**
 172    * ID of the connection.
 173    */
 174   struct GNUNET_CADET_Hash id;
 175
 176   /**
 177    * Path being used for the tunnel. At the origin of the connection
 178    * it's a pointer to the destination's path pool, otherwise just a copy.
 179    */
 180   struct CadetPeerPath *path;
 181
 182   /**
 183    * Task to keep the used paths alive at the owner,
 184    * time tunnel out on all the other peers.
 185    */
 186   struct GNUNET_SCHEDULER_Task *fwd_maintenance_task;
 187
 188   /**
 189    * Task to keep the used paths alive at the destination,
 190    * time tunnel out on all the other peers.
 191    */
 192   struct GNUNET_SCHEDULER_Task *bck_maintenance_task;
 193
 194   /**
 195    * Queue handle for maintainance traffic. One handle for FWD and BCK since
 196    * one peer never needs to maintain both directions (no loopback connections).
 197    */
 198   struct CadetPeerQueue *maintenance_q;
 199
 200   /**
 201    * Should equal #get_next_hop().
 202    */
 203   struct CadetPeer *next_peer;
 204
 205   /**
 206    * Should equal #get_prev_hop().
 207    */
 208   struct CadetPeer *prev_peer;
 209
 210   /**
 211    * State of the connection.
 212    */
 213   enum CadetConnectionState state;
 214
 215   /**
 216    * Position of the local peer in the path.
 217    */
 218   unsigned int own_pos;
 219
 220   /**
 221    * Pending message count.
 222    */
 223   unsigned int pending_messages;
 224
 225   /**
 226    * Destroy flag:
 227    * - if 0, connection in use.
 228    * - if 1, destroy on last message.
 229    * - if 2, connection is being destroyed don't re-enter.
 230    */
 231   int destroy;
 232
 233   /**
 234    * Counter to do exponential backoff when creating a connection (max 64).
 235    */
 236   unsigned short create_retry;
 237 };
 238
 239
 240 /**
 241  * Handle for messages queued but not yet sent.
 242  */
 243 struct CadetConnectionQueue
 244 {
 245   /**
 246    * Peer queue handle, to cancel if necessary.
 247    */
 248   struct CadetPeerQueue *q;
 249
 250   /**
 251    * Continuation to call once sent.
 252    */
 253   GCC_sent cont;
 254
 255   /**
 256    * Closure for @e cont.
 257    */
 258   void *cont_cls;
 259
 260   /**
 261    * Was this a forced message? (Do not account for it)
 262    */
 263   int forced;
 264 };
 265
 266 /******************************************************************************/
 267 /*******************************   GLOBALS  ***********************************/
 268 /******************************************************************************/
 269
 270 /**
 271  * Global handle to the statistics service.
 272  */
 273 extern struct GNUNET_STATISTICS_Handle *stats;
 274
 275 /**
 276  * Local peer own ID (memory efficient handle).
 277  */
 278 extern GNUNET_PEER_Id myid;
 279
 280 /**
 281  * Local peer own ID (full value).
 282  */
 283 extern struct GNUNET_PeerIdentity my_full_id;
 284
 285 /**
 286  * Connections known, indexed by cid (CadetConnection).
 287  */
 288 static struct GNUNET_CONTAINER_MultiHashMap *connections;
 289
 290 /**
 291  * How many connections are we willing to maintain.
 292  * Local connections are always allowed, even if there are more connections than max.
 293  */
 294 static unsigned long long max_connections;
 295
 296 /**
 297  * How many messages *in total* are we willing to queue, divide by number of
 298  * connections to get connection queue size.
 299  */
 300 static unsigned long long max_msgs_queue;
 301
 302 /**
 303  * How often to send path keepalives. Paths timeout after 4 missed.
 304  */
 305 static struct GNUNET_TIME_Relative refresh_connection_time;
 306
 307 /**
 308  * How often to send path create / ACKs.
 309  */
 310 static struct GNUNET_TIME_Relative create_connection_time;
 311
 312
 313 /******************************************************************************/
 314 /********************************   STATIC  ***********************************/
 315 /******************************************************************************/
 316
 317
 318
 319 #if 0 // avoid compiler warning for unused static function
 320 static void
 321 fc_debug (struct CadetFlowControl *fc)
 322 {
 323   LOG (GNUNET_ERROR_TYPE_DEBUG, "    IN: %u/%u\n",
 324               fc->last_pid_recv, fc->last_ack_sent);
 325   LOG (GNUNET_ERROR_TYPE_DEBUG, "    OUT: %u/%u\n",
 326               fc->last_pid_sent, fc->last_ack_recv);
 327   LOG (GNUNET_ERROR_TYPE_DEBUG, "    QUEUE: %u/%u\n",
 328               fc->queue_n, fc->queue_max);
 329 }
 330
 331 static void
 332 connection_debug (struct CadetConnection *c)
 333 {
 334   if (NULL == c)
 335   {
 336     LOG (GNUNET_ERROR_TYPE_INFO, "DEBUG NULL CONNECTION\n");
 337     return;
 338   }
 339   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s:%X\n",
 340               peer2s (c->t->peer), GCC_2s (c));
 341   LOG (GNUNET_ERROR_TYPE_DEBUG, "  state: %u, pending msgs: %u\n",
 342               c->state, c->pending_messages);
 343   LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD FC\n");
 344   fc_debug (&c->fwd_fc);
 345   LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK FC\n");
 346   fc_debug (&c->bck_fc);
 347 }
 348 #endif
 349
 350
 351 /**
 352  * Schedule next keepalive task, taking in consideration
 353  * the connection state and number of retries.
 354  *
 355  * @param c Connection for which to schedule the next keepalive.
 356  * @param fwd Direction for the next keepalive.
 357  */
 358 static void
 359 schedule_next_keepalive (struct CadetConnection *c, int fwd);
 360
 361
 362 /**
 363  * Resets the connection timeout task, some other message has done the
 364  * task's job.
 365  * - For the first peer on the direction this means to send
 366  *   a keepalive or a path confirmation message (either create or ACK).
 367  * - For all other peers, this means to destroy the connection,
 368  *   due to lack of activity.
 369  * Starts the timeout if no timeout was running (connection just created).
 370  *
 371  * @param c Connection whose timeout to reset.
 372  * @param fwd Is this forward?
 373  */
 374 static void
 375 connection_reset_timeout (struct CadetConnection *c, int fwd);
 376
 377
 378 /**
 379  * Get string description for tunnel state. Reentrant.
 380  *
 381  * @param s Tunnel state.
 382  *
 383  * @return String representation.
 384  */
 385 static const char *
 386 GCC_state2s (enum CadetConnectionState s)
 387 {
 388   switch (s)
 389   {
 390     case CADET_CONNECTION_NEW:
 391       return "CADET_CONNECTION_NEW";
 392     case CADET_CONNECTION_SENT:
 393       return "CADET_CONNECTION_SENT";
 394     case CADET_CONNECTION_ACK:
 395       return "CADET_CONNECTION_ACK";
 396     case CADET_CONNECTION_READY:
 397       return "CADET_CONNECTION_READY";
 398     case CADET_CONNECTION_DESTROYED:
 399       return "CADET_CONNECTION_DESTROYED";
 400     case CADET_CONNECTION_BROKEN:
 401       return "CADET_CONNECTION_BROKEN";
 402     default:
 403       GNUNET_break (0);
 404       LOG (GNUNET_ERROR_TYPE_ERROR, " conn state %u unknown!\n", s);
 405       return "CADET_CONNECTION_STATE_ERROR";
 406   }
 407 }
 408
 409
 410 /**
 411  * Initialize a Flow Control structure to the initial state.
 412  *
 413  * @param fc Flow Control structure to initialize.
 414  */
 415 static void
 416 fc_init (struct CadetFlowControl *fc)
 417 {
 418   fc->last_pid_sent = (uint32_t) -1; /* Next (expected) = 0 */
 419   fc->last_pid_recv = (uint32_t) -1;
 420   fc->last_ack_sent = (uint32_t) 0;
 421   fc->last_ack_recv = (uint32_t) 0;
 422   fc->poll_task = NULL;
 423   fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
 424   fc->queue_n = 0;
 425   fc->queue_max = (max_msgs_queue / max_connections) + 1;
 426 }
 427
 428
 429 /**
 430  * Find a connection.
 431  *
 432  * @param cid Connection ID.
 433  */
 434 static struct CadetConnection *
 435 connection_get (const struct GNUNET_CADET_Hash *cid)
 436 {
 437   return GNUNET_CONTAINER_multihashmap_get (connections, GC_h2hc (cid));
 438 }
 439
 440
 441 static void
 442 connection_change_state (struct CadetConnection* c,
 443                          enum CadetConnectionState state)
 444 {
 445   LOG (GNUNET_ERROR_TYPE_DEBUG,
 446        "Connection %s state %s -> %s\n",
 447        GCC_2s (c), GCC_state2s (c->state), GCC_state2s (state));
 448   if (CADET_CONNECTION_DESTROYED <= c->state) /* Destroyed or broken. */
 449   {
 450     LOG (GNUNET_ERROR_TYPE_DEBUG, "state not changing anymore\n");
 451     return;
 452   }
 453   c->state = state;
 454   if (CADET_CONNECTION_READY == state)
 455     c->create_retry = 1;
 456 }
 457
 458
 459 /**
 460  * Callback called when a queued ACK message is sent.
 461  *
 462  * @param cls Closure (FC).
 463  * @param c Connection this message was on.
 464  * @param q Queue handler this call invalidates.
 465  * @param type Type of message sent.
 466  * @param fwd Was this a FWD going message?
 467  * @param size Size of the message.
 468  */
 469 static void
 470 ack_sent (void *cls,
 471           struct CadetConnection *c,
 472           struct CadetConnectionQueue *q,
 473           uint16_t type, int fwd, size_t size)
 474 {
 475   struct CadetFlowControl *fc = cls;
 476
 477   fc->ack_msg = NULL;
 478 }
 479
 480
 481 /**
 482  * Send an ACK on the connection, informing the predecessor about
 483  * the available buffer space. Should not be called in case the peer
 484  * is origin (no predecessor) in the @c fwd direction.
 485  *
 486  * Note that for fwd ack, the FWD mean forward *traffic* (root->dest),
 487  * the ACK itself goes "back" (dest->root).
 488  *
 489  * @param c Connection on which to send the ACK.
 490  * @param buffer How much space free to advertise?
 491  * @param fwd Is this FWD ACK? (Going dest -> root)
 492  * @param force Don't optimize out.
 493  */
 494 static void
 495 send_ack (struct CadetConnection *c, unsigned int buffer, int fwd, int force)
 496 {
 497   struct CadetFlowControl *next_fc;
 498   struct CadetFlowControl *prev_fc;
 499   struct GNUNET_CADET_ACK msg;
 500   uint32_t ack;
 501   int delta;
 502
 503   /* If origin, there is no connection to send ACKs. Wrong function! */
 504   GCC_check_connections ();
 505   if (GCC_is_origin (c, fwd))
 506   {
 507     LOG (GNUNET_ERROR_TYPE_DEBUG, "connection %s is origin in %s\n",
 508          GCC_2s (c), GC_f2s (fwd));
 509     GNUNET_break (0);
 510     return;
 511   }
 512
 513   next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
 514   prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
 515
 516   LOG (GNUNET_ERROR_TYPE_DEBUG, "connection send %s ack on %s\n",
 517        GC_f2s (fwd), GCC_2s (c));
 518
 519   /* Check if we need to transmit the ACK. */
 520   delta = prev_fc->last_ack_sent - prev_fc->last_pid_recv;
 521   if (3 < delta && buffer < delta && GNUNET_NO == force)
 522   {
 523     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer > 3\n");
 524     LOG (GNUNET_ERROR_TYPE_DEBUG,
 525          "  last pid recv: %u, last ack sent: %u\n",
 526          prev_fc->last_pid_recv, prev_fc->last_ack_sent);
 527     GCC_check_connections ();
 528     return;
 529   }
 530
 531   /* Ok, ACK might be necessary, what PID to ACK? */
 532   ack = prev_fc->last_pid_recv + buffer;
 533   LOG (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
 534   LOG (GNUNET_ERROR_TYPE_DEBUG,
 535        " last pid %u, last ack %u, qmax %u, q %u\n",
 536        prev_fc->last_pid_recv, prev_fc->last_ack_sent,
 537        next_fc->queue_max, next_fc->queue_n);
 538   if (ack == prev_fc->last_ack_sent && GNUNET_NO == force)
 539   {
 540     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
 541     GCC_check_connections ();
 542     return;
 543   }
 544
 545   /* Check if message is already in queue */
 546   if (NULL != prev_fc->ack_msg)
 547   {
 548     if (GC_is_pid_bigger (ack, prev_fc->last_ack_sent))
 549     {
 550       LOG (GNUNET_ERROR_TYPE_DEBUG, " canceling old ACK\n");
 551       GCC_cancel (prev_fc->ack_msg);
 552       /* GCC_cancel triggers ack_sent(), which clears fc->ack_msg */
 553     }
 554     else
 555     {
 556       LOG (GNUNET_ERROR_TYPE_DEBUG, " same ACK already in queue\n");
 557       GCC_check_connections ();
 558       return;
 559     }
 560   }
 561
 562   prev_fc->last_ack_sent = ack;
 563
 564   /* Build ACK message and send on connection */
 565   msg.header.size = htons (sizeof (msg));
 566   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_ACK);
 567   msg.ack = htonl (ack);
 568   msg.cid = c->id;
 569
 570   prev_fc->ack_msg = GCC_send_prebuilt_message (&msg.header, 0, ack, c,
 571                                                 !fwd, GNUNET_YES,
 572                                                 &ack_sent, prev_fc);
 573   GNUNET_assert (NULL != prev_fc->ack_msg);
 574   GCC_check_connections ();
 575 }
 576
 577
 578 /**
 579  * Callback called when a connection queued message is sent.
 580  *
 581  * Calculates the average time and connection packet tracking.
 582  *
 583  * @param cls Closure (ConnectionQueue Handle).
 584  * @param c Connection this message was on.
 585  * @param sent Was it really sent? (Could have been canceled)
 586  * @param type Type of message sent.
 587  * @param pid Packet ID, or 0 if not applicable (create, destroy, etc).
 588  * @param fwd Was this a FWD going message?
 589  * @param size Size of the message.
 590  * @param wait Time spent waiting for core (only the time for THIS message)
 591  *
 592  * @return #GNUNET_YES if connection was destroyed, #GNUNET_NO otherwise.
 593  */
 594 static int
 595 conn_message_sent (void *cls,
 596                    struct CadetConnection *c, int sent,
 597                    uint16_t type, uint32_t pid, int fwd, size_t size,
 598                    struct GNUNET_TIME_Relative wait)
 599 {
 600   struct CadetConnectionPerformance *p;
 601   struct CadetFlowControl *fc;
 602   struct CadetConnectionQueue *q = cls;
 603   double usecsperbyte;
 604   int forced;
 605
 606   GCC_check_connections ();
 607   LOG (GNUNET_ERROR_TYPE_DEBUG, "connection message_sent\n");
 608
 609   GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
 610
 611   fc = fwd ? &c->fwd_fc : &c->bck_fc;
 612   LOG (GNUNET_ERROR_TYPE_DEBUG, " %ssent %s %s pid %u\n",
 613        sent ? "" : "not ", GC_f2s (fwd), GC_m2s (type), pid);
 614   if (NULL != q)
 615   {
 616     forced = q->forced;
 617     if (NULL != q->cont)
 618     {
 619       LOG (GNUNET_ERROR_TYPE_DEBUG, " calling cont\n");
 620       q->cont (q->cont_cls, c, q, type, fwd, size);
 621     }
 622     GNUNET_free (q);
 623   }
 624   else if (type == GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED
 625            || type == GNUNET_MESSAGE_TYPE_CADET_AX)
 626   {
 627     /* If NULL == q and ENCRYPTED == type, message must have been ch_mngmnt */
 628     forced = GNUNET_YES;
 629   }
 630   else
 631   {
 632     forced = GNUNET_NO;
 633   }
 634   if (NULL == c)
 635   {
 636     if (type != GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
 637         && type != GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY)
 638     {
 639       LOG (GNUNET_ERROR_TYPE_ERROR, "Message %s sent on NULL connection!\n",
 640            GC_m2s (type));
 641     }
 642     GCC_check_connections ();
 643     return GNUNET_NO;
 644   }
 645   LOG (GNUNET_ERROR_TYPE_DEBUG, " C_P- %p %u\n", c, c->pending_messages);
 646   c->pending_messages--;
 647   if ( (GNUNET_YES == c->destroy) &&
 648        (0 == c->pending_messages) )
 649   {
 650     LOG (GNUNET_ERROR_TYPE_DEBUG,
 651          "!  destroying connection!\n");
 652     GCC_destroy (c);
 653     GCC_check_connections ();
 654     return GNUNET_YES;
 655   }
 656   /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
 657   switch (type)
 658   {
 659     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE:
 660     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_ACK:
 661       c->maintenance_q = NULL;
 662       /* Don't trigger a keepalive for sent ACKs, only SYN and SYNACKs */
 663       if (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE == type || !fwd)
 664         schedule_next_keepalive (c, fwd);
 665       break;
 666
 667     case GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED:
 668     case GNUNET_MESSAGE_TYPE_CADET_AX:
 669       if (GNUNET_YES == sent)
 670       {
 671         GNUNET_assert (NULL != q);
 672         fc->last_pid_sent = pid;
 673         if (GC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
 674           GCC_start_poll (c, fwd);
 675         GCC_send_ack (c, fwd, GNUNET_NO);
 676         connection_reset_timeout (c, fwd);
 677       }
 678
 679       LOG (GNUNET_ERROR_TYPE_DEBUG, "!  Q_N- %p %u\n", fc, fc->queue_n);
 680       if (GNUNET_NO == forced)
 681       {
 682         fc->queue_n--;
 683         LOG (GNUNET_ERROR_TYPE_DEBUG,
 684             "!   accounting pid %u\n",
 685             fc->last_pid_sent);
 686       }
 687       else
 688       {
 689         LOG (GNUNET_ERROR_TYPE_DEBUG,
 690              "!   forced, Q_N not accounting pid %u\n",
 691              fc->last_pid_sent);
 692       }
 693       break;
 694
 695     case GNUNET_MESSAGE_TYPE_CADET_KX:
 696       if (GNUNET_YES == sent)
 697         connection_reset_timeout (c, fwd);
 698       break;
 699
 700     case GNUNET_MESSAGE_TYPE_CADET_POLL:
 701       fc->poll_msg = NULL;
 702       break;
 703
 704     case GNUNET_MESSAGE_TYPE_CADET_ACK:
 705       fc->ack_msg = NULL;
 706       break;
 707
 708     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN:
 709       break;
 710
 711     default:
 712       LOG (GNUNET_ERROR_TYPE_ERROR, "%s unknown\n", GC_m2s (type));
 713       GNUNET_break (0);
 714       break;
 715   }
 716   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  message sent!\n");
 717
 718   if (NULL == c->perf)
 719     return GNUNET_NO; /* Only endpoints are interested in timing. */
 720
 721   p = c->perf;
 722   usecsperbyte = ((double) wait.rel_value_us) / size;
 723   if (p->size == AVG_MSGS)
 724   {
 725     /* Array is full. Substract oldest value, add new one and store. */
 726     p->avg -= (p->usecsperbyte[p->idx] / AVG_MSGS);
 727     p->usecsperbyte[p->idx] = usecsperbyte;
 728     p->avg += (p->usecsperbyte[p->idx] / AVG_MSGS);
 729   }
 730   else
 731   {
 732     /* Array not yet full. Add current value to avg and store. */
 733     p->usecsperbyte[p->idx] = usecsperbyte;
 734     p->avg *= p->size;
 735     p->avg += p->usecsperbyte[p->idx];
 736     p->size++;
 737     p->avg /= p->size;
 738   }
 739   p->idx = (p->idx + 1) % AVG_MSGS;
 740   GCC_check_connections ();
 741   return GNUNET_NO;
 742 }
 743
 744
 745 /**
 746  * Get the previous hop in a connection
 747  *
 748  * @param c Connection.
 749  *
 750  * @return Previous peer in the connection.
 751  */
 752 static struct CadetPeer *
 753 get_prev_hop (const struct CadetConnection *c)
 754 {
 755   GNUNET_PEER_Id id;
 756
 757   if (NULL == c->path)
 758     return NULL;
 759   LOG (GNUNET_ERROR_TYPE_DEBUG,
 760        " get prev hop %s [%u/%u]\n",
 761        GCC_2s (c), c->own_pos, c->path->length);
 762   if (0 == c->own_pos || c->path->length < 2)
 763     id = c->path->peers[0];
 764   else
 765     id = c->path->peers[c->own_pos - 1];
 766
 767   LOG (GNUNET_ERROR_TYPE_DEBUG, "  ID: %s (%u)\n",
 768        GNUNET_i2s (GNUNET_PEER_resolve2 (id)), id);
 769
 770   return GCP_get_short (id);
 771 }
 772
 773
 774 /**
 775  * Get the next hop in a connection
 776  *
 777  * @param c Connection.
 778  *
 779  * @return Next peer in the connection.
 780  */
 781 static struct CadetPeer *
 782 get_next_hop (const struct CadetConnection *c)
 783 {
 784   GNUNET_PEER_Id id;
 785
 786   if (NULL == c->path)
 787     return NULL;
 788
 789   LOG (GNUNET_ERROR_TYPE_DEBUG, " get next hop %s [%u/%u]\n",
 790        GCC_2s (c), c->own_pos, c->path->length);
 791   if ((c->path->length - 1) == c->own_pos || c->path->length < 2)
 792     id = c->path->peers[c->path->length - 1];
 793   else
 794     id = c->path->peers[c->own_pos + 1];
 795
 796   LOG (GNUNET_ERROR_TYPE_DEBUG, "  ID: %s (%u)\n",
 797        GNUNET_i2s (GNUNET_PEER_resolve2 (id)), id);
 798
 799   return GCP_get_short (id);
 800 }
 801
 802
 803 /**
 804  * Check that the direct neighbours (previous and next hop)
 805  * are properly associated with this connection.
 806  *
 807  * @param c connection to check
 808  */
 809 static void
 810 check_neighbours (const struct CadetConnection *c)
 811 {
 812   if (NULL == c->path)
 813     return; /* nothing to check */
 814   GCP_check_connection (get_next_hop (c), c);
 815   GCP_check_connection (get_prev_hop (c), c);
 816 }
 817
 818
 819 /**
 820  * Helper for #check_connections().  Calls #check_neighbours().
 821  *
 822  * @param cls NULL
 823  * @param key ignored
 824  * @param value the `struct CadetConnection` to check
 825  * @return #GNUNET_OK (continue to iterate)
 826  */
 827 static int
 828 check_connection (void *cls,
 829                   const struct GNUNET_HashCode *key,
 830                   void *value)
 831 {
 832   struct CadetConnection *c = value;
 833
 834   check_neighbours (c);
 835   return GNUNET_OK;
 836 }
 837
 838
 839 /**
 840  * Check invariants for all connections using #check_neighbours().
 841  */
 842 void
 843 GCC_check_connections ()
 844 {
 845   if (NULL == connections)
 846     return;
 847   GNUNET_CONTAINER_multihashmap_iterate (connections,
 848                                          &check_connection,
 849                                          NULL);
 850 }
 851
 852
 853 /**
 854  * Get the hop in a connection.
 855  *
 856  * @param c Connection.
 857  * @param fwd Next in the FWD direction?
 858  *
 859  * @return Next peer in the connection.
 860  */
 861 static struct CadetPeer *
 862 get_hop (struct CadetConnection *c, int fwd)
 863 {
 864   if (fwd)
 865     return get_next_hop (c);
 866   return get_prev_hop (c);
 867 }
 868
 869
 870 /**
 871  * Get a bit mask for a message received out-of-order.
 872  *
 873  * @param last_pid_recv Last PID we received prior to the out-of-order.
 874  * @param ooo_pid PID of the out-of-order message.
 875  */
 876 static uint32_t
 877 get_recv_bitmask (uint32_t last_pid_recv, uint32_t ooo_pid)
 878 {
 879   return 1 << (last_pid_recv - ooo_pid);
 880 }
 881
 882
 883 /**
 884  * Check is an out-of-order message is ok:
 885  * - at most 31 messages behind.
 886  * - not duplicate.
 887  *
 888  * @param last_pid_recv Last in-order PID received.
 889  */
 890 static int
 891 is_ooo_ok (uint32_t last_pid_recv, uint32_t ooo_pid, uint32_t ooo_bitmap)
 892 {
 893   uint32_t mask;
 894
 895   if (GC_is_pid_bigger (last_pid_recv - 31, ooo_pid))
 896     return GNUNET_NO;
 897
 898   mask = get_recv_bitmask (last_pid_recv, ooo_pid);
 899   if (0 != (ooo_bitmap & mask))
 900     return GNUNET_NO;
 901
 902   return GNUNET_YES;
 903 }
 904
 905
 906 /**
 907  * Is traffic coming from this sender 'FWD' traffic?
 908  *
 909  * @param c Connection to check.
 910  * @param sender Peer identity of neighbor.
 911  *
 912  * @return #GNUNET_YES in case the sender is the 'prev' hop and therefore
 913  *         the traffic is 'FWD'.
 914  *         #GNUNET_NO for BCK.
 915  *         #GNUNET_SYSERR for errors.
 916  */
 917 static int
 918 is_fwd (const struct CadetConnection *c,
 919         const struct GNUNET_PeerIdentity *sender)
 920 {
 921   GNUNET_PEER_Id id;
 922
 923   id = GNUNET_PEER_search (sender);
 924   if (GCP_get_short_id (get_prev_hop (c)) == id)
 925     return GNUNET_YES;
 926
 927   if (GCP_get_short_id (get_next_hop (c)) == id)
 928     return GNUNET_NO;
 929
 930   GNUNET_break (0);
 931   return GNUNET_SYSERR;
 932 }
 933
 934
 935 /**
 936  * Sends a CONNECTION ACK message in reponse to a received CONNECTION_CREATE
 937  * or a first CONNECTION_ACK directed to us.
 938  *
 939  * @param connection Connection to confirm.
 940  * @param fwd Should we send it FWD? (root->dest)
 941  *            (First (~SYNACK) goes BCK, second (~ACK) goes FWD)
 942  */
 943 static void
 944 send_connection_ack (struct CadetConnection *connection, int fwd)
 945 {
 946   struct CadetTunnel *t;
 947
 948   GCC_check_connections ();
 949   t = connection->t;
 950   LOG (GNUNET_ERROR_TYPE_INFO, "---> {%14s ACK} on connection %s\n",
 951        GC_f2s (!fwd), GCC_2s (connection));
 952   GCP_queue_add (get_hop (connection, fwd), NULL,
 953                  GNUNET_MESSAGE_TYPE_CADET_CONNECTION_ACK, 0, 0,
 954                  sizeof (struct GNUNET_CADET_ConnectionACK),
 955                  connection, fwd, &conn_message_sent, NULL);
 956   connection->pending_messages++;
 957   if (CADET_TUNNEL_NEW == GCT_get_cstate (t))
 958     GCT_change_cstate (t, CADET_TUNNEL_WAITING);
 959   if (CADET_CONNECTION_READY != connection->state)
 960     connection_change_state (connection, CADET_CONNECTION_SENT);
 961   GCC_check_connections ();
 962 }
 963
 964
 965 /**
 966  * Send a notification that a connection is broken.
 967  *
 968  * @param c Connection that is broken.
 969  * @param id1 Peer that has disconnected.
 970  * @param id2 Peer that has disconnected.
 971  * @param fwd Direction towards which to send it.
 972  */
 973 static void
 974 send_broken (struct CadetConnection *c,
 975              const struct GNUNET_PeerIdentity *id1,
 976              const struct GNUNET_PeerIdentity *id2,
 977              int fwd)
 978 {
 979   struct GNUNET_CADET_ConnectionBroken msg;
 980
 981   GCC_check_connections ();
 982   msg.header.size = htons (sizeof (struct GNUNET_CADET_ConnectionBroken));
 983   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 984   msg.cid = c->id;
 985   msg.peer1 = *id1;
 986   msg.peer2 = *id2;
 987   GNUNET_assert (NULL ==
 988                  GCC_send_prebuilt_message (&msg.header, 0, 0, c, fwd,
 989                                             GNUNET_YES, NULL, NULL));
 990     GCC_check_connections ();
 991 }
 992
 993
 994 /**
 995  * Send a notification that a connection is broken, when a connection
 996  * isn't even known to the local peer or soon to be destroyed.
 997  *
 998  * @param connection_id Connection ID.
 999  * @param id1 Peer that has disconnected, probably local peer.
1000  * @param id2 Peer that has disconnected can be NULL if unknown.
1001  * @param peer Peer to notify (neighbor who sent the connection).
1002  */
1003 static void
1004 send_broken_unknown (const struct GNUNET_CADET_Hash *connection_id,
1005                      const struct GNUNET_PeerIdentity *id1,
1006                      const struct GNUNET_PeerIdentity *id2,
1007                      const struct GNUNET_PeerIdentity *peer_id)
1008 {
1009   struct GNUNET_CADET_ConnectionBroken *msg;
1010   struct CadetPeer *neighbor;
1011
1012   GCC_check_connections ();
1013   LOG (GNUNET_ERROR_TYPE_INFO, "---> BROKEN on unknown connection %s\n",
1014        GNUNET_h2s (GC_h2hc (connection_id)));
1015
1016   msg = GNUNET_new (struct GNUNET_CADET_ConnectionBroken);
1017   msg->header.size = htons (sizeof (struct GNUNET_CADET_ConnectionBroken));
1018   msg->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
1019   msg->cid = *connection_id;
1020   msg->peer1 = *id1;
1021   if (NULL != id2)
1022     msg->peer2 = *id2;
1023   else
1024     memset (&msg->peer2, 0, sizeof (msg->peer2));
1025   neighbor = GCP_get (peer_id);
1026   GCP_queue_add (neighbor, msg, GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN,
1027                  0, 2, sizeof (struct GNUNET_CADET_ConnectionBroken),
1028                  NULL, GNUNET_SYSERR, /* connection, fwd */
1029                  NULL, NULL); /* continuation */
1030   GCC_check_connections ();
1031 }
1032
1033
1034 /**
1035  * Send keepalive packets for a connection.
1036  *
1037  * @param c Connection to keep alive..
1038  * @param fwd Is this a FWD keepalive? (owner -> dest).
1039  */
1040 static void
1041 send_connection_keepalive (struct CadetConnection *c, int fwd)
1042 {
1043   struct GNUNET_MessageHeader msg;
1044   struct CadetFlowControl *fc;
1045
1046   GCC_check_connections ();
1047   LOG (GNUNET_ERROR_TYPE_INFO,
1048        "keepalive %s for connection %s\n",
1049        GC_f2s (fwd), GCC_2s (c));
1050
1051   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1052   if (0 < fc->queue_n)
1053   {
1054     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, traffic in queue\n");
1055     return;
1056   }
1057
1058   GNUNET_STATISTICS_update (stats, "# keepalives sent", 1, GNUNET_NO);
1059
1060   GNUNET_assert (NULL != c->t);
1061   msg.size = htons (sizeof (msg));
1062   msg.type = htons (GNUNET_MESSAGE_TYPE_CADET_KEEPALIVE);
1063
1064   GNUNET_assert (NULL ==
1065                  GCT_send_prebuilt_message (&msg, c->t, c,
1066                                             GNUNET_NO, NULL, NULL));
1067   GCC_check_connections ();
1068 }
1069
1070
1071 /**
1072  * Send CONNECTION_{CREATE/ACK} packets for a connection.
1073  *
1074  * @param c Connection for which to send the message.
1075  * @param fwd If #GNUNET_YES, send CREATE, otherwise send ACK.
1076  */
1077 static void
1078 connection_recreate (struct CadetConnection *c, int fwd)
1079 {
1080   LOG (GNUNET_ERROR_TYPE_DEBUG,
1081        "sending connection recreate\n");
1082   if (fwd)
1083     GCC_send_create (c);
1084   else
1085     send_connection_ack (c, GNUNET_NO);
1086 }
1087
1088
1089 /**
1090  * Generic connection timer management.
1091  * Depending on the role of the peer in the connection will send the
1092  * appropriate message (build or keepalive)
1093  *
1094  * @param c Conncetion to maintain.
1095  * @param fwd Is FWD?
1096  */
1097 static void
1098 connection_maintain (struct CadetConnection *c, int fwd)
1099 {
1100   if (GNUNET_NO != c->destroy)
1101   {
1102     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, being destroyed\n");
1103     return;
1104   }
1105
1106   if (NULL == c->t)
1107   {
1108     GNUNET_break (0);
1109     GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
1110     return;
1111   }
1112
1113   if (CADET_TUNNEL_SEARCHING == GCT_get_cstate (c->t))
1114   {
1115     /* If status is SEARCHING, why is there a connection? Should be WAITING */
1116     GNUNET_break (0);
1117     GCT_debug (c->t, GNUNET_ERROR_TYPE_ERROR);
1118     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, tunnel SEARCHING\n");
1119     schedule_next_keepalive (c, fwd);
1120     return;
1121   }
1122   switch (c->state)
1123   {
1124     case CADET_CONNECTION_NEW:
1125       GNUNET_break (0);
1126       /* fall-through */
1127     case CADET_CONNECTION_SENT:
1128       connection_recreate (c, fwd);
1129       break;
1130     case CADET_CONNECTION_READY:
1131       send_connection_keepalive (c, fwd);
1132       break;
1133     default:
1134       break;
1135   }
1136 }
1137
1138
1139 /**
1140  * Keep the connection alive.
1141  *
1142  * @param c Connection to keep alive.
1143  * @param fwd Direction.
1144  * @param shutdown Are we shutting down? (Don't send traffic)
1145  *                 Non-zero value for true, not necessarily GNUNET_YES.
1146  */
1147 static void
1148 connection_keepalive (struct CadetConnection *c, int fwd, int shutdown)
1149 {
1150   GCC_check_connections ();
1151   LOG (GNUNET_ERROR_TYPE_DEBUG,
1152        "%s keepalive for %s\n",
1153        GC_f2s (fwd), GCC_2s (c));
1154
1155   if (fwd)
1156     c->fwd_maintenance_task = NULL;
1157   else
1158     c->bck_maintenance_task = NULL;
1159
1160   if (GNUNET_NO != shutdown)
1161     return;
1162
1163   connection_maintain (c, fwd);
1164   GCC_check_connections ();
1165   /* Next execution will be scheduled by message_sent or _maintain*/
1166 }
1167
1168
1169 /**
1170  * Keep the connection alive in the FWD direction.
1171  *
1172  * @param cls Closure (connection to keepalive).
1173  * @param tc TaskContext.
1174  */
1175 static void
1176 connection_fwd_keepalive (void *cls,
1177                           const struct GNUNET_SCHEDULER_TaskContext *tc)
1178 {
1179   GCC_check_connections ();
1180   connection_keepalive ((struct CadetConnection *) cls,
1181                         GNUNET_YES,
1182                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
1183   GCC_check_connections ();
1184 }
1185
1186
1187 /**
1188  * Keep the connection alive in the BCK direction.
1189  *
1190  * @param cls Closure (connection to keepalive).
1191  * @param tc TaskContext.
1192  */
1193 static void
1194 connection_bck_keepalive (void *cls,
1195                           const struct GNUNET_SCHEDULER_TaskContext *tc)
1196 {
1197   GCC_check_connections ();
1198   connection_keepalive ((struct CadetConnection *) cls,
1199                         GNUNET_NO,
1200                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
1201   GCC_check_connections ();
1202 }
1203
1204
1205 /**
1206  * Schedule next keepalive task, taking in consideration
1207  * the connection state and number of retries.
1208  *
1209  * If the peer is not the origin, do nothing.
1210  *
1211  * @param c Connection for which to schedule the next keepalive.
1212  * @param fwd Direction for the next keepalive.
1213  */
1214 static void
1215 schedule_next_keepalive (struct CadetConnection *c, int fwd)
1216 {
1217   struct GNUNET_TIME_Relative delay;
1218   struct GNUNET_SCHEDULER_Task * *task_id;
1219   GNUNET_SCHEDULER_TaskCallback keepalive_task;
1220
1221   GCC_check_connections ();
1222   if (GNUNET_NO == GCC_is_origin (c, fwd))
1223     return;
1224
1225   /* Calculate delay to use, depending on the state of the connection */
1226   if (CADET_CONNECTION_READY == c->state)
1227   {
1228     delay = refresh_connection_time;
1229   }
1230   else
1231   {
1232     if (1 > c->create_retry)
1233       c->create_retry = 1;
1234     delay = GNUNET_TIME_relative_multiply (create_connection_time,
1235                                            c->create_retry);
1236     if (c->create_retry < 64)
1237       c->create_retry *= 2;
1238   }
1239
1240   /* Select direction-dependent parameters */
1241   if (GNUNET_YES == fwd)
1242   {
1243     task_id = &c->fwd_maintenance_task;
1244     keepalive_task = &connection_fwd_keepalive;
1245   }
1246   else
1247   {
1248     task_id = &c->bck_maintenance_task;
1249     keepalive_task = &connection_bck_keepalive;
1250   }
1251
1252   /* Check that no one scheduled it before us */
1253   if (NULL != *task_id)
1254   {
1255     /* No need for a _break. It can happen for instance when sending a SYNACK
1256      * for a duplicate SYN: the first SYNACK scheduled the task. */
1257     GNUNET_SCHEDULER_cancel (*task_id);
1258   }
1259
1260   /* Schedule the task */
1261   *task_id = GNUNET_SCHEDULER_add_delayed (delay,
1262                                            keepalive_task,
1263                                            c);
1264   LOG (GNUNET_ERROR_TYPE_DEBUG, "next keepalive in %s\n",
1265        GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
1266   GCC_check_connections ();
1267 }
1268
1269
1270 /**
1271  * @brief Re-initiate traffic on this connection if necessary.
1272  *
1273  * Check if there is traffic queued towards this peer
1274  * and the core transmit handle is NULL (traffic was stalled).
1275  * If so, call core tmt rdy.
1276  *
1277  * @param c Connection on which initiate traffic.
1278  * @param fwd Is this about fwd traffic?
1279  */
1280 static void
1281 connection_unlock_queue (struct CadetConnection *c, int fwd)
1282 {
1283   struct CadetPeer *peer;
1284
1285   GCC_check_connections ();
1286   LOG (GNUNET_ERROR_TYPE_DEBUG,
1287        "connection_unlock_queue %s on %s\n",
1288        GC_f2s (fwd), GCC_2s (c));
1289
1290   if (GCC_is_terminal (c, fwd))
1291   {
1292     LOG (GNUNET_ERROR_TYPE_DEBUG, " is terminal, can unlock!\n");
1293     return;
1294   }
1295
1296   peer = get_hop (c, fwd);
1297   GCP_queue_unlock (peer, c);
1298   GCC_check_connections ();
1299 }
1300
1301
1302 /**
1303  * Cancel all transmissions that belong to a certain connection.
1304  *
1305  * If the connection is scheduled for destruction and no more messages are left,
1306  * the connection will be destroyed by the continuation call.
1307  *
1308  * @param c Connection which to cancel. Might be destroyed during this call.
1309  * @param fwd Cancel fwd traffic?
1310  */
1311 static void
1312 connection_cancel_queues (struct CadetConnection *c, int fwd)
1313 {
1314   struct CadetFlowControl *fc;
1315   struct CadetPeer *peer;
1316
1317   GCC_check_connections ();
1318   LOG (GNUNET_ERROR_TYPE_DEBUG,
1319        "Cancel %s queues for connection %s\n",
1320        GC_f2s (fwd), GCC_2s (c));
1321   if (NULL == c)
1322   {
1323     GNUNET_break (0);
1324     return;
1325   }
1326
1327   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1328   if (NULL != fc->poll_task)
1329   {
1330     GNUNET_SCHEDULER_cancel (fc->poll_task);
1331     fc->poll_task = NULL;
1332     LOG (GNUNET_ERROR_TYPE_DEBUG, "Cancel POLL in ccq for fc %p\n", fc);
1333   }
1334   peer = get_hop (c, fwd);
1335   GCP_queue_cancel (peer, c);
1336   GCC_check_connections ();
1337 }
1338
1339
1340 /**
1341  * Function called if a connection has been stalled for a while,
1342  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1343  *
1344  * @param cls Closure (poll ctx).
1345  * @param tc TaskContext.
1346  */
1347 static void
1348 connection_poll (void *cls,
1349                  const struct GNUNET_SCHEDULER_TaskContext *tc);
1350
1351
1352 /**
1353  * Callback called when a queued POLL message is sent.
1354  *
1355  * @param cls Closure (FC).
1356  * @param c Connection this message was on.
1357  * @param q Queue handler this call invalidates.
1358  * @param type Type of message sent.
1359  * @param fwd Was this a FWD going message?
1360  * @param size Size of the message.
1361  */
1362 static void
1363 poll_sent (void *cls,
1364            struct CadetConnection *c,
1365            struct CadetConnectionQueue *q,
1366            uint16_t type, int fwd, size_t size)
1367 {
1368   struct CadetFlowControl *fc = cls;
1369
1370   if (2 == c->destroy)
1371   {
1372     LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL canceled on shutdown\n");
1373     return;
1374   }
1375   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL sent for %s, scheduling new one!\n",
1376        GCC_2s (c));
1377   fc->poll_msg = NULL;
1378   fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
1379   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
1380                                                 &connection_poll,
1381                                                 fc);
1382   LOG (GNUNET_ERROR_TYPE_DEBUG, " task %u\n", fc->poll_task);
1383 }
1384
1385
1386 /**
1387  * Function called if a connection has been stalled for a while,
1388  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1389  *
1390  * @param cls Closure (poll ctx).
1391  * @param tc TaskContext.
1392  */
1393 static void
1394 connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1395 {
1396   struct CadetFlowControl *fc = cls;
1397   struct GNUNET_CADET_Poll msg;
1398   struct CadetConnection *c;
1399   int fwd;
1400
1401   fc->poll_task = NULL;
1402   GCC_check_connections ();
1403   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1404   {
1405     return;
1406   }
1407
1408   c = fc->c;
1409   fwd = fc == &c->fwd_fc;
1410   LOG (GNUNET_ERROR_TYPE_DEBUG, "Polling connection %s %s\n",
1411        GCC_2s (c),  GC_f2s (fwd));
1412
1413   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_POLL);
1414   msg.header.size = htons (sizeof (msg));
1415   msg.pid = htonl (fc->last_pid_sent);
1416   LOG (GNUNET_ERROR_TYPE_DEBUG, " last pid sent: %u\n", fc->last_pid_sent);
1417   fc->poll_msg =
1418       GCC_send_prebuilt_message (&msg.header, 0, fc->last_pid_sent, c,
1419                                  fc == &c->fwd_fc, GNUNET_YES, &poll_sent, fc);
1420   GNUNET_assert (NULL != fc->poll_msg);
1421   GCC_check_connections ();
1422 }
1423
1424
1425 /**
1426  * Resend all queued messages for a connection on other connections of the
1427  * same tunnel, if possible. The connection WILL BE DESTROYED by this function.
1428  *
1429  * @param c Connection whose messages to resend.
1430  * @param fwd Resend fwd messages?
1431  */
1432 static void
1433 resend_messages_and_destroy (struct CadetConnection *c, int fwd)
1434 {
1435   struct GNUNET_MessageHeader *out_msg;
1436   struct CadetTunnel *t = c->t;
1437   struct CadetPeer *neighbor;
1438   unsigned int pending;
1439   int destroyed;
1440
1441   GCC_check_connections ();
1442   c->state = CADET_CONNECTION_DESTROYED;
1443   c->destroy = GNUNET_YES;
1444
1445   destroyed = GNUNET_NO;
1446   neighbor = get_hop (c, fwd);
1447   pending = c->pending_messages;
1448
1449   while (NULL != (out_msg = GCP_connection_pop (neighbor, c, &destroyed)))
1450   {
1451     if (NULL != t)
1452       GCT_resend_message (out_msg, t);
1453     GNUNET_free (out_msg);
1454   }
1455
1456   /* All pending messages should have been popped,
1457    * and the connection destroyed by the continuation.
1458    */
1459   if (GNUNET_YES != destroyed)
1460   {
1461     if (0 != pending)
1462     {
1463       GNUNET_break (0);
1464       GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
1465       if (NULL != t) GCT_debug (t, GNUNET_ERROR_TYPE_ERROR);
1466     }
1467     GCC_destroy (c);
1468   }
1469   GCC_check_connections ();
1470 }
1471
1472
1473 /**
1474  * Generic connection timeout implementation.
1475  *
1476  * Timeout function due to lack of keepalive/traffic from an endpoint.
1477  * Destroys connection if called.
1478  *
1479  * @param c Connection to destroy.
1480  * @param fwd Was the timeout from the origin? (FWD timeout)
1481  */
1482 static void
1483 connection_timeout (struct CadetConnection *c, int fwd)
1484 {
1485   struct CadetFlowControl *reverse_fc;
1486
1487   GCC_check_connections ();
1488   reverse_fc = fwd ? &c->bck_fc : &c->fwd_fc;
1489
1490   LOG (GNUNET_ERROR_TYPE_INFO,
1491        "Connection %s %s timed out. Destroying.\n",
1492        GCC_2s (c),
1493        GC_f2s (fwd));
1494   GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
1495
1496   if (GCC_is_origin (c, fwd)) /* Loopback? Something is wrong! */
1497   {
1498     GNUNET_break (0);
1499     return;
1500   }
1501
1502   /* If dest, salvage queued traffic. */
1503   if (GCC_is_origin (c, !fwd))
1504   {
1505     const struct GNUNET_PeerIdentity *next_hop;
1506
1507     next_hop = GCP_get_id (fwd ? get_prev_hop (c) : get_next_hop (c));
1508     send_broken_unknown (&c->id, &my_full_id, NULL, next_hop);
1509     if (0 < reverse_fc->queue_n)
1510       resend_messages_and_destroy (c, !fwd);
1511     GCC_check_connections ();
1512     return;
1513   }
1514
1515   GCC_destroy (c);
1516   GCC_check_connections ();
1517 }
1518
1519
1520 /**
1521  * Timeout function due to lack of keepalive/traffic from the owner.
1522  * Destroys connection if called.
1523  *
1524  * @param cls Closure (connection to destroy).
1525  * @param tc TaskContext.
1526  */
1527 static void
1528 connection_fwd_timeout (void *cls,
1529                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1530 {
1531   struct CadetConnection *c = cls;
1532
1533   c->fwd_maintenance_task = NULL;
1534   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1535     return;
1536   GCC_check_connections ();
1537   connection_timeout (c, GNUNET_YES);
1538   GCC_check_connections ();
1539 }
1540
1541
1542 /**
1543  * Timeout function due to lack of keepalive/traffic from the destination.
1544  * Destroys connection if called.
1545  *
1546  * @param cls Closure (connection to destroy).
1547  * @param tc TaskContext
1548  */
1549 static void
1550 connection_bck_timeout (void *cls,
1551                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1552 {
1553   struct CadetConnection *c = cls;
1554
1555   c->bck_maintenance_task = NULL;
1556   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1557     return;
1558   GCC_check_connections ();
1559   connection_timeout (c, GNUNET_NO);
1560   GCC_check_connections ();
1561 }
1562
1563
1564 /**
1565  * Resets the connection timeout task, some other message has done the
1566  * task's job.
1567  * - For the first peer on the direction this means to send
1568  *   a keepalive or a path confirmation message (either create or ACK).
1569  * - For all other peers, this means to destroy the connection,
1570  *   due to lack of activity.
1571  * Starts the timeout if no timeout was running (connection just created).
1572  *
1573  * @param c Connection whose timeout to reset.
1574  * @param fwd Is this forward?
1575  *
1576  * TODO use heap to improve efficiency of scheduler.
1577  */
1578 static void
1579 connection_reset_timeout (struct CadetConnection *c, int fwd)
1580 {
1581   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s reset timeout\n", GC_f2s (fwd));
1582   if (GCC_is_origin (c, fwd)) /* Startpoint */
1583   {
1584     schedule_next_keepalive (c, fwd);
1585   }
1586   else /* Relay, endpoint. */
1587   {
1588     struct GNUNET_TIME_Relative delay;
1589     struct GNUNET_SCHEDULER_Task * *ti;
1590     GNUNET_SCHEDULER_TaskCallback f;
1591
1592     ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
1593
1594     if (NULL != *ti)
1595       GNUNET_SCHEDULER_cancel (*ti);
1596     delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
1597     LOG (GNUNET_ERROR_TYPE_DEBUG, "  timing out in %s\n",
1598          GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_NO));
1599     f = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
1600     *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
1601   }
1602 }
1603
1604
1605 /**
1606  * Add the connection to the list of both neighbors.
1607  *
1608  * @param c Connection.
1609  *
1610  * @return #GNUNET_OK if everything went fine
1611  *         #GNUNET_SYSERR if the was an error and @c c is malformed.
1612  */
1613 static int
1614 register_neighbors (struct CadetConnection *c)
1615 {
1616   c->next_peer = get_next_hop (c);
1617   c->prev_peer = get_prev_hop (c);
1618   GNUNET_assert (c->next_peer != c->prev_peer);
1619   LOG (GNUNET_ERROR_TYPE_DEBUG,
1620        "register neighbors for connection %s\n",
1621        GCC_2s (c));
1622   path_debug (c->path);
1623   LOG (GNUNET_ERROR_TYPE_DEBUG,
1624        "own pos %u\n", c->own_pos);
1625   LOG (GNUNET_ERROR_TYPE_DEBUG,
1626        "putting connection %s to next peer %p\n",
1627        GCC_2s (c),
1628        c->next_peer);
1629   LOG (GNUNET_ERROR_TYPE_DEBUG, "next peer %p %s\n",
1630        c->next_peer,
1631        GCP_2s (c->next_peer));
1632   LOG (GNUNET_ERROR_TYPE_DEBUG,
1633        "putting connection %s to prev peer %p\n",
1634        GCC_2s (c),
1635        c->prev_peer);
1636   LOG (GNUNET_ERROR_TYPE_DEBUG,
1637        "prev peer %p %s\n",
1638        c->prev_peer,
1639        GCP_2s (c->prev_peer));
1640
1641   if ( (GNUNET_NO == GCP_is_neighbor (c->next_peer)) ||
1642        (GNUNET_NO == GCP_is_neighbor (c->prev_peer)) )
1643   {
1644     if (GCC_is_origin (c, GNUNET_YES))
1645       GNUNET_STATISTICS_update (stats, "# local bad paths", 1, GNUNET_NO);
1646     GNUNET_STATISTICS_update (stats, "# bad paths", 1, GNUNET_NO);
1647
1648     LOG (GNUNET_ERROR_TYPE_DEBUG,
1649          "  register neighbors failed\n");
1650     LOG (GNUNET_ERROR_TYPE_DEBUG,
1651          "  prev: %s, neighbor?: %d\n",
1652          GCP_2s (c->prev_peer),
1653          GCP_is_neighbor (c->prev_peer));
1654     LOG (GNUNET_ERROR_TYPE_DEBUG,
1655          "  next: %s, neighbor?: %d\n",
1656          GCP_2s (c->next_peer),
1657          GCP_is_neighbor (c->next_peer));
1658     return GNUNET_SYSERR;
1659   }
1660   GCP_add_connection (c->next_peer, c, GNUNET_NO);
1661   GCP_add_connection (c->prev_peer, c, GNUNET_YES);
1662
1663   return GNUNET_OK;
1664 }
1665
1666
1667 /**
1668  * Remove the connection from the list of both neighbors.
1669  *
1670  * @param c Connection.
1671  */
1672 static void
1673 unregister_neighbors (struct CadetConnection *c)
1674 {
1675   struct CadetPeer *peer;
1676
1677   /* Either already unregistered or never got registered, it's ok either way. */
1678   if (NULL == c->path)
1679     return;
1680
1681   peer = get_next_hop (c);
1682   GNUNET_assert (c->next_peer == peer);
1683   GCP_remove_connection (peer, c);
1684   peer = get_prev_hop (c);
1685   GNUNET_assert (c->prev_peer == peer);
1686   GCP_remove_connection (peer, c);
1687 }
1688
1689
1690 /**
1691  * Bind the connection to the peer and the tunnel to that peer.
1692  *
1693  * If the peer has no tunnel, create one. Update tunnel and connection
1694  * data structres to reflect new status.
1695  *
1696  * @param c Connection.
1697  * @param peer Peer.
1698  */
1699 static void
1700 add_to_peer (struct CadetConnection *c,
1701              struct CadetPeer *peer)
1702 {
1703   GCP_add_tunnel (peer);
1704   c->t = GCP_get_tunnel (peer);
1705   GCT_add_connection (c->t, c);
1706 }
1707
1708
1709 /**
1710  * Iterator to compare each connection's path with the path of a new connection.
1711  *
1712  * If the connection conincides, the c member of path is set to the connection
1713  * and the destroy flag of the connection is set.
1714  *
1715  * @param cls Closure (new path).
1716  * @param c Connection in the tunnel to check.
1717  */
1718 static void
1719 check_path (void *cls, struct CadetConnection *c)
1720 {
1721   struct CadetConnection *new_conn = cls;
1722   struct CadetPeerPath *path = new_conn->path;
1723
1724   LOG (GNUNET_ERROR_TYPE_DEBUG, "  checking %s (%p), length %u\n",
1725        GCC_2s (c), c, c->path->length);
1726
1727   if (c != new_conn
1728       && c->destroy == GNUNET_NO
1729       && c->state != CADET_CONNECTION_BROKEN
1730       && c->state != CADET_CONNECTION_DESTROYED
1731       && path_equivalent (path, c->path))
1732   {
1733     new_conn->destroy = GNUNET_YES;
1734     new_conn->path->c = c;
1735     LOG (GNUNET_ERROR_TYPE_DEBUG, "  MATCH!\n");
1736   }
1737 }
1738
1739
1740 /**
1741  * Finds out if this path is already being used by and existing connection.
1742  *
1743  * Checks the tunnel towards the first peer in the path to see if it contains
1744  * any connection with the same path.
1745  *
1746  * If the existing connection is ready, it is kept.
1747  * Otherwise if the sender has a smaller ID that ours, we accept it (and
1748  * the peer will eventually reject our attempt).
1749  *
1750  * @param path Path to check.
1751  * @return #GNUNET_YES if the tunnel has a connection with the same path,
1752  *         #GNUNET_NO otherwise.
1753  */
1754 static int
1755 does_connection_exist (struct CadetConnection *conn)
1756 {
1757   struct CadetPeer *p;
1758   struct CadetTunnel *t;
1759   struct CadetConnection *c;
1760
1761   p = GCP_get_short (conn->path->peers[0]);
1762   t = GCP_get_tunnel (p);
1763   if (NULL == t)
1764     return GNUNET_NO;
1765
1766   LOG (GNUNET_ERROR_TYPE_DEBUG,
1767        "Checking for duplicates\n");
1768
1769   GCT_iterate_connections (t, &check_path, conn);
1770
1771   if (GNUNET_YES == conn->destroy)
1772   {
1773     c = conn->path->c;
1774     conn->destroy = GNUNET_NO;
1775     conn->path->c = conn;
1776     LOG (GNUNET_ERROR_TYPE_DEBUG, " found duplicate of %s\n", GCC_2s (conn));
1777     LOG (GNUNET_ERROR_TYPE_DEBUG, " duplicate: %s\n", GCC_2s (c));
1778     GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
1779     if (CADET_CONNECTION_READY == c->state)
1780     {
1781       /* The other peer confirmed a live connection with this path,
1782        * why is it trying to duplicate it. */
1783       return GNUNET_YES;
1784     }
1785
1786     if (GNUNET_CRYPTO_cmp_peer_identity (&my_full_id, GCP_get_id (p)) > 0)
1787     {
1788       struct CadetPeer *neighbor;
1789
1790       LOG (GNUNET_ERROR_TYPE_DEBUG, " duplicate allowed (killing old)\n");
1791       if (GCC_is_origin (c, GNUNET_YES))
1792         neighbor = get_next_hop (c);
1793       else
1794         neighbor = get_prev_hop (c);
1795       send_broken_unknown (&c->id, &my_full_id, NULL,
1796                            GCP_get_id (neighbor));
1797       GCC_destroy (c);
1798       return GNUNET_NO;
1799     }
1800     else
1801       return GNUNET_YES;
1802   }
1803   else
1804   {
1805     LOG (GNUNET_ERROR_TYPE_DEBUG, " %s has no duplicates\n", GCC_2s (conn));
1806     return GNUNET_NO;
1807   }
1808 }
1809
1810
1811 /**
1812  * Log receipt of message on stderr (INFO level).
1813  *
1814  * @param message Message received.
1815  * @param peer Peer who sent the message.
1816  * @param hash Connection ID.
1817  */
1818 static void
1819 log_message (const struct GNUNET_MessageHeader *message,
1820              const struct GNUNET_PeerIdentity *peer,
1821              const struct GNUNET_CADET_Hash *hash)
1822 {
1823   uint16_t size;
1824
1825   size = ntohs (message->size);
1826   LOG (GNUNET_ERROR_TYPE_INFO, "\n");
1827   LOG (GNUNET_ERROR_TYPE_INFO, "\n");
1828   LOG (GNUNET_ERROR_TYPE_INFO, "<-- %s on connection %s from %s, %6u bytes\n",
1829        GC_m2s (ntohs (message->type)), GNUNET_h2s (GC_h2hc (hash)),
1830        GNUNET_i2s (peer), (unsigned int) size);
1831 }
1832
1833 /******************************************************************************/
1834 /********************************    API    ***********************************/
1835 /******************************************************************************/
1836
1837 /**
1838  * Core handler for connection creation.
1839  *
1840  * @param cls Closure (unused).
1841  * @param peer Sender (neighbor).
1842  * @param message Message.
1843  * @return #GNUNET_OK to keep the connection open,
1844  *         #GNUNET_SYSERR to close it (signal serious error)
1845  */
1846 int
1847 GCC_handle_create (void *cls, const struct GNUNET_PeerIdentity *peer,
1848                    const struct GNUNET_MessageHeader *message)
1849 {
1850   struct GNUNET_CADET_ConnectionCreate *msg;
1851   struct GNUNET_PeerIdentity *id;
1852   struct GNUNET_CADET_Hash *cid;
1853   struct CadetPeerPath *path;
1854   struct CadetPeer *dest_peer;
1855   struct CadetPeer *orig_peer;
1856   struct CadetConnection *c;
1857   unsigned int own_pos;
1858   uint16_t size;
1859
1860   GCC_check_connections ();
1861   /* Check size */
1862   size = ntohs (message->size);
1863   if (size < sizeof (struct GNUNET_CADET_ConnectionCreate))
1864   {
1865     GNUNET_break_op (0);
1866     return GNUNET_OK;
1867   }
1868
1869   /* Calculate hops */
1870   size -= sizeof (struct GNUNET_CADET_ConnectionCreate);
1871   if (size % sizeof (struct GNUNET_PeerIdentity))
1872   {
1873     GNUNET_break_op (0);
1874     return GNUNET_OK;
1875   }
1876   if (0 != size % sizeof (struct GNUNET_PeerIdentity))
1877   {
1878     GNUNET_break_op (0);
1879     return GNUNET_OK;
1880   }
1881   size /= sizeof (struct GNUNET_PeerIdentity);
1882   if (1 > size)
1883   {
1884     GNUNET_break_op (0);
1885     return GNUNET_OK;
1886   }
1887   LOG (GNUNET_ERROR_TYPE_DEBUG, "    path has %u hops.\n", size);
1888
1889   /* Get parameters */
1890   msg = (struct GNUNET_CADET_ConnectionCreate *) message;
1891   cid = &msg->cid;
1892   log_message (message, peer, cid);
1893   id = (struct GNUNET_PeerIdentity *) &msg[1];
1894   LOG (GNUNET_ERROR_TYPE_DEBUG, "    origin: %s\n", GNUNET_i2s (id));
1895
1896   /* Create connection */
1897   c = connection_get (cid);
1898   if (NULL == c)
1899   {
1900     path = path_build_from_peer_ids ((struct GNUNET_PeerIdentity *) &msg[1],
1901                                      size, myid, &own_pos);
1902     if (NULL == path)
1903     {
1904       /* Path was malformed, probably our own ID was not in it. */
1905       GNUNET_STATISTICS_update (stats, "# malformed paths", 1, GNUNET_NO);
1906       GNUNET_break_op (0);
1907       return GNUNET_OK;
1908     }
1909
1910     if (0 == own_pos)
1911     {
1912       /* We received this request from a neighbor, we cannot be origin */
1913       GNUNET_STATISTICS_update (stats, "# fake paths", 1, GNUNET_NO);
1914       GNUNET_break_op (0);
1915       path_destroy (path);
1916       return GNUNET_OK;
1917     }
1918
1919     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Own position: %u\n", own_pos);
1920     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Creating connection\n");
1921     c = GCC_new (cid, NULL, path, own_pos);
1922     if (NULL == c)
1923     {
1924       if (path->length - 1 == own_pos)
1925       {
1926         /* If we are destination, why did the creation fail? */
1927         GNUNET_break (0);
1928         path_destroy (path);
1929         GCC_check_connections ();
1930         return GNUNET_OK;
1931       }
1932       send_broken_unknown (cid, &my_full_id,
1933                            GNUNET_PEER_resolve2 (path->peers[own_pos + 1]),
1934                            peer);
1935       path_destroy (path);
1936       GCC_check_connections ();
1937       return GNUNET_OK;
1938     }
1939     GCP_add_path_to_all (path, GNUNET_NO);
1940     connection_reset_timeout (c, GNUNET_YES);
1941   }
1942   else
1943   {
1944     path = path_duplicate (c->path);
1945   }
1946   if (CADET_CONNECTION_NEW == c->state)
1947     connection_change_state (c, CADET_CONNECTION_SENT);
1948
1949   /* Remember peers */
1950   dest_peer = GCP_get (&id[size - 1]);
1951   orig_peer = GCP_get (&id[0]);
1952
1953   /* Is it a connection to us? */
1954   if (c->own_pos == path->length - 1)
1955   {
1956     LOG (GNUNET_ERROR_TYPE_DEBUG, "  It's for us!\n");
1957     GCP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_YES);
1958
1959     add_to_peer (c, orig_peer);
1960     if (GNUNET_YES == does_connection_exist (c))
1961     {
1962       // FIXME Peer created a connection equal to one we think exists
1963       //       and is fine. What should we do?
1964       // Use explicit duplicate?
1965       // Accept new conn and destroy the old? (interruption in higher level)
1966       // Keep both and postpone disambiguation?
1967       // Keep the one created by peer with higher ID?
1968       // For now: reject new connection until current confirmed dead
1969       GNUNET_break_op (0);
1970       GCC_debug (c, GNUNET_ERROR_TYPE_WARNING);
1971       path_destroy (path);
1972       GCC_destroy (c);
1973       send_broken_unknown (cid, &my_full_id, NULL, peer);
1974       GCC_check_connections ();
1975       return GNUNET_OK;
1976     }
1977
1978     if (CADET_TUNNEL_NEW == GCT_get_cstate (c->t))
1979       GCT_change_cstate (c->t,  CADET_TUNNEL_WAITING);
1980
1981     send_connection_ack (c, GNUNET_NO);
1982     if (CADET_CONNECTION_SENT == c->state)
1983       connection_change_state (c, CADET_CONNECTION_ACK);
1984   }
1985   else
1986   {
1987     /* It's for somebody else! Retransmit. */
1988     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Retransmitting.\n");
1989     GCP_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
1990     GCP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_NO);
1991     GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c,
1992                                                       GNUNET_YES, GNUNET_YES,
1993                                                       NULL, NULL));
1994   }
1995   path_destroy (path);
1996   GCC_check_connections ();
1997   return GNUNET_OK;
1998 }
1999
2000
2001 /**
2002  * Core handler for path confirmations.
2003  *
2004  * @param cls closure
2005  * @param message message
2006  * @param peer peer identity this notification is about
2007  * @return #GNUNET_OK to keep the connection open,
2008  *         #GNUNET_SYSERR to close it (signal serious error)
2009  */
2010 int
2011 GCC_handle_confirm (void *cls,
2012                     const struct GNUNET_PeerIdentity *peer,
2013                     const struct GNUNET_MessageHeader *message)
2014 {
2015   struct GNUNET_CADET_ConnectionACK *msg;
2016   struct CadetConnection *c;
2017   struct CadetPeerPath *p;
2018   struct CadetPeer *pi;
2019   enum CadetConnectionState oldstate;
2020   int fwd;
2021
2022   GCC_check_connections ();
2023   msg = (struct GNUNET_CADET_ConnectionACK *) message;
2024   log_message (message, peer, &msg->cid);
2025   c = connection_get (&msg->cid);
2026   if (NULL == c)
2027   {
2028     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
2029                               1, GNUNET_NO);
2030     LOG (GNUNET_ERROR_TYPE_DEBUG,
2031          "  don't know the connection!\n");
2032     send_broken_unknown (&msg->cid, &my_full_id, NULL, peer);
2033     GCC_check_connections ();
2034     return GNUNET_OK;
2035   }
2036
2037   if (GNUNET_NO != c->destroy)
2038   {
2039     LOG (GNUNET_ERROR_TYPE_DEBUG,
2040          "  connection being destroyed\n");
2041     GCC_check_connections ();
2042     return GNUNET_OK;
2043   }
2044
2045   oldstate = c->state;
2046   LOG (GNUNET_ERROR_TYPE_DEBUG, "  via peer %s\n", GNUNET_i2s (peer));
2047   pi = GCP_get (peer);
2048   if (get_next_hop (c) == pi)
2049   {
2050     LOG (GNUNET_ERROR_TYPE_DEBUG, "  SYNACK\n");
2051     fwd = GNUNET_NO;
2052     if (CADET_CONNECTION_SENT == oldstate)
2053       connection_change_state (c, CADET_CONNECTION_ACK);
2054   }
2055   else if (get_prev_hop (c) == pi)
2056   {
2057     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FINAL ACK\n");
2058     fwd = GNUNET_YES;
2059     connection_change_state (c, CADET_CONNECTION_READY);
2060   }
2061   else
2062   {
2063     GNUNET_break_op (0);
2064     return GNUNET_OK;
2065   }
2066
2067   connection_reset_timeout (c, fwd);
2068
2069   /* Add path to peers? */
2070   p = c->path;
2071   if (NULL != p)
2072   {
2073     GCP_add_path_to_all (p, GNUNET_YES);
2074   }
2075   else
2076   {
2077     GNUNET_break (0);
2078   }
2079
2080   /* Message for us as creator? */
2081   if (GCC_is_origin (c, GNUNET_YES))
2082   {
2083     if (GNUNET_NO != fwd)
2084     {
2085       GNUNET_break_op (0);
2086       return GNUNET_OK;
2087     }
2088     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection (SYN)ACK for us!\n");
2089
2090     /* If just created, cancel the short timeout and start a long one */
2091     if (CADET_CONNECTION_SENT == oldstate)
2092       connection_reset_timeout (c, GNUNET_YES);
2093
2094     /* Change connection state */
2095     connection_change_state (c, CADET_CONNECTION_READY);
2096     send_connection_ack (c, GNUNET_YES);
2097
2098     /* Change tunnel state, trigger KX */
2099     if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2100       GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2101     GCC_check_connections ();
2102     return GNUNET_OK;
2103   }
2104
2105   /* Message for us as destination? */
2106   if (GCC_is_terminal (c, GNUNET_YES))
2107   {
2108     if (GNUNET_YES != fwd)
2109     {
2110       GNUNET_break_op (0);
2111       return GNUNET_OK;
2112     }
2113     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection ACK for us!\n");
2114
2115     /* If just created, cancel the short timeout and start a long one */
2116     if (CADET_CONNECTION_ACK == oldstate)
2117       connection_reset_timeout (c, GNUNET_NO);
2118
2119     /* Change tunnel state */
2120     if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2121       GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2122     GCC_check_connections ();
2123     return GNUNET_OK;
2124   }
2125
2126   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2127   GNUNET_assert (NULL ==
2128                  GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2129                                             GNUNET_YES, NULL, NULL));
2130   GCC_check_connections ();
2131   return GNUNET_OK;
2132 }
2133
2134
2135 /**
2136  * Core handler for notifications of broken connections.
2137  *
2138  * @param cls Closure (unused).
2139  * @param id Peer identity of sending neighbor.
2140  * @param message Message.
2141  * @return #GNUNET_OK to keep the connection open,
2142  *         #GNUNET_SYSERR to close it (signal serious error)
2143  */
2144 int
2145 GCC_handle_broken (void* cls,
2146                    const struct GNUNET_PeerIdentity* id,
2147                    const struct GNUNET_MessageHeader* message)
2148 {
2149   struct GNUNET_CADET_ConnectionBroken *msg;
2150   struct CadetConnection *c;
2151   struct CadetTunnel *t;
2152   int pending;
2153   int fwd;
2154
2155   GCC_check_connections ();
2156   msg = (struct GNUNET_CADET_ConnectionBroken *) message;
2157   log_message (message, id, &msg->cid);
2158   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
2159               GNUNET_i2s (&msg->peer1));
2160   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
2161               GNUNET_i2s (&msg->peer2));
2162   c = connection_get (&msg->cid);
2163   if (NULL == c)
2164   {
2165     LOG (GNUNET_ERROR_TYPE_DEBUG, "  duplicate CONNECTION_BROKEN\n");
2166     GCC_check_connections ();
2167     return GNUNET_OK;
2168   }
2169
2170   t = c->t;
2171
2172   fwd = is_fwd (c, id);
2173   c->destroy = GNUNET_YES;
2174   if (GCC_is_terminal (c, fwd))
2175   {
2176     struct CadetPeer *endpoint;
2177
2178     if (NULL == t)
2179     {
2180       /* A terminal connection should not have 't' set to NULL. */
2181       GNUNET_break (0);
2182       GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
2183       return GNUNET_OK;
2184     }
2185     endpoint = GCP_get_short (c->path->peers[c->path->length - 1]);
2186     if (2 < c->path->length)
2187       path_invalidate (c->path);
2188     GCP_notify_broken_link (endpoint, &msg->peer1, &msg->peer2);
2189
2190     c->state = CADET_CONNECTION_BROKEN;
2191     GCT_remove_connection (t, c);
2192     c->t = NULL;
2193
2194     pending = c->pending_messages;
2195     if (0 < pending)
2196       resend_messages_and_destroy (c, !fwd);
2197     else
2198       GCC_destroy (c);
2199   }
2200   else
2201   {
2202     GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2203                                                       GNUNET_YES, NULL, NULL));
2204     connection_cancel_queues (c, !fwd);
2205   }
2206   GCC_check_connections ();
2207   return GNUNET_OK;
2208 }
2209
2210
2211 /**
2212  * Core handler for tunnel destruction
2213  *
2214  * @param cls Closure (unused).
2215  * @param peer Peer identity of sending neighbor.
2216  * @param message Message.
2217  * @return #GNUNET_OK to keep the connection open,
2218  *         #GNUNET_SYSERR to close it (signal serious error)
2219  */
2220 int
2221 GCC_handle_destroy (void *cls,
2222                     const struct GNUNET_PeerIdentity *peer,
2223                     const struct GNUNET_MessageHeader *message)
2224 {
2225   const struct GNUNET_CADET_ConnectionDestroy *msg;
2226   struct CadetConnection *c;
2227   int fwd;
2228
2229   GCC_check_connections ();
2230   msg = (const struct GNUNET_CADET_ConnectionDestroy *) message;
2231   log_message (message, peer, &msg->cid);
2232   c = connection_get (&msg->cid);
2233   if (NULL == c)
2234   {
2235     /* Probably already got the message from another path,
2236      * destroyed the tunnel and retransmitted to children.
2237      * Safe to ignore.
2238      */
2239     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
2240                               1, GNUNET_NO);
2241     LOG (GNUNET_ERROR_TYPE_DEBUG,
2242          "  connection unknown: already destroyed?\n");
2243     GCC_check_connections ();
2244     return GNUNET_OK;
2245   }
2246   fwd = is_fwd (c, peer);
2247   if (GNUNET_SYSERR == fwd)
2248   {
2249     GNUNET_break_op (0); /* FIXME */
2250     return GNUNET_OK;
2251   }
2252   if (GNUNET_NO == GCC_is_terminal (c, fwd))
2253     GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2254                                                       GNUNET_YES, NULL, NULL));
2255   else if (0 == c->pending_messages)
2256   {
2257     LOG (GNUNET_ERROR_TYPE_DEBUG, "  directly destroying connection!\n");
2258     GCC_destroy (c);
2259     GCC_check_connections ();
2260     return GNUNET_OK;
2261   }
2262   c->destroy = GNUNET_YES;
2263   c->state = CADET_CONNECTION_DESTROYED;
2264   if (NULL != c->t)
2265   {
2266     GCT_remove_connection (c->t, c);
2267     c->t = NULL;
2268   }
2269   GCC_check_connections ();
2270
2271   return GNUNET_OK;
2272 }
2273
2274
2275 /**
2276  * Check the message against internal state and test if it goes FWD or BCK.
2277  *
2278  * Updates the PID, state and timeout values for the connection.
2279  *
2280  * @param message Message to check. It must belong to an existing connection.
2281  * @param minimum_size The message cannot be smaller than this value.
2282  * @param cid Connection ID (even if @a c is NULL, the ID is still needed).
2283  * @param c Connection this message should belong. If NULL, check fails.
2284  * @param neighbor Neighbor that sent the message.
2285  */
2286 static int
2287 check_message (const struct GNUNET_MessageHeader *message,
2288                size_t minimum_size,
2289                const struct GNUNET_CADET_Hash* cid,
2290                struct CadetConnection *c,
2291                const struct GNUNET_PeerIdentity *neighbor,
2292                uint32_t pid)
2293 {
2294   GNUNET_PEER_Id neighbor_id;
2295   struct CadetFlowControl *fc;
2296   struct CadetPeer *hop;
2297   int fwd;
2298   uint16_t type;
2299
2300   /* Check size */
2301   if (ntohs (message->size) < minimum_size)
2302   {
2303     GNUNET_break_op (0);
2304     LOG (GNUNET_ERROR_TYPE_WARNING, "Size %u < %u\n",
2305          ntohs (message->size), minimum_size);
2306     return GNUNET_SYSERR;
2307   }
2308
2309   /* Check connection */
2310   if (NULL == c)
2311   {
2312     GNUNET_STATISTICS_update (stats, "# unknown connection", 1, GNUNET_NO);
2313     LOG (GNUNET_ERROR_TYPE_DEBUG, "%s on unknown connection %s\n",
2314          GC_m2s (ntohs (message->type)), GNUNET_h2s (GC_h2hc (cid)));
2315     send_broken_unknown (cid, &my_full_id, NULL, neighbor);
2316     return GNUNET_SYSERR;
2317   }
2318
2319   /* Check if origin is as expected */
2320   neighbor_id = GNUNET_PEER_search (neighbor);
2321   hop = get_prev_hop (c);
2322   if (neighbor_id == GCP_get_short_id (hop))
2323   {
2324     fwd = GNUNET_YES;
2325   }
2326   else
2327   {
2328     hop = get_next_hop (c);
2329     GNUNET_break (hop == c->next_peer);
2330     if (neighbor_id == GCP_get_short_id (hop))
2331     {
2332       fwd = GNUNET_NO;
2333     }
2334     else
2335     {
2336       /* Unexpected peer sending traffic on a connection. */
2337       GNUNET_break_op (0);
2338       return GNUNET_SYSERR;
2339     }
2340   }
2341
2342   /* Check PID for payload messages */
2343   type = ntohs (message->type);
2344   if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type
2345       || GNUNET_MESSAGE_TYPE_CADET_AX == type)
2346   {
2347     fc = fwd ? &c->bck_fc : &c->fwd_fc;
2348     LOG (GNUNET_ERROR_TYPE_DEBUG, " PID %u (expected %u - %u)\n",
2349          pid, fc->last_pid_recv + 1, fc->last_ack_sent);
2350     if (GC_is_pid_bigger (pid, fc->last_ack_sent))
2351     {
2352       GNUNET_break_op (0);
2353       GNUNET_STATISTICS_update (stats, "# unsolicited message", 1, GNUNET_NO);
2354       LOG (GNUNET_ERROR_TYPE_WARNING, "Received PID %u, (prev %u), ACK %u\n",
2355           pid, fc->last_pid_recv, fc->last_ack_sent);
2356       return GNUNET_SYSERR;
2357     }
2358     if (GC_is_pid_bigger (pid, fc->last_pid_recv))
2359     {
2360       unsigned int delta;
2361
2362       delta = pid - fc->last_pid_recv;
2363       fc->last_pid_recv = pid;
2364       fc->recv_bitmap <<= delta;
2365       fc->recv_bitmap |= 1;
2366     }
2367     else
2368     {
2369       GNUNET_STATISTICS_update (stats, "# out of order PID", 1, GNUNET_NO);
2370       if (GNUNET_NO == is_ooo_ok (fc->last_pid_recv, pid, fc->recv_bitmap))
2371       {
2372         LOG (GNUNET_ERROR_TYPE_WARNING, "PID %u unexpected (%u+), dropping!\n",
2373              pid, fc->last_pid_recv - 31);
2374         return GNUNET_SYSERR;
2375       }
2376       fc->recv_bitmap |= get_recv_bitmask (fc->last_pid_recv, pid);
2377     }
2378   }
2379
2380   /* Count as connection confirmation. */
2381   if (CADET_CONNECTION_SENT == c->state || CADET_CONNECTION_ACK == c->state)
2382   {
2383     connection_change_state (c, CADET_CONNECTION_READY);
2384     if (NULL != c->t)
2385     {
2386       if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2387         GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2388     }
2389   }
2390   connection_reset_timeout (c, fwd);
2391
2392   return fwd;
2393 }
2394
2395
2396 /**
2397  * Generic handler for cadet network encrypted traffic.
2398  *
2399  * @param peer Peer identity this notification is about.
2400  * @param msg Encrypted message.
2401  * @return #GNUNET_OK to keep the connection open,
2402  *         #GNUNET_SYSERR to close it (signal serious error)
2403  */
2404 static int
2405 handle_cadet_encrypted (const struct GNUNET_PeerIdentity *peer,
2406                         const struct GNUNET_MessageHeader *message)
2407 {
2408   const struct GNUNET_CADET_Encrypted *otr_msg;
2409   const struct GNUNET_CADET_AX *ax_msg;
2410   const struct GNUNET_CADET_Hash* cid;
2411   struct CadetConnection *c;
2412   size_t minumum_size;
2413   size_t overhead;
2414   uint32_t pid;
2415   uint32_t ttl;
2416   int fwd;
2417
2418   GCC_check_connections ();
2419   if (GNUNET_MESSAGE_TYPE_CADET_AX == ntohs (message->type))
2420   {
2421     overhead = sizeof (struct GNUNET_CADET_AX);
2422     ax_msg = (const struct GNUNET_CADET_AX *) message;
2423     cid = &ax_msg->cid;
2424     pid = ntohl (ax_msg->pid);
2425     otr_msg = NULL;
2426   }
2427   else
2428   {
2429     overhead = sizeof (struct GNUNET_CADET_Encrypted);
2430     otr_msg = (const struct GNUNET_CADET_Encrypted *) message;
2431     cid = &otr_msg->cid;
2432     pid = ntohl (otr_msg->pid);
2433   }
2434
2435   log_message (message, peer, cid);
2436
2437   minumum_size = sizeof (struct GNUNET_MessageHeader) + overhead;
2438   c = connection_get (cid);
2439   fwd = check_message (message, minumum_size, cid, c, peer, pid);
2440
2441   /* If something went wrong, discard message. */
2442   if (GNUNET_SYSERR == fwd)
2443   {
2444     GCC_check_connections ();
2445     return GNUNET_OK;
2446   }
2447
2448   /* Is this message for us? */
2449   if (GCC_is_terminal (c, fwd))
2450   {
2451     GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
2452
2453     if (NULL == c->t)
2454     {
2455       GNUNET_break (GNUNET_NO != c->destroy);
2456       return GNUNET_OK;
2457     }
2458     GCT_handle_encrypted (c->t, message);
2459     GCC_send_ack (c, fwd, GNUNET_NO);
2460     GCC_check_connections ();
2461     return GNUNET_OK;
2462   }
2463
2464   /* Message not for us: forward to next hop */
2465   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2466   if (NULL != otr_msg) /* only otr has ttl */
2467   {
2468     ttl = ntohl (otr_msg->ttl);
2469     LOG (GNUNET_ERROR_TYPE_DEBUG, "   ttl: %u\n", ttl);
2470     if (ttl == 0)
2471     {
2472       GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
2473       LOG (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
2474       GCC_send_ack (c, fwd, GNUNET_NO);
2475       GCC_check_connections ();
2476       return GNUNET_OK;
2477     }
2478   }
2479
2480   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2481   GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2482                                                     GNUNET_NO, NULL, NULL));
2483   GCC_check_connections ();
2484   return GNUNET_OK;
2485 }
2486
2487
2488 /**
2489  * Generic handler for cadet network encrypted traffic.
2490  *
2491  * @param peer Peer identity this notification is about.
2492  * @param msg Encrypted message.
2493  * @return #GNUNET_OK to keep the connection open,
2494  *         #GNUNET_SYSERR to close it (signal serious error)
2495  */
2496 static int
2497 handle_cadet_kx (const struct GNUNET_PeerIdentity *peer,
2498                  const struct GNUNET_CADET_KX *msg)
2499 {
2500   const struct GNUNET_CADET_Hash* cid;
2501   struct CadetConnection *c;
2502   size_t expected_size;
2503   int fwd;
2504
2505   GCC_check_connections ();
2506   cid = &msg->cid;
2507   log_message (&msg->header, peer, cid);
2508
2509   expected_size = sizeof (struct GNUNET_CADET_KX)
2510                   + sizeof (struct GNUNET_MessageHeader);
2511   c = connection_get (cid);
2512   fwd = check_message (&msg->header, expected_size, cid, c, peer, 0);
2513
2514   /* If something went wrong, discard message. */
2515   if (GNUNET_SYSERR == fwd)
2516     return GNUNET_OK;
2517
2518   /* Is this message for us? */
2519   if (GCC_is_terminal (c, fwd))
2520   {
2521     LOG (GNUNET_ERROR_TYPE_DEBUG, "  message for us!\n");
2522     GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
2523     if (NULL == c->t)
2524     {
2525       GNUNET_break (0);
2526       return GNUNET_OK;
2527     }
2528     GCT_handle_kx (c->t, &msg[1].header);
2529     GCC_check_connections ();
2530     return GNUNET_OK;
2531   }
2532
2533   /* Message not for us: forward to next hop */
2534   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2535   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2536   GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg->header, 0, 0, c, fwd,
2537                                                     GNUNET_NO, NULL, NULL));
2538   GCC_check_connections ();
2539   return GNUNET_OK;
2540 }
2541
2542
2543 /**
2544  * Core handler for key exchange traffic (ephemeral key, ping, pong).
2545  *
2546  * @param cls Closure (unused).
2547  * @param message Message received.
2548  * @param peer Peer who sent the message.
2549  * @return #GNUNET_OK to keep the connection open,
2550  *         #GNUNET_SYSERR to close it (signal serious error)
2551  */
2552 int
2553 GCC_handle_kx (void *cls,
2554                const struct GNUNET_PeerIdentity *peer,
2555                const struct GNUNET_MessageHeader *message)
2556 {
2557   GCC_check_connections ();
2558   return handle_cadet_kx (peer, (struct GNUNET_CADET_KX *) message);
2559 }
2560
2561
2562 /**
2563  * Core handler for encrypted cadet network traffic (channel mgmt, data).
2564  *
2565  * @param cls Closure (unused).
2566  * @param message Message received.
2567  * @param peer Peer who sent the message.
2568  * @return #GNUNET_OK to keep the connection open,
2569  *         #GNUNET_SYSERR to close it (signal serious error)
2570  */
2571 int
2572 GCC_handle_encrypted (void *cls, const struct GNUNET_PeerIdentity *peer,
2573                       const struct GNUNET_MessageHeader *message)
2574 {
2575   GCC_check_connections ();
2576   return handle_cadet_encrypted (peer, message);
2577 }
2578
2579
2580 /**
2581  * Core handler for cadet network traffic point-to-point acks.
2582  *
2583  * @param cls closure
2584  * @param message message
2585  * @param peer peer identity this notification is about
2586  * @return #GNUNET_OK to keep the connection open,
2587  *         #GNUNET_SYSERR to close it (signal serious error)
2588  */
2589 int
2590 GCC_handle_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
2591                 const struct GNUNET_MessageHeader *message)
2592 {
2593   struct GNUNET_CADET_ACK *msg;
2594   struct CadetConnection *c;
2595   struct CadetFlowControl *fc;
2596   GNUNET_PEER_Id id;
2597   uint32_t ack;
2598   int fwd;
2599
2600   GCC_check_connections ();
2601   msg = (struct GNUNET_CADET_ACK *) message;
2602   log_message (message, peer, &msg->cid);
2603   c = connection_get (&msg->cid);
2604   if (NULL == c)
2605   {
2606     GNUNET_STATISTICS_update (stats, "# ack on unknown connection", 1,
2607                               GNUNET_NO);
2608     send_broken_unknown (&msg->cid, &my_full_id, NULL, peer);
2609     GCC_check_connections ();
2610     return GNUNET_OK;
2611   }
2612
2613   /* Is this a forward or backward ACK? */
2614   id = GNUNET_PEER_search (peer);
2615   if (GCP_get_short_id (get_next_hop (c)) == id)
2616   {
2617     fc = &c->fwd_fc;
2618     fwd = GNUNET_YES;
2619   }
2620   else if (GCP_get_short_id (get_prev_hop (c)) == id)
2621   {
2622     fc = &c->bck_fc;
2623     fwd = GNUNET_NO;
2624   }
2625   else
2626   {
2627     GNUNET_break_op (0);
2628     return GNUNET_OK;
2629   }
2630
2631   ack = ntohl (msg->ack);
2632   LOG (GNUNET_ERROR_TYPE_DEBUG, " %s ACK %u (was %u)\n",
2633        GC_f2s (fwd), ack, fc->last_ack_recv);
2634   if (GC_is_pid_bigger (ack, fc->last_ack_recv))
2635     fc->last_ack_recv = ack;
2636
2637   /* Cancel polling if the ACK is big enough. */
2638   if (NULL != fc->poll_task &&
2639       GC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
2640   {
2641     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Cancel poll\n");
2642     GNUNET_SCHEDULER_cancel (fc->poll_task);
2643     fc->poll_task = NULL;
2644     fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
2645   }
2646
2647   connection_unlock_queue (c, fwd);
2648   GCC_check_connections ();
2649   return GNUNET_OK;
2650 }
2651
2652
2653 /**
2654  * Core handler for cadet network traffic point-to-point ack polls.
2655  *
2656  * @param cls closure
2657  * @param message message
2658  * @param peer peer identity this notification is about
2659  * @return #GNUNET_OK to keep the connection open,
2660  *         #GNUNET_SYSERR to close it (signal serious error)
2661  */
2662 int
2663 GCC_handle_poll (void *cls,
2664                  const struct GNUNET_PeerIdentity *peer,
2665                  const struct GNUNET_MessageHeader *message)
2666 {
2667   struct GNUNET_CADET_Poll *msg;
2668   struct CadetConnection *c;
2669   struct CadetFlowControl *fc;
2670   GNUNET_PEER_Id id;
2671   uint32_t pid;
2672   int fwd;
2673
2674   GCC_check_connections ();
2675   msg = (struct GNUNET_CADET_Poll *) message;
2676   log_message (message, peer, &msg->cid);
2677   c = connection_get (&msg->cid);
2678   if (NULL == c)
2679   {
2680     GNUNET_STATISTICS_update (stats, "# poll on unknown connection", 1,
2681                               GNUNET_NO);
2682     LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL message on unknown connection %s!\n",
2683          GNUNET_h2s (GC_h2hc (&msg->cid)));
2684     send_broken_unknown (&msg->cid, &my_full_id, NULL, peer);
2685     GCC_check_connections ();
2686     return GNUNET_OK;
2687   }
2688
2689   /* Is this a forward or backward ACK?
2690    * Note: a poll should never be needed in a loopback case,
2691    * since there is no possiblility of packet loss there, so
2692    * this way of discerining FWD/BCK should not be a problem.
2693    */
2694   id = GNUNET_PEER_search (peer);
2695   if (GCP_get_short_id (get_next_hop (c)) == id)
2696   {
2697     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD FC\n");
2698     fc = &c->fwd_fc;
2699   }
2700   else if (GCP_get_short_id (get_prev_hop (c)) == id)
2701   {
2702     LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK FC\n");
2703     fc = &c->bck_fc;
2704   }
2705   else
2706   {
2707     GNUNET_break_op (0);
2708     return GNUNET_OK;
2709   }
2710
2711   pid = ntohl (msg->pid);
2712   LOG (GNUNET_ERROR_TYPE_DEBUG, "  PID %u, OLD %u\n", pid, fc->last_pid_recv);
2713   fc->last_pid_recv = pid;
2714   fwd = fc == &c->bck_fc;
2715   GCC_send_ack (c, fwd, GNUNET_YES);
2716   GCC_check_connections ();
2717
2718   return GNUNET_OK;
2719 }
2720
2721
2722 /**
2723  * Send an ACK on the appropriate connection/channel, depending on
2724  * the direction and the position of the peer.
2725  *
2726  * @param c Which connection to send the hop-by-hop ACK.
2727  * @param fwd Is this a fwd ACK? (will go dest->root).
2728  * @param force Send the ACK even if suboptimal (e.g. requested by POLL).
2729  */
2730 void
2731 GCC_send_ack (struct CadetConnection *c, int fwd, int force)
2732 {
2733   unsigned int buffer;
2734
2735   GCC_check_connections ();
2736   LOG (GNUNET_ERROR_TYPE_DEBUG, "GCC send %s ACK on %s\n",
2737        GC_f2s (fwd), GCC_2s (c));
2738
2739   if (NULL == c)
2740   {
2741     GNUNET_break (0);
2742     return;
2743   }
2744
2745   if (GNUNET_NO != c->destroy)
2746   {
2747     LOG (GNUNET_ERROR_TYPE_DEBUG, "  being destroyed, why bother...\n");
2748     GCC_check_connections ();
2749     return;
2750   }
2751
2752   /* Get available buffer space */
2753   if (GCC_is_terminal (c, fwd))
2754   {
2755     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from all channels\n");
2756     buffer = GCT_get_channels_buffer (c->t);
2757   }
2758   else
2759   {
2760     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from one connection\n");
2761     buffer = GCC_get_buffer (c, fwd);
2762   }
2763   LOG (GNUNET_ERROR_TYPE_DEBUG, "  buffer available: %u\n", buffer);
2764   if (0 == buffer && GNUNET_NO == force)
2765   {
2766     GCC_check_connections ();
2767     return;
2768   }
2769
2770   /* Send available buffer space */
2771   if (GCC_is_origin (c, fwd))
2772   {
2773     GNUNET_assert (NULL != c->t);
2774     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on channels...\n");
2775     GCT_unchoke_channels (c->t);
2776   }
2777   else
2778   {
2779     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on connection\n");
2780     send_ack (c, buffer, fwd, force);
2781   }
2782   GCC_check_connections ();
2783 }
2784
2785
2786 /**
2787  * Initialize the connections subsystem
2788  *
2789  * @param c Configuration handle.
2790  */
2791 void
2792 GCC_init (const struct GNUNET_CONFIGURATION_Handle *c)
2793 {
2794   LOG (GNUNET_ERROR_TYPE_DEBUG, "init\n");
2795   if (GNUNET_OK !=
2796       GNUNET_CONFIGURATION_get_value_number (c, "CADET", "MAX_MSGS_QUEUE",
2797                                              &max_msgs_queue))
2798   {
2799     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2800                                "CADET", "MAX_MSGS_QUEUE", "MISSING");
2801     GNUNET_SCHEDULER_shutdown ();
2802     return;
2803   }
2804
2805   if (GNUNET_OK !=
2806       GNUNET_CONFIGURATION_get_value_number (c, "CADET", "MAX_CONNECTIONS",
2807                                              &max_connections))
2808   {
2809     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2810                                "CADET", "MAX_CONNECTIONS", "MISSING");
2811     GNUNET_SCHEDULER_shutdown ();
2812     return;
2813   }
2814
2815   if (GNUNET_OK !=
2816       GNUNET_CONFIGURATION_get_value_time (c, "CADET", "REFRESH_CONNECTION_TIME",
2817                                            &refresh_connection_time))
2818   {
2819     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2820                                "CADET", "REFRESH_CONNECTION_TIME", "MISSING");
2821     GNUNET_SCHEDULER_shutdown ();
2822     return;
2823   }
2824   create_connection_time = GNUNET_TIME_UNIT_SECONDS;
2825   connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_NO);
2826 }
2827
2828
2829 /**
2830  * Destroy each connection on shutdown.
2831  *
2832  * @param cls Closure (unused).
2833  * @param key Current key code (CID, unused).
2834  * @param value Value in the hash map (`struct CadetConnection`)
2835  *
2836  * @return #GNUNET_YES, because we should continue to iterate
2837  */
2838 static int
2839 shutdown_iterator (void *cls,
2840                    const struct GNUNET_HashCode *key,
2841                    void *value)
2842 {
2843   struct CadetConnection *c = value;
2844
2845   c->state = CADET_CONNECTION_DESTROYED;
2846   GCC_destroy (c);
2847   return GNUNET_YES;
2848 }
2849
2850
2851 /**
2852  * Shut down the connections subsystem.
2853  */
2854 void
2855 GCC_shutdown (void)
2856 {
2857   GCC_check_connections ();
2858   GNUNET_CONTAINER_multihashmap_iterate (connections,
2859                                          &shutdown_iterator,
2860                                          NULL);
2861   GNUNET_CONTAINER_multihashmap_destroy (connections);
2862   connections = NULL;
2863 }
2864
2865
2866 /**
2867  * Create a connection.
2868  *
2869  * @param cid Connection ID (either created locally or imposed remotely).
2870  * @param t Tunnel this connection belongs to (or NULL);
2871  * @param path Path this connection has to use (copy is made).
2872  * @param own_pos Own position in the @c path path.
2873  *
2874  * @return Newly created connection, NULL in case of error (own id not in path).
2875  */
2876 struct CadetConnection *
2877 GCC_new (const struct GNUNET_CADET_Hash *cid,
2878          struct CadetTunnel *t,
2879          struct CadetPeerPath *path,
2880          unsigned int own_pos)
2881 {
2882   struct CadetConnection *c;
2883   struct CadetPeerPath *p;
2884
2885   GCC_check_connections ();
2886   p = path_duplicate (path);
2887   c = GNUNET_new (struct CadetConnection);
2888   c->id = *cid;
2889   GNUNET_assert (GNUNET_OK ==
2890                  GNUNET_CONTAINER_multihashmap_put (connections,
2891                                                     GCC_get_h (c), c,
2892                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
2893   fc_init (&c->fwd_fc);
2894   fc_init (&c->bck_fc);
2895   c->fwd_fc.c = c;
2896   c->bck_fc.c = c;
2897
2898   c->t = t;
2899   GNUNET_assert (own_pos <= p->length - 1);
2900   c->own_pos = own_pos;
2901   c->path = p;
2902   p->c = c;
2903   GNUNET_assert (NULL != p);
2904   if (GNUNET_OK != register_neighbors (c))
2905   {
2906     if (0 == own_pos)
2907     {
2908       /* We were the origin of this request, this means we have invalid
2909        * info about the paths to reach the destination. We must invalidate
2910        * the *original* path to avoid trying it again in the next minute.
2911        */
2912       path_invalidate (path);
2913       c->t = NULL;
2914     }
2915     path_destroy (c->path);
2916     c->path = NULL;
2917     GCC_destroy (c);
2918     return NULL;
2919   }
2920   LOG (GNUNET_ERROR_TYPE_INFO, "New connection %s\n", GCC_2s (c));
2921   GCC_check_connections ();
2922   return c;
2923 }
2924
2925
2926 void
2927 GCC_destroy (struct CadetConnection *c)
2928 {
2929   GCC_check_connections ();
2930   if (NULL == c)
2931   {
2932     GNUNET_break (0);
2933     return;
2934   }
2935
2936   if (2 == c->destroy) /* cancel queues -> GCP_queue_cancel -> q_destroy -> */
2937     return;            /* -> message_sent -> GCC_destroy. Don't loop. */
2938   c->destroy = 2;
2939
2940   LOG (GNUNET_ERROR_TYPE_DEBUG,
2941        "destroying connection %s\n",
2942        GCC_2s (c));
2943   LOG (GNUNET_ERROR_TYPE_DEBUG,
2944        " fc's f: %p, b: %p\n",
2945        &c->fwd_fc, &c->bck_fc);
2946   LOG (GNUNET_ERROR_TYPE_DEBUG,
2947        " fc tasks f: %u, b: %u\n",
2948        c->fwd_fc.poll_task,
2949        c->bck_fc.poll_task);
2950
2951   /* Cancel all traffic */
2952   if (NULL != c->path)
2953   {
2954     connection_cancel_queues (c, GNUNET_YES);
2955     connection_cancel_queues (c, GNUNET_NO);
2956   }
2957   unregister_neighbors (c);
2958   path_destroy (c->path);
2959   c->path = NULL;
2960
2961   /* Cancel maintainance task (keepalive/timeout) */
2962   if (NULL != c->fwd_fc.poll_msg)
2963   {
2964     GCC_cancel (c->fwd_fc.poll_msg);
2965     LOG (GNUNET_ERROR_TYPE_DEBUG,
2966          " POLL msg FWD canceled\n");
2967   }
2968   if (NULL != c->bck_fc.poll_msg)
2969   {
2970     GCC_cancel (c->bck_fc.poll_msg);
2971     LOG (GNUNET_ERROR_TYPE_DEBUG,
2972          " POLL msg BCK canceled\n");
2973   }
2974
2975   /* Delete from tunnel */
2976   if (NULL != c->t)
2977     GCT_remove_connection (c->t, c);
2978
2979   if (NULL != c->fwd_maintenance_task)
2980     GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
2981   if (NULL != c->bck_maintenance_task)
2982     GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
2983   if (NULL != c->fwd_fc.poll_task)
2984   {
2985     GNUNET_SCHEDULER_cancel (c->fwd_fc.poll_task);
2986     LOG (GNUNET_ERROR_TYPE_DEBUG, " POLL task FWD canceled\n");
2987   }
2988   if (NULL != c->bck_fc.poll_task)
2989   {
2990     GNUNET_SCHEDULER_cancel (c->bck_fc.poll_task);
2991     LOG (GNUNET_ERROR_TYPE_DEBUG, " POLL task BCK canceled\n");
2992   }
2993
2994   GNUNET_break (GNUNET_YES ==
2995                 GNUNET_CONTAINER_multihashmap_remove (connections,
2996                                                       GCC_get_h (c),
2997                                                       c));
2998   GNUNET_STATISTICS_update (stats,
2999                             "# connections",
3000                             -1,
3001                             GNUNET_NO);
3002   GNUNET_free (c);
3003   GCC_check_connections ();
3004 }
3005
3006
3007 /**
3008  * Get the connection ID.
3009  *
3010  * @param c Connection to get the ID from.
3011  *
3012  * @return ID of the connection.
3013  */
3014 const struct GNUNET_CADET_Hash *
3015 GCC_get_id (const struct CadetConnection *c)
3016 {
3017   return &c->id;
3018 }
3019
3020
3021 /**
3022  * Get the connection ID.
3023  *
3024  * @param c Connection to get the ID from.
3025  *
3026  * @return ID of the connection.
3027  */
3028 const struct GNUNET_HashCode *
3029 GCC_get_h (const struct CadetConnection *c)
3030 {
3031   return GC_h2hc (&c->id);
3032 }
3033
3034
3035 /**
3036  * Get the connection path.
3037  *
3038  * @param c Connection to get the path from.
3039  *
3040  * @return path used by the connection.
3041  */
3042 const struct CadetPeerPath *
3043 GCC_get_path (const struct CadetConnection *c)
3044 {
3045   if (GNUNET_NO == c->destroy)
3046     return c->path;
3047   return NULL;
3048 }
3049
3050
3051 /**
3052  * Get the connection state.
3053  *
3054  * @param c Connection to get the state from.
3055  *
3056  * @return state of the connection.
3057  */
3058 enum CadetConnectionState
3059 GCC_get_state (const struct CadetConnection *c)
3060 {
3061   return c->state;
3062 }
3063
3064 /**
3065  * Get the connection tunnel.
3066  *
3067  * @param c Connection to get the tunnel from.
3068  *
3069  * @return tunnel of the connection.
3070  */
3071 struct CadetTunnel *
3072 GCC_get_tunnel (const struct CadetConnection *c)
3073 {
3074   return c->t;
3075 }
3076
3077
3078 /**
3079  * Get free buffer space in a connection.
3080  *
3081  * @param c Connection.
3082  * @param fwd Is query about FWD traffic?
3083  *
3084  * @return Free buffer space [0 - max_msgs_queue/max_connections]
3085  */
3086 unsigned int
3087 GCC_get_buffer (struct CadetConnection *c, int fwd)
3088 {
3089   struct CadetFlowControl *fc;
3090
3091   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3092
3093   LOG (GNUNET_ERROR_TYPE_DEBUG, "  Get %s buffer on %s: %u - %u\n",
3094        GC_f2s (fwd), GCC_2s (c), fc->queue_max, fc->queue_n);
3095   GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
3096
3097   return (fc->queue_max - fc->queue_n);
3098 }
3099
3100
3101 /**
3102  * Get how many messages have we allowed to send to us from a direction.
3103  *
3104  * @param c Connection.
3105  * @param fwd Are we asking about traffic from FWD (BCK messages)?
3106  *
3107  * @return last_ack_sent - last_pid_recv
3108  */
3109 unsigned int
3110 GCC_get_allowed (struct CadetConnection *c, int fwd)
3111 {
3112   struct CadetFlowControl *fc;
3113
3114   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3115   if (CADET_CONNECTION_READY != c->state
3116       || GC_is_pid_bigger (fc->last_pid_recv, fc->last_ack_sent))
3117   {
3118     return 0;
3119   }
3120   return (fc->last_ack_sent - fc->last_pid_recv);
3121 }
3122
3123
3124 /**
3125  * Get messages queued in a connection.
3126  *
3127  * @param c Connection.
3128  * @param fwd Is query about FWD traffic?
3129  *
3130  * @return Number of messages queued.
3131  */
3132 unsigned int
3133 GCC_get_qn (struct CadetConnection *c, int fwd)
3134 {
3135   struct CadetFlowControl *fc;
3136
3137   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3138
3139   return fc->queue_n;
3140 }
3141
3142
3143 /**
3144  * Get next PID to use.
3145  *
3146  * @param c Connection.
3147  * @param fwd Is query about FWD traffic?
3148  *
3149  * @return Last PID used + 1.
3150  */
3151 unsigned int
3152 GCC_get_pid (struct CadetConnection *c, int fwd)
3153 {
3154   struct CadetFlowControl *fc;
3155
3156   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3157
3158   return fc->last_pid_sent + 1;
3159 }
3160
3161
3162 /**
3163  * Allow the connection to advertise a buffer of the given size.
3164  *
3165  * The connection will send an @c fwd ACK message (so: in direction !fwd)
3166  * allowing up to last_pid_recv + buffer.
3167  *
3168  * @param c Connection.
3169  * @param buffer How many more messages the connection can accept.
3170  * @param fwd Is this about FWD traffic? (The ack will go dest->root).
3171  */
3172 void
3173 GCC_allow (struct CadetConnection *c, unsigned int buffer, int fwd)
3174 {
3175   LOG (GNUNET_ERROR_TYPE_DEBUG, "  allowing %s %u messages %s\n",
3176        GCC_2s (c), buffer, GC_f2s (fwd));
3177   send_ack (c, buffer, fwd, GNUNET_NO);
3178 }
3179
3180
3181 /**
3182  * Notify other peers on a connection of a broken link. Mark connections
3183  * to destroy after all traffic has been sent.
3184  *
3185  * @param c Connection on which there has been a disconnection.
3186  * @param peer Peer that disconnected.
3187  */
3188 void
3189 GCC_notify_broken (struct CadetConnection *c,
3190                    struct CadetPeer *peer)
3191 {
3192   struct CadetPeer *hop;
3193   int fwd;
3194
3195   GCC_check_connections ();
3196   LOG (GNUNET_ERROR_TYPE_DEBUG,
3197        "Notify broken on %s due to %s disconnect\n",
3198        GCC_2s (c),
3199        GCP_2s (peer));
3200   hop = get_prev_hop (c);
3201   if (NULL == hop)
3202   {
3203     /* Path was NULL, we should have deleted the connection. */
3204     GNUNET_break (0);
3205     return;
3206   }
3207   fwd = (peer == hop);
3208   if (GNUNET_YES == GCC_is_terminal (c, fwd))
3209   {
3210     /* Local shutdown, no one to notify about this. */
3211     GCC_destroy (c);
3212     GCC_check_connections ();
3213     return;
3214   }
3215   if (GNUNET_NO == c->destroy)
3216     send_broken (c, &my_full_id, GCP_get_id (peer), fwd);
3217
3218   /* Connection will have at least one pending message
3219    * (the one we just scheduled), so no point in checking whether to
3220    * destroy immediately. */
3221   c->destroy = GNUNET_YES;
3222   c->state = CADET_CONNECTION_DESTROYED;
3223
3224   /**
3225    * Cancel all queues, if no message is left, connection will be destroyed.
3226    */
3227   connection_cancel_queues (c, ! fwd);
3228   GCC_check_connections ();
3229 }
3230
3231
3232 /**
3233  * Is this peer the first one on the connection?
3234  *
3235  * @param c Connection.
3236  * @param fwd Is this about fwd traffic?
3237  *
3238  * @return #GNUNET_YES if origin, #GNUNET_NO if relay/terminal.
3239  */
3240 int
3241 GCC_is_origin (struct CadetConnection *c, int fwd)
3242 {
3243   if (!fwd && c->path->length - 1 == c->own_pos )
3244     return GNUNET_YES;
3245   if (fwd && 0 == c->own_pos)
3246     return GNUNET_YES;
3247   return GNUNET_NO;
3248 }
3249
3250
3251 /**
3252  * Is this peer the last one on the connection?
3253  *
3254  * @param c Connection.
3255  * @param fwd Is this about fwd traffic?
3256  *            Note that the ROOT is the terminal for BCK traffic!
3257  *
3258  * @return #GNUNET_YES if terminal, #GNUNET_NO if relay/origin.
3259  */
3260 int
3261 GCC_is_terminal (struct CadetConnection *c, int fwd)
3262 {
3263   return GCC_is_origin (c, !fwd);
3264 }
3265
3266
3267 /**
3268  * See if we are allowed to send by the next hop in the given direction.
3269  *
3270  * @param c Connection.
3271  * @param fwd Is this about fwd traffic?
3272  *
3273  * @return #GNUNET_YES in case it's OK to send.
3274  */
3275 int
3276 GCC_is_sendable (struct CadetConnection *c, int fwd)
3277 {
3278   struct CadetFlowControl *fc;
3279
3280   LOG (GNUNET_ERROR_TYPE_DEBUG,
3281        " checking sendability of %s traffic on %s\n",
3282        GC_f2s (fwd), GCC_2s (c));
3283   if (NULL == c)
3284   {
3285     GNUNET_break (0);
3286     return GNUNET_YES;
3287   }
3288   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3289   LOG (GNUNET_ERROR_TYPE_DEBUG,
3290        " last ack recv: %u, last pid sent: %u\n",
3291        fc->last_ack_recv, fc->last_pid_sent);
3292   if (GC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
3293   {
3294     LOG (GNUNET_ERROR_TYPE_DEBUG, " sendable\n");
3295     return GNUNET_YES;
3296   }
3297   LOG (GNUNET_ERROR_TYPE_DEBUG, " not sendable\n");
3298   return GNUNET_NO;
3299 }
3300
3301
3302 /**
3303  * Check if this connection is a direct one (never trim a direct connection).
3304  *
3305  * @param c Connection.
3306  *
3307  * @return #GNUNET_YES in case it's a direct connection, #GNUNET_NO otherwise.
3308  */
3309 int
3310 GCC_is_direct (struct CadetConnection *c)
3311 {
3312   return (c->path->length == 2) ? GNUNET_YES : GNUNET_NO;
3313 }
3314
3315
3316 /**
3317  * Sends an already built message on a connection, properly registering
3318  * all used resources.
3319  *
3320  * @param message Message to send. Function makes a copy of it.
3321  *                If message is not hop-by-hop, decrements TTL of copy.
3322  * @param payload_type Type of payload, in case the message is encrypted.
3323  * @param c Connection on which this message is transmitted.
3324  * @param fwd Is this a fwd message?
3325  * @param force Force the connection to accept the message (buffer overfill).
3326  * @param cont Continuation called once message is sent. Can be NULL.
3327  * @param cont_cls Closure for @c cont.
3328  *
3329  * @return Handle to cancel the message before it's sent.
3330  *         NULL on error or if @c cont is NULL.
3331  *         Invalid on @c cont call.
3332  */
3333 struct CadetConnectionQueue *
3334 GCC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
3335                            uint16_t payload_type, uint32_t payload_id,
3336                            struct CadetConnection *c, int fwd, int force,
3337                            GCC_sent cont, void *cont_cls)
3338 {
3339   struct CadetFlowControl *fc;
3340   struct CadetConnectionQueue *q;
3341   void *data;
3342   size_t size;
3343   uint16_t type;
3344   int droppable;
3345
3346   GCC_check_connections ();
3347   size = ntohs (message->size);
3348   data = GNUNET_malloc (size);
3349   memcpy (data, message, size);
3350   type = ntohs (message->type);
3351   LOG (GNUNET_ERROR_TYPE_INFO, "--> %s (%s %4u) on connection %s (%u bytes)\n",
3352        GC_m2s (type), GC_m2s (payload_type), payload_id, GCC_2s (c), size);
3353
3354   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3355   droppable = GNUNET_NO == force;
3356   switch (type)
3357   {
3358     struct GNUNET_CADET_AX        *axmsg;
3359     struct GNUNET_CADET_Encrypted *emsg;
3360     struct GNUNET_CADET_KX        *kmsg;
3361     struct GNUNET_CADET_ACK       *amsg;
3362     struct GNUNET_CADET_Poll      *pmsg;
3363     struct GNUNET_CADET_ConnectionDestroy *dmsg;
3364     struct GNUNET_CADET_ConnectionBroken  *bmsg;
3365     uint32_t ttl;
3366
3367     case GNUNET_MESSAGE_TYPE_CADET_AX:
3368     case GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED:
3369       if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type)
3370       {
3371         emsg = (struct GNUNET_CADET_Encrypted *) data;
3372         ttl = ntohl (emsg->ttl);
3373         if (0 == ttl)
3374         {
3375           GNUNET_break_op (0);
3376           GNUNET_free (data);
3377           return NULL;
3378         }
3379         emsg->cid = c->id;
3380         emsg->ttl = htonl (ttl - 1);
3381       }
3382       else
3383       {
3384         axmsg = (struct GNUNET_CADET_AX *) data;
3385         axmsg->cid = c->id;
3386       }
3387       LOG (GNUNET_ERROR_TYPE_DEBUG, "  Q_N+ %p %u\n", fc, fc->queue_n);
3388       LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid sent %u\n", fc->last_pid_sent);
3389       LOG (GNUNET_ERROR_TYPE_DEBUG, "     ack recv %u\n", fc->last_ack_recv);
3390       if (GNUNET_YES == droppable)
3391       {
3392         fc->queue_n++;
3393       }
3394       else
3395       {
3396         LOG (GNUNET_ERROR_TYPE_DEBUG, "  not droppable, Q_N stays the same\n");
3397       }
3398       break;
3399
3400     case GNUNET_MESSAGE_TYPE_CADET_KX:
3401       kmsg = (struct GNUNET_CADET_KX *) data;
3402       kmsg->cid = c->id;
3403       break;
3404
3405     case GNUNET_MESSAGE_TYPE_CADET_ACK:
3406       amsg = (struct GNUNET_CADET_ACK *) data;
3407       amsg->cid = c->id;
3408       LOG (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
3409       droppable = GNUNET_NO;
3410       break;
3411
3412     case GNUNET_MESSAGE_TYPE_CADET_POLL:
3413       pmsg = (struct GNUNET_CADET_Poll *) data;
3414       pmsg->cid = c->id;
3415       LOG (GNUNET_ERROR_TYPE_DEBUG, " POLL %u\n", ntohl (pmsg->pid));
3416       droppable = GNUNET_NO;
3417       break;
3418
3419     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY:
3420       dmsg = (struct GNUNET_CADET_ConnectionDestroy *) data;
3421       dmsg->cid = c->id;
3422       break;
3423
3424     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN:
3425       bmsg = (struct GNUNET_CADET_ConnectionBroken *) data;
3426       bmsg->cid = c->id;
3427       break;
3428
3429     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE:
3430     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_ACK:
3431       break;
3432
3433     default:
3434       GNUNET_break (0);
3435       GNUNET_free (data);
3436       return NULL;
3437   }
3438
3439   if (fc->queue_n > fc->queue_max && droppable)
3440   {
3441     GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
3442                               1, GNUNET_NO);
3443     GNUNET_break (0);
3444     LOG (GNUNET_ERROR_TYPE_DEBUG, "queue full: %u/%u\n",
3445          fc->queue_n, fc->queue_max);
3446     if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type
3447         || GNUNET_MESSAGE_TYPE_CADET_AX == type)
3448     {
3449       fc->queue_n--;
3450     }
3451     GNUNET_free (data);
3452     return NULL; /* Drop this message */
3453   }
3454
3455   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %s %u\n",
3456        GCC_2s (c), c->pending_messages);
3457   c->pending_messages++;
3458
3459   q = GNUNET_new (struct CadetConnectionQueue);
3460   q->forced = !droppable;
3461   q->q = GCP_queue_add (get_hop (c, fwd), data, type, payload_type, payload_id,
3462                         size, c, fwd, &conn_message_sent, q);
3463   if (NULL == q->q)
3464   {
3465     LOG (GNUNET_ERROR_TYPE_DEBUG, "dropping msg on %s, NULL q\n", GCC_2s (c));
3466     GNUNET_free (data);
3467     GNUNET_free (q);
3468     GCC_check_connections ();
3469     return NULL;
3470   }
3471   q->cont = cont;
3472   q->cont_cls = cont_cls;
3473   GCC_check_connections ();
3474   return (NULL == cont) ? NULL : q;
3475 }
3476
3477
3478 /**
3479  * Cancel a previously sent message while it's in the queue.
3480  *
3481  * ONLY can be called before the continuation given to the send function
3482  * is called. Once the continuation is called, the message is no longer in the
3483  * queue.
3484  *
3485  * @param q Handle to the queue.
3486  */
3487 void
3488 GCC_cancel (struct CadetConnectionQueue *q)
3489 {
3490   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  GCC cancel message\n");
3491
3492   /* queue destroy calls message_sent, which calls q->cont and frees q */
3493   GCP_queue_destroy (q->q, GNUNET_YES, GNUNET_NO, 0);
3494   GCC_check_connections ();
3495 }
3496
3497
3498 /**
3499  * Sends a CREATE CONNECTION message for a path to a peer.
3500  * Changes the connection and tunnel states if necessary.
3501  *
3502  * @param connection Connection to create.
3503  */
3504 void
3505 GCC_send_create (struct CadetConnection *connection)
3506 {
3507   enum CadetTunnelCState state;
3508   size_t size;
3509
3510   GCC_check_connections ();
3511   size = sizeof (struct GNUNET_CADET_ConnectionCreate);
3512   size += connection->path->length * sizeof (struct GNUNET_PeerIdentity);
3513
3514   LOG (GNUNET_ERROR_TYPE_INFO, "---> %s on connection %s  (%u bytes)\n",
3515        GC_m2s (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE),
3516        GCC_2s (connection), size);
3517   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %p %u (create)\n",
3518        connection, connection->pending_messages);
3519   connection->pending_messages++;
3520
3521   connection->maintenance_q =
3522     GCP_queue_add (get_next_hop (connection), NULL,
3523                    GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE, 0, 0,
3524                    size, connection, GNUNET_YES, &conn_message_sent, NULL);
3525
3526   state = GCT_get_cstate (connection->t);
3527   if (CADET_TUNNEL_SEARCHING == state || CADET_TUNNEL_NEW == state)
3528     GCT_change_cstate (connection->t, CADET_TUNNEL_WAITING);
3529   if (CADET_CONNECTION_NEW == connection->state)
3530     connection_change_state (connection, CADET_CONNECTION_SENT);
3531   GCC_check_connections ();
3532 }
3533
3534
3535 /**
3536  * Send a message to all peers in this connection that the connection
3537  * is no longer valid.
3538  *
3539  * If some peer should not receive the message, it should be zero'ed out
3540  * before calling this function.
3541  *
3542  * @param c The connection whose peers to notify.
3543  */
3544 void
3545 GCC_send_destroy (struct CadetConnection *c)
3546 {
3547   struct GNUNET_CADET_ConnectionDestroy msg;
3548
3549   if (GNUNET_YES == c->destroy)
3550     return;
3551   GCC_check_connections ();
3552   msg.header.size = htons (sizeof (msg));
3553   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY);
3554   msg.cid = c->id;
3555   LOG (GNUNET_ERROR_TYPE_DEBUG,
3556               "  sending connection destroy for connection %s\n",
3557               GCC_2s (c));
3558
3559   if (GNUNET_NO == GCC_is_terminal (c, GNUNET_YES))
3560     GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg.header, 0, 0, c,
3561                                                       GNUNET_YES, GNUNET_YES,
3562                                                       NULL, NULL));
3563   if (GNUNET_NO == GCC_is_terminal (c, GNUNET_NO))
3564     GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg.header, 0, 0, c,
3565                                                       GNUNET_NO, GNUNET_YES,
3566                                                       NULL, NULL));
3567   c->destroy = GNUNET_YES;
3568   c->state = CADET_CONNECTION_DESTROYED;
3569   GCC_check_connections ();
3570 }
3571
3572
3573 /**
3574  * @brief Start a polling timer for the connection.
3575  *
3576  * When a neighbor does not accept more traffic on the connection it could be
3577  * caused by a simple congestion or by a lost ACK. Polling enables to check
3578  * for the lastest ACK status for a connection.
3579  *
3580  * @param c Connection.
3581  * @param fwd Should we poll in the FWD direction?
3582  */
3583 void
3584 GCC_start_poll (struct CadetConnection *c, int fwd)
3585 {
3586   struct CadetFlowControl *fc;
3587
3588   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3589   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL %s requested\n",
3590        GC_f2s (fwd));
3591   if (NULL != fc->poll_task || NULL != fc->poll_msg)
3592   {
3593     LOG (GNUNET_ERROR_TYPE_DEBUG, "  POLL not needed (%p, %p)\n",
3594          fc->poll_task, fc->poll_msg);
3595     return;
3596   }
3597   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL started on request\n");
3598   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
3599                                                 &connection_poll,
3600                                                 fc);
3601 }
3602
3603
3604 /**
3605  * @brief Stop polling a connection for ACKs.
3606  *
3607  * Once we have enough ACKs for future traffic, polls are no longer necessary.
3608  *
3609  * @param c Connection.
3610  * @param fwd Should we stop the poll in the FWD direction?
3611  */
3612 void
3613 GCC_stop_poll (struct CadetConnection *c, int fwd)
3614 {
3615   struct CadetFlowControl *fc;
3616
3617   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3618   if (NULL != fc->poll_task)
3619   {
3620     GNUNET_SCHEDULER_cancel (fc->poll_task);
3621     fc->poll_task = NULL;
3622   }
3623 }
3624
3625
3626 /**
3627  * Get a (static) string for a connection.
3628  *
3629  * @param c Connection.
3630  */
3631 const char *
3632 GCC_2s (const struct CadetConnection *c)
3633 {
3634   if (NULL == c)
3635     return "NULL";
3636
3637   if (NULL != c->t)
3638   {
3639     static char buf[128];
3640
3641     SPRINTF (buf, "%s (->%s)",
3642              GNUNET_h2s (GC_h2hc (GCC_get_id (c))), GCT_2s (c->t));
3643     return buf;
3644   }
3645   return GNUNET_h2s (GC_h2hc (&c->id));
3646 }
3647
3648
3649 /**
3650  * Log all possible info about the connection state.
3651  *
3652  * @param c Connection to debug.
3653  * @param level Debug level to use.
3654  */
3655 void
3656 GCC_debug (const struct CadetConnection *c, enum GNUNET_ErrorType level)
3657 {
3658   int do_log;
3659   char *s;
3660
3661   do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
3662                                        "cadet-con",
3663                                        __FILE__, __FUNCTION__, __LINE__);
3664   if (0 == do_log)
3665     return;
3666
3667   if (NULL == c)
3668   {
3669     LOG2 (level, "CCC DEBUG NULL CONNECTION\n");
3670     return;
3671   }
3672
3673   LOG2 (level, "CCC DEBUG CONNECTION %s\n", GCC_2s (c));
3674   s = path_2s (c->path);
3675   LOG2 (level, "CCC  path %s, own pos: %u\n", s, c->own_pos);
3676   GNUNET_free (s);
3677   LOG2 (level, "CCC  state: %s, destroy: %u\n",
3678         GCC_state2s (c->state), c->destroy);
3679   LOG2 (level, "CCC  pending messages: %u\n", c->pending_messages);
3680   if (NULL != c->perf)
3681     LOG2 (level, "CCC  us/byte: %f\n", c->perf->avg);
3682
3683   LOG2 (level, "CCC  FWD flow control:\n");
3684   LOG2 (level, "CCC   queue: %u/%u\n", c->fwd_fc.queue_n, c->fwd_fc.queue_max);
3685   LOG2 (level, "CCC   last PID sent: %5u, recv: %5u\n",
3686         c->fwd_fc.last_pid_sent, c->fwd_fc.last_pid_recv);
3687   LOG2 (level, "CCC   last ACK sent: %5u, recv: %5u\n",
3688         c->fwd_fc.last_ack_sent, c->fwd_fc.last_ack_recv);
3689   LOG2 (level, "CCC   recv PID bitmap: %X\n", c->fwd_fc.recv_bitmap);
3690   LOG2 (level, "CCC   poll: task %d, msg  %p, msg_ack %p)\n",
3691         c->fwd_fc.poll_task, c->fwd_fc.poll_msg, c->fwd_fc.ack_msg);
3692
3693   LOG2 (level, "CCC  BCK flow control:\n");
3694   LOG2 (level, "CCC   queue: %u/%u\n", c->bck_fc.queue_n, c->bck_fc.queue_max);
3695   LOG2 (level, "CCC   last PID sent: %5u, recv: %5u\n",
3696         c->bck_fc.last_pid_sent, c->bck_fc.last_pid_recv);
3697   LOG2 (level, "CCC   last ACK sent: %5u, recv: %5u\n",
3698         c->bck_fc.last_ack_sent, c->bck_fc.last_ack_recv);
3699   LOG2 (level, "CCC   recv PID bitmap: %X\n", c->bck_fc.recv_bitmap);
3700   LOG2 (level, "CCC   poll: task %d, msg  %p, msg_ack %p)\n",
3701         c->bck_fc.poll_task, c->bck_fc.poll_msg, c->bck_fc.ack_msg);
3702
3703   LOG2 (level, "CCC DEBUG CONNECTION END\n");
3704 }