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