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, 0, ack, c,
 593                                                 !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
 970   GCC_check_connections ();
 971   t = connection->t;
 972   LOG (GNUNET_ERROR_TYPE_INFO, "--> {%14s ACK} on conn %s\n",
 973        GC_f2s (!fwd), GCC_2s (connection));
 974   GCP_queue_add (get_hop (connection, fwd), NULL,
 975                  GNUNET_MESSAGE_TYPE_CADET_CONNECTION_ACK, 0, 0,
 976                  sizeof (struct GNUNET_CADET_ConnectionACK),
 977                  connection, fwd, &conn_message_sent, NULL);
 978   connection->pending_messages++;
 979   if (CADET_TUNNEL_NEW == GCT_get_cstate (t))
 980     GCT_change_cstate (t, CADET_TUNNEL_WAITING);
 981   if (CADET_CONNECTION_READY != connection->state)
 982     connection_change_state (connection, CADET_CONNECTION_SENT);
 983   GCC_check_connections ();
 984 }
 985
 986
 987 /**
 988  * Send a notification that a connection is broken.
 989  *
 990  * @param c Connection that is broken.
 991  * @param id1 Peer that has disconnected.
 992  * @param id2 Peer that has disconnected.
 993  * @param fwd Direction towards which to send it.
 994  */
 995 static void
 996 send_broken (struct CadetConnection *c,
 997              const struct GNUNET_PeerIdentity *id1,
 998              const struct GNUNET_PeerIdentity *id2,
 999              int fwd)
1000 {
1001   struct GNUNET_CADET_ConnectionBroken msg;
1002
1003   GCC_check_connections ();
1004   msg.header.size = htons (sizeof (struct GNUNET_CADET_ConnectionBroken));
1005   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
1006   msg.cid = c->id;
1007   msg.peer1 = *id1;
1008   msg.peer2 = *id2;
1009   GNUNET_assert (NULL ==
1010                  GCC_send_prebuilt_message (&msg.header, 0, 0, c, fwd,
1011                                             GNUNET_YES, NULL, NULL));
1012   GCC_check_connections ();
1013 }
1014
1015
1016 /**
1017  * Send a notification that a connection is broken, when a connection
1018  * isn't even known to the local peer or soon to be destroyed.
1019  *
1020  * @param connection_id Connection ID.
1021  * @param id1 Peer that has disconnected, probably local peer.
1022  * @param id2 Peer that has disconnected can be NULL if unknown.
1023  * @param peer Peer to notify (neighbor who sent the connection).
1024  */
1025 static void
1026 send_broken_unknown (const struct GNUNET_CADET_Hash *connection_id,
1027                      const struct GNUNET_PeerIdentity *id1,
1028                      const struct GNUNET_PeerIdentity *id2,
1029                      const struct GNUNET_PeerIdentity *peer_id)
1030 {
1031   struct GNUNET_CADET_ConnectionBroken *msg;
1032   struct CadetPeer *neighbor;
1033
1034   GCC_check_connections ();
1035   LOG (GNUNET_ERROR_TYPE_INFO, "--> BROKEN on unknown connection %s\n",
1036        GNUNET_h2s (GC_h2hc (connection_id)));
1037
1038   msg = GNUNET_new (struct GNUNET_CADET_ConnectionBroken);
1039   msg->header.size = htons (sizeof (struct GNUNET_CADET_ConnectionBroken));
1040   msg->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
1041   msg->cid = *connection_id;
1042   msg->peer1 = *id1;
1043   if (NULL != id2)
1044     msg->peer2 = *id2;
1045   else
1046     memset (&msg->peer2, 0, sizeof (msg->peer2));
1047   neighbor = GCP_get (peer_id, GNUNET_NO); /* We MUST know neighbor. */
1048   GNUNET_assert (NULL != neighbor);
1049   GCP_queue_add (neighbor, msg, GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN,
1050                  0, 2, sizeof (struct GNUNET_CADET_ConnectionBroken),
1051                  NULL, GNUNET_SYSERR, /* connection, fwd */
1052                  NULL, NULL); /* continuation */
1053   GCC_check_connections ();
1054 }
1055
1056
1057 /**
1058  * Send keepalive packets for a connection.
1059  *
1060  * @param c Connection to keep alive..
1061  * @param fwd Is this a FWD keepalive? (owner -> dest).
1062  */
1063 static void
1064 send_connection_keepalive (struct CadetConnection *c, int fwd)
1065 {
1066   struct GNUNET_MessageHeader msg;
1067   struct CadetFlowControl *fc;
1068
1069   GCC_check_connections ();
1070   LOG (GNUNET_ERROR_TYPE_INFO,
1071        "keepalive %s for connection %s\n",
1072        GC_f2s (fwd), GCC_2s (c));
1073
1074   GNUNET_assert (NULL != c->t);
1075   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1076   if (0 < fc->queue_n || GNUNET_YES == GCT_has_queued_traffic (c->t))
1077   {
1078     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, traffic in queue\n");
1079     return;
1080   }
1081
1082   GNUNET_STATISTICS_update (stats, "# keepalives sent", 1, GNUNET_NO);
1083
1084   GNUNET_assert (NULL != c->t);
1085   msg.size = htons (sizeof (msg));
1086   msg.type = htons (GNUNET_MESSAGE_TYPE_CADET_KEEPALIVE);
1087
1088   GNUNET_assert (NULL ==
1089                  GCT_send_prebuilt_message (&msg, c->t, c,
1090                                             GNUNET_NO, NULL, NULL));
1091   GCC_check_connections ();
1092 }
1093
1094
1095 /**
1096  * Send CONNECTION_{CREATE/ACK} packets for a connection.
1097  *
1098  * @param c Connection for which to send the message.
1099  * @param fwd If #GNUNET_YES, send CREATE, otherwise send ACK.
1100  */
1101 static void
1102 connection_recreate (struct CadetConnection *c, int fwd)
1103 {
1104   LOG (GNUNET_ERROR_TYPE_DEBUG,
1105        "sending connection recreate\n");
1106   if (fwd)
1107     GCC_send_create (c);
1108   else
1109     send_connection_ack (c, GNUNET_NO);
1110 }
1111
1112
1113 /**
1114  * Generic connection timer management.
1115  * Depending on the role of the peer in the connection will send the
1116  * appropriate message (build or keepalive)
1117  *
1118  * @param c Conncetion to maintain.
1119  * @param fwd Is FWD?
1120  */
1121 static void
1122 connection_maintain (struct CadetConnection *c, int fwd)
1123 {
1124   if (GNUNET_NO != c->destroy)
1125   {
1126     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, being destroyed\n");
1127     return;
1128   }
1129
1130   if (NULL == c->t)
1131   {
1132     GNUNET_break (0);
1133     GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
1134     return;
1135   }
1136
1137   if (CADET_TUNNEL_SEARCHING == GCT_get_cstate (c->t))
1138   {
1139     /* If status is SEARCHING, why is there a connection? Should be WAITING */
1140     GNUNET_break (0);
1141     GCT_debug (c->t, GNUNET_ERROR_TYPE_ERROR);
1142     LOG (GNUNET_ERROR_TYPE_INFO, "not sending keepalive, tunnel SEARCHING\n");
1143     schedule_next_keepalive (c, fwd);
1144     return;
1145   }
1146   switch (c->state)
1147   {
1148     case CADET_CONNECTION_NEW:
1149       GNUNET_break (0);
1150       /* fall-through */
1151     case CADET_CONNECTION_SENT:
1152       connection_recreate (c, fwd);
1153       break;
1154     case CADET_CONNECTION_READY:
1155       send_connection_keepalive (c, fwd);
1156       break;
1157     default:
1158       break;
1159   }
1160 }
1161
1162
1163 /**
1164  * Keep the connection alive.
1165  *
1166  * @param c Connection to keep alive.
1167  * @param fwd Direction.
1168  * @param shutdown Are we shutting down? (Don't send traffic)
1169  *                 Non-zero value for true, not necessarily GNUNET_YES.
1170  */
1171 static void
1172 connection_keepalive (struct CadetConnection *c, int fwd, int shutdown)
1173 {
1174   GCC_check_connections ();
1175   LOG (GNUNET_ERROR_TYPE_DEBUG,
1176        "%s keepalive for %s\n",
1177        GC_f2s (fwd), GCC_2s (c));
1178
1179   if (fwd)
1180     c->fwd_maintenance_task = NULL;
1181   else
1182     c->bck_maintenance_task = NULL;
1183
1184   if (GNUNET_NO != shutdown)
1185     return;
1186
1187   connection_maintain (c, fwd);
1188   GCC_check_connections ();
1189   /* Next execution will be scheduled by message_sent or _maintain*/
1190 }
1191
1192
1193 /**
1194  * Keep the connection alive in the FWD direction.
1195  *
1196  * @param cls Closure (connection to keepalive).
1197  * @param tc TaskContext.
1198  */
1199 static void
1200 connection_fwd_keepalive (void *cls,
1201                           const struct GNUNET_SCHEDULER_TaskContext *tc)
1202 {
1203   GCC_check_connections ();
1204   connection_keepalive ((struct CadetConnection *) cls,
1205                         GNUNET_YES,
1206                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
1207   GCC_check_connections ();
1208 }
1209
1210
1211 /**
1212  * Keep the connection alive in the BCK direction.
1213  *
1214  * @param cls Closure (connection to keepalive).
1215  * @param tc TaskContext.
1216  */
1217 static void
1218 connection_bck_keepalive (void *cls,
1219                           const struct GNUNET_SCHEDULER_TaskContext *tc)
1220 {
1221   GCC_check_connections ();
1222   connection_keepalive ((struct CadetConnection *) cls,
1223                         GNUNET_NO,
1224                         tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN);
1225   GCC_check_connections ();
1226 }
1227
1228
1229 /**
1230  * Schedule next keepalive task, taking in consideration
1231  * the connection state and number of retries.
1232  *
1233  * If the peer is not the origin, do nothing.
1234  *
1235  * @param c Connection for which to schedule the next keepalive.
1236  * @param fwd Direction for the next keepalive.
1237  */
1238 static void
1239 schedule_next_keepalive (struct CadetConnection *c, int fwd)
1240 {
1241   struct GNUNET_TIME_Relative delay;
1242   struct GNUNET_SCHEDULER_Task * *task_id;
1243   GNUNET_SCHEDULER_TaskCallback keepalive_task;
1244
1245   GCC_check_connections ();
1246   if (GNUNET_NO == GCC_is_origin (c, fwd))
1247     return;
1248
1249   /* Calculate delay to use, depending on the state of the connection */
1250   if (CADET_CONNECTION_READY == c->state)
1251   {
1252     delay = refresh_connection_time;
1253   }
1254   else
1255   {
1256     if (1 > c->create_retry)
1257       c->create_retry = 1;
1258     delay = GNUNET_TIME_relative_multiply (create_connection_time,
1259                                            c->create_retry);
1260     if (c->create_retry < 64)
1261       c->create_retry *= 2;
1262   }
1263
1264   /* Select direction-dependent parameters */
1265   if (GNUNET_YES == fwd)
1266   {
1267     task_id = &c->fwd_maintenance_task;
1268     keepalive_task = &connection_fwd_keepalive;
1269   }
1270   else
1271   {
1272     task_id = &c->bck_maintenance_task;
1273     keepalive_task = &connection_bck_keepalive;
1274   }
1275
1276   /* Check that no one scheduled it before us */
1277   if (NULL != *task_id)
1278   {
1279     /* No need for a _break. It can happen for instance when sending a SYNACK
1280      * for a duplicate SYN: the first SYNACK scheduled the task. */
1281     GNUNET_SCHEDULER_cancel (*task_id);
1282   }
1283
1284   /* Schedule the task */
1285   *task_id = GNUNET_SCHEDULER_add_delayed (delay,
1286                                            keepalive_task,
1287                                            c);
1288   LOG (GNUNET_ERROR_TYPE_DEBUG, "next keepalive in %s\n",
1289        GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
1290   GCC_check_connections ();
1291 }
1292
1293
1294 /**
1295  * @brief Re-initiate traffic on this connection if necessary.
1296  *
1297  * Check if there is traffic queued towards this peer
1298  * and the core transmit handle is NULL (traffic was stalled).
1299  * If so, call core tmt rdy.
1300  *
1301  * @param c Connection on which initiate traffic.
1302  * @param fwd Is this about fwd traffic?
1303  */
1304 static void
1305 connection_unlock_queue (struct CadetConnection *c, int fwd)
1306 {
1307   struct CadetPeer *peer;
1308
1309   GCC_check_connections ();
1310   LOG (GNUNET_ERROR_TYPE_DEBUG,
1311        "connection_unlock_queue %s on %s\n",
1312        GC_f2s (fwd), GCC_2s (c));
1313
1314   if (GCC_is_terminal (c, fwd))
1315   {
1316     LOG (GNUNET_ERROR_TYPE_DEBUG, " is terminal, can unlock!\n");
1317     return;
1318   }
1319
1320   peer = get_hop (c, fwd);
1321   GCP_queue_unlock (peer, c);
1322   GCC_check_connections ();
1323 }
1324
1325
1326 /**
1327  * Cancel all transmissions that belong to a certain connection.
1328  *
1329  * If the connection is scheduled for destruction and no more messages are left,
1330  * the connection will be destroyed by the continuation call.
1331  *
1332  * @param c Connection which to cancel. Might be destroyed during this call.
1333  * @param fwd Cancel fwd traffic?
1334  */
1335 static void
1336 connection_cancel_queues (struct CadetConnection *c,
1337                           int fwd)
1338 {
1339   struct CadetFlowControl *fc;
1340   struct CadetPeer *peer;
1341
1342   GCC_check_connections ();
1343   LOG (GNUNET_ERROR_TYPE_DEBUG,
1344        "Cancel %s queues for connection %s\n",
1345        GC_f2s (fwd), GCC_2s (c));
1346   if (NULL == c)
1347   {
1348     GNUNET_break (0);
1349     return;
1350   }
1351
1352   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1353   if (NULL != fc->poll_task)
1354   {
1355     GNUNET_SCHEDULER_cancel (fc->poll_task);
1356     fc->poll_task = NULL;
1357     LOG (GNUNET_ERROR_TYPE_DEBUG, "  cancelled POLL task for fc %p\n", fc);
1358   }
1359   if (NULL != fc->poll_msg)
1360   {
1361     GCC_cancel (fc->poll_msg);
1362     LOG (GNUNET_ERROR_TYPE_DEBUG, "  cancelled POLL msg for fc %p\n", fc);
1363   }
1364   peer = get_hop (c, fwd);
1365   GCP_queue_cancel (peer, c);
1366   GCC_check_connections ();
1367 }
1368
1369
1370 /**
1371  * Function called if a connection has been stalled for a while,
1372  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1373  *
1374  * @param cls Closure (poll ctx).
1375  * @param tc TaskContext.
1376  */
1377 static void
1378 connection_poll (void *cls,
1379                  const struct GNUNET_SCHEDULER_TaskContext *tc);
1380
1381
1382 /**
1383  * Callback called when a queued POLL message is sent.
1384  *
1385  * @param cls Closure (FC).
1386  * @param c Connection this message was on.
1387  * @param q Queue handler this call invalidates.
1388  * @param type Type of message sent.
1389  * @param fwd Was this a FWD going message?
1390  * @param size Size of the message.
1391  */
1392 static void
1393 poll_sent (void *cls,
1394            struct CadetConnection *c,
1395            struct CadetConnectionQueue *q,
1396            uint16_t type, int fwd, size_t size)
1397 {
1398   struct CadetFlowControl *fc = cls;
1399
1400   if (2 == c->destroy)
1401   {
1402     LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL canceled on shutdown\n");
1403     return;
1404   }
1405   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL sent for %s, scheduling new one!\n",
1406        GCC_2s (c));
1407   fc->poll_msg = NULL;
1408   fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
1409   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
1410                                                 &connection_poll,
1411                                                 fc);
1412   LOG (GNUNET_ERROR_TYPE_DEBUG, " task %u\n", fc->poll_task);
1413 }
1414
1415
1416 /**
1417  * Function called if a connection has been stalled for a while,
1418  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1419  *
1420  * @param cls Closure (poll ctx).
1421  * @param tc TaskContext.
1422  */
1423 static void
1424 connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1425 {
1426   struct CadetFlowControl *fc = cls;
1427   struct GNUNET_CADET_Poll msg;
1428   struct CadetConnection *c;
1429   int fwd;
1430
1431   fc->poll_task = NULL;
1432   GCC_check_connections ();
1433   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1434   {
1435     return;
1436   }
1437
1438   c = fc->c;
1439   fwd = fc == &c->fwd_fc;
1440   LOG (GNUNET_ERROR_TYPE_DEBUG, "Polling connection %s %s\n",
1441        GCC_2s (c),  GC_f2s (fwd));
1442
1443   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_POLL);
1444   msg.header.size = htons (sizeof (msg));
1445   msg.pid = htonl (fc->last_pid_sent);
1446   LOG (GNUNET_ERROR_TYPE_DEBUG, " last pid sent: %u\n", fc->last_pid_sent);
1447   fc->poll_msg =
1448       GCC_send_prebuilt_message (&msg.header, 0, fc->last_pid_sent, c,
1449                                  fc == &c->fwd_fc, GNUNET_YES, &poll_sent, fc);
1450   GNUNET_assert (NULL != fc->poll_msg);
1451   GCC_check_connections ();
1452 }
1453
1454
1455 /**
1456  * Resend all queued messages for a connection on other connections of the
1457  * same tunnel, if possible. The connection WILL BE DESTROYED by this function.
1458  *
1459  * @param c Connection whose messages to resend.
1460  * @param fwd Resend fwd messages?
1461  */
1462 static void
1463 resend_messages_and_destroy (struct CadetConnection *c, int fwd)
1464 {
1465   struct GNUNET_MessageHeader *out_msg;
1466   struct CadetTunnel *t = c->t;
1467   struct CadetPeer *neighbor;
1468   unsigned int pending;
1469   int destroyed;
1470
1471   GCC_check_connections ();
1472   c->state = CADET_CONNECTION_DESTROYED;
1473   c->destroy = GNUNET_YES;
1474
1475   destroyed = GNUNET_NO;
1476   neighbor = get_hop (c, fwd);
1477   pending = c->pending_messages;
1478
1479   while (NULL != (out_msg = GCP_connection_pop (neighbor, c, &destroyed)))
1480   {
1481     if (NULL != t)
1482       GCT_resend_message (out_msg, t);
1483     GNUNET_free (out_msg);
1484   }
1485
1486   /* All pending messages should have been popped,
1487    * and the connection destroyed by the continuation.
1488    */
1489   if (GNUNET_YES != destroyed)
1490   {
1491     if (0 != pending)
1492     {
1493       GNUNET_break (0);
1494       GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
1495       if (NULL != t) GCT_debug (t, GNUNET_ERROR_TYPE_ERROR);
1496     }
1497     GCC_destroy (c);
1498   }
1499   GCC_check_connections ();
1500 }
1501
1502
1503 /**
1504  * Generic connection timeout implementation.
1505  *
1506  * Timeout function due to lack of keepalive/traffic from an endpoint.
1507  * Destroys connection if called.
1508  *
1509  * @param c Connection to destroy.
1510  * @param fwd Was the timeout from the origin? (FWD timeout)
1511  */
1512 static void
1513 connection_timeout (struct CadetConnection *c, int fwd)
1514 {
1515   struct CadetFlowControl *reverse_fc;
1516
1517   GCC_check_connections ();
1518   reverse_fc = fwd ? &c->bck_fc : &c->fwd_fc;
1519
1520   LOG (GNUNET_ERROR_TYPE_INFO,
1521        "Connection %s %s timed out. Destroying.\n",
1522        GCC_2s (c),
1523        GC_f2s (fwd));
1524   GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
1525
1526   if (GCC_is_origin (c, fwd)) /* Loopback? Something is wrong! */
1527   {
1528     GNUNET_break (0);
1529     return;
1530   }
1531
1532   /* If dest, salvage queued traffic. */
1533   if (GCC_is_origin (c, !fwd))
1534   {
1535     const struct GNUNET_PeerIdentity *next_hop;
1536
1537     next_hop = GCP_get_id (fwd ? get_prev_hop (c) : get_next_hop (c));
1538     send_broken_unknown (&c->id, &my_full_id, NULL, next_hop);
1539     if (0 < reverse_fc->queue_n)
1540       resend_messages_and_destroy (c, !fwd);
1541     GCC_check_connections ();
1542     return;
1543   }
1544
1545   GCC_destroy (c);
1546   GCC_check_connections ();
1547 }
1548
1549
1550 /**
1551  * Timeout function due to lack of keepalive/traffic from the owner.
1552  * Destroys connection if called.
1553  *
1554  * @param cls Closure (connection to destroy).
1555  * @param tc TaskContext.
1556  */
1557 static void
1558 connection_fwd_timeout (void *cls,
1559                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1560 {
1561   struct CadetConnection *c = cls;
1562
1563   c->fwd_maintenance_task = NULL;
1564   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1565     return;
1566   GCC_check_connections ();
1567   connection_timeout (c, GNUNET_YES);
1568   GCC_check_connections ();
1569 }
1570
1571
1572 /**
1573  * Timeout function due to lack of keepalive/traffic from the destination.
1574  * Destroys connection if called.
1575  *
1576  * @param cls Closure (connection to destroy).
1577  * @param tc TaskContext
1578  */
1579 static void
1580 connection_bck_timeout (void *cls,
1581                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1582 {
1583   struct CadetConnection *c = cls;
1584
1585   c->bck_maintenance_task = NULL;
1586   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1587     return;
1588   GCC_check_connections ();
1589   connection_timeout (c, GNUNET_NO);
1590   GCC_check_connections ();
1591 }
1592
1593
1594 /**
1595  * Resets the connection timeout task, some other message has done the
1596  * task's job.
1597  * - For the first peer on the direction this means to send
1598  *   a keepalive or a path confirmation message (either create or ACK).
1599  * - For all other peers, this means to destroy the connection,
1600  *   due to lack of activity.
1601  * Starts the timeout if no timeout was running (connection just created).
1602  *
1603  * @param c Connection whose timeout to reset.
1604  * @param fwd Is this forward?
1605  *
1606  * TODO use heap to improve efficiency of scheduler.
1607  */
1608 static void
1609 connection_reset_timeout (struct CadetConnection *c, int fwd)
1610 {
1611   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connection %s reset timeout\n", GC_f2s (fwd));
1612   if (GCC_is_origin (c, fwd)) /* Startpoint */
1613   {
1614     schedule_next_keepalive (c, fwd);
1615   }
1616   else /* Relay, endpoint. */
1617   {
1618     struct GNUNET_TIME_Relative delay;
1619     struct GNUNET_SCHEDULER_Task * *ti;
1620     GNUNET_SCHEDULER_TaskCallback f;
1621
1622     ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
1623
1624     if (NULL != *ti)
1625       GNUNET_SCHEDULER_cancel (*ti);
1626     delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
1627     LOG (GNUNET_ERROR_TYPE_DEBUG, "  timing out in %s\n",
1628          GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_NO));
1629     f = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
1630     *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
1631   }
1632 }
1633
1634
1635 /**
1636  * Iterator to compare each connection's path with the path of a new connection.
1637  *
1638  * If the connection conincides, the c member of path is set to the connection
1639  * and the destroy flag of the connection is set.
1640  *
1641  * @param cls Closure (new path).
1642  * @param c Connection in the tunnel to check.
1643  */
1644 static void
1645 check_path (void *cls, struct CadetConnection *c)
1646 {
1647   struct CadetConnection *new_conn = cls;
1648   struct CadetPeerPath *path = new_conn->path;
1649
1650   LOG (GNUNET_ERROR_TYPE_DEBUG, "  checking %s (%p), length %u\n",
1651        GCC_2s (c), c, c->path->length);
1652
1653   if (c != new_conn
1654       && c->destroy == GNUNET_NO
1655       && c->state != CADET_CONNECTION_BROKEN
1656       && c->state != CADET_CONNECTION_DESTROYED
1657       && path_equivalent (path, c->path))
1658   {
1659     new_conn->destroy = GNUNET_YES;
1660     new_conn->path->c = c;
1661     LOG (GNUNET_ERROR_TYPE_DEBUG, "  MATCH!\n");
1662   }
1663 }
1664
1665
1666 /**
1667  * Finds out if this path is already being used by an existing connection.
1668  *
1669  * Checks the tunnel towards the destination to see if it contains
1670  * any connection with the same path.
1671  *
1672  * If the existing connection is ready, it is kept.
1673  * Otherwise if the sender has a smaller ID that ours, we accept it (and
1674  * the peer will eventually reject our attempt).
1675  *
1676  * @param path Path to check.
1677  * @return #GNUNET_YES if the tunnel has a connection with the same path,
1678  *         #GNUNET_NO otherwise.
1679  */
1680 static int
1681 does_connection_exist (struct CadetConnection *conn)
1682 {
1683   struct CadetPeer *p;
1684   struct CadetTunnel *t;
1685   struct CadetConnection *c;
1686
1687   p = GCP_get_short (conn->path->peers[0], GNUNET_NO);
1688   if (NULL == p)
1689     return GNUNET_NO;
1690   t = GCP_get_tunnel (p);
1691   if (NULL == t)
1692     return GNUNET_NO;
1693
1694   LOG (GNUNET_ERROR_TYPE_DEBUG, "Checking for duplicates\n");
1695
1696   GCT_iterate_connections (t, &check_path, conn);
1697
1698   if (GNUNET_YES == conn->destroy)
1699   {
1700     c = conn->path->c;
1701     conn->destroy = GNUNET_NO;
1702     conn->path->c = conn;
1703     LOG (GNUNET_ERROR_TYPE_DEBUG, " found duplicate of %s\n", GCC_2s (conn));
1704     LOG (GNUNET_ERROR_TYPE_DEBUG, " duplicate: %s\n", GCC_2s (c));
1705     GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
1706     if (CADET_CONNECTION_READY == c->state)
1707     {
1708       /* The other peer confirmed a live connection with this path,
1709        * why is it trying to duplicate it. */
1710       GNUNET_STATISTICS_update (stats, "# duplicate connections", 1, GNUNET_NO);
1711       return GNUNET_YES;
1712     }
1713     LOG (GNUNET_ERROR_TYPE_DEBUG, " duplicate not valid, connection unique\n");
1714     return GNUNET_NO;
1715   }
1716   else
1717   {
1718     LOG (GNUNET_ERROR_TYPE_DEBUG, " %s has no duplicates\n", GCC_2s (conn));
1719     return GNUNET_NO;
1720   }
1721 }
1722
1723
1724 /**
1725  * @brief Check if the tunnel this connection belongs to has any other
1726  * connection with the same path, and destroy one if so.
1727  *
1728  * @param cls Closure (connection to check).
1729  * @param tc Task context.
1730  */
1731 static void
1732 check_duplicates (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1733 {
1734   struct CadetConnection *c = cls;
1735
1736   c->check_duplicates_task = NULL;
1737   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1738     return;
1739
1740   if (GNUNET_YES == does_connection_exist (c))
1741   {
1742     GCT_debug (c->t, GNUNET_ERROR_TYPE_DEBUG);
1743     send_broken (c, &my_full_id, &my_full_id, GCC_is_origin (c, GNUNET_YES));
1744     GCC_destroy (c);
1745   }
1746 }
1747
1748
1749
1750 /**
1751  * Wait for enough time to let any dead connections time out and check for
1752  * any remaining duplicates.
1753  *
1754  * @param c Connection that is a potential duplicate.
1755  */
1756 static void
1757 schedule_check_duplicates (struct CadetConnection *c)
1758 {
1759   struct GNUNET_TIME_Relative delay;
1760
1761   if (NULL != c->check_duplicates_task)
1762     return;
1763
1764   delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 5);
1765   c->check_duplicates_task = GNUNET_SCHEDULER_add_delayed (delay,
1766                                                            &check_duplicates,
1767                                                            c);
1768 }
1769
1770
1771
1772 /**
1773  * Add the connection to the list of both neighbors.
1774  *
1775  * @param c Connection.
1776  *
1777  * @return #GNUNET_OK if everything went fine
1778  *         #GNUNET_SYSERR if the was an error and @c c is malformed.
1779  */
1780 static int
1781 register_neighbors (struct CadetConnection *c)
1782 {
1783   c->next_peer = get_next_hop (c);
1784   c->prev_peer = get_prev_hop (c);
1785   GNUNET_assert (c->next_peer != c->prev_peer);
1786   LOG (GNUNET_ERROR_TYPE_DEBUG,
1787        "register neighbors for connection %s\n",
1788        GCC_2s (c));
1789   path_debug (c->path);
1790   LOG (GNUNET_ERROR_TYPE_DEBUG,
1791        "own pos %u\n", c->own_pos);
1792   LOG (GNUNET_ERROR_TYPE_DEBUG,
1793        "putting connection %s to next peer %p\n",
1794        GCC_2s (c),
1795        c->next_peer);
1796   LOG (GNUNET_ERROR_TYPE_DEBUG, "next peer %p %s\n",
1797        c->next_peer,
1798        GCP_2s (c->next_peer));
1799   LOG (GNUNET_ERROR_TYPE_DEBUG,
1800        "putting connection %s to prev peer %p\n",
1801        GCC_2s (c),
1802        c->prev_peer);
1803   LOG (GNUNET_ERROR_TYPE_DEBUG,
1804        "prev peer %p %s\n",
1805        c->prev_peer,
1806        GCP_2s (c->prev_peer));
1807
1808   if ( (GNUNET_NO == GCP_is_neighbor (c->next_peer)) ||
1809        (GNUNET_NO == GCP_is_neighbor (c->prev_peer)) )
1810   {
1811     if (GCC_is_origin (c, GNUNET_YES))
1812       GNUNET_STATISTICS_update (stats, "# local bad paths", 1, GNUNET_NO);
1813     GNUNET_STATISTICS_update (stats, "# bad paths", 1, GNUNET_NO);
1814
1815     LOG (GNUNET_ERROR_TYPE_DEBUG,
1816          "  register neighbors failed\n");
1817     LOG (GNUNET_ERROR_TYPE_DEBUG,
1818          "  prev: %s, neighbor?: %d\n",
1819          GCP_2s (c->prev_peer),
1820          GCP_is_neighbor (c->prev_peer));
1821     LOG (GNUNET_ERROR_TYPE_DEBUG,
1822          "  next: %s, neighbor?: %d\n",
1823          GCP_2s (c->next_peer),
1824          GCP_is_neighbor (c->next_peer));
1825     return GNUNET_SYSERR;
1826   }
1827   GCP_add_connection (c->next_peer, c, GNUNET_NO);
1828   GCP_add_connection (c->prev_peer, c, GNUNET_YES);
1829
1830   return GNUNET_OK;
1831 }
1832
1833
1834 /**
1835  * Remove the connection from the list of both neighbors.
1836  *
1837  * @param c Connection.
1838  */
1839 static void
1840 unregister_neighbors (struct CadetConnection *c)
1841 {
1842 //  struct CadetPeer *peer; FIXME dont use next_peer, prev_peer
1843   /* Either already unregistered or never got registered, it's ok either way. */
1844   if (NULL == c->path)
1845     return;
1846   if (NULL != c->next_peer)
1847   {
1848     GCP_remove_connection (c->next_peer, c);
1849     c->next_peer = NULL;
1850   }
1851   if (NULL != c->prev_peer)
1852   {
1853     GCP_remove_connection (c->prev_peer, c);
1854     c->prev_peer = NULL;
1855   }
1856 }
1857
1858
1859 /**
1860  * Invalidates all paths towards all peers that comprise the connection which
1861  * rely on the disconnected peer.
1862  *
1863  * ~O(n^3) (peers in connection * paths/peer * links/path)
1864  *
1865  * @param c Connection whose peers' paths to clean.
1866  * @param disconnected Peer that disconnected.
1867  */
1868 static void
1869 invalidate_paths (struct CadetConnection *c, struct CadetPeer *disconnected)
1870 {
1871   struct CadetPeer *peer;
1872   unsigned int i;
1873
1874   for (i = 0; i < c->path->length; i++)
1875   {
1876     peer = GCP_get_short (c->path->peers[i], GNUNET_NO);
1877     if (NULL != peer)
1878       GCP_notify_broken_link (peer, &my_full_id, GCP_get_id (disconnected));
1879   }
1880 }
1881
1882
1883 /**
1884  * Bind the connection to the peer and the tunnel to that peer.
1885  *
1886  * If the peer has no tunnel, create one. Update tunnel and connection
1887  * data structres to reflect new status.
1888  *
1889  * @param c Connection.
1890  * @param peer Peer.
1891  */
1892 static void
1893 add_to_peer (struct CadetConnection *c,
1894              struct CadetPeer *peer)
1895 {
1896   GCP_add_tunnel (peer);
1897   c->t = GCP_get_tunnel (peer);
1898   GCT_add_connection (c->t, c);
1899 }
1900
1901
1902 /**
1903  * Log receipt of message on stderr (INFO level).
1904  *
1905  * @param message Message received.
1906  * @param peer Peer who sent the message.
1907  * @param hash Connection ID.
1908  */
1909 static void
1910 log_message (const struct GNUNET_MessageHeader *message,
1911              const struct GNUNET_PeerIdentity *peer,
1912              const struct GNUNET_CADET_Hash *hash)
1913 {
1914   uint16_t size;
1915
1916   size = ntohs (message->size);
1917   LOG (GNUNET_ERROR_TYPE_INFO, "\n");
1918   LOG (GNUNET_ERROR_TYPE_INFO, "\n");
1919   LOG (GNUNET_ERROR_TYPE_INFO, "<-- %s on conn %s from %s, %6u bytes\n",
1920        GC_m2s (ntohs (message->type)), GNUNET_h2s (GC_h2hc (hash)),
1921        GNUNET_i2s (peer), (unsigned int) size);
1922 }
1923
1924 /******************************************************************************/
1925 /********************************    API    ***********************************/
1926 /******************************************************************************/
1927
1928 /**
1929  * Core handler for connection creation.
1930  *
1931  * @param cls Closure (unused).
1932  * @param peer Sender (neighbor).
1933  * @param message Message.
1934  * @return #GNUNET_OK to keep the connection open,
1935  *         #GNUNET_SYSERR to close it (signal serious error)
1936  */
1937 int
1938 GCC_handle_create (void *cls,
1939                    const struct GNUNET_PeerIdentity *peer,
1940                    const struct GNUNET_MessageHeader *message)
1941 {
1942   struct GNUNET_CADET_ConnectionCreate *msg;
1943   struct GNUNET_PeerIdentity *id;
1944   struct GNUNET_CADET_Hash *cid;
1945   struct CadetPeerPath *path;
1946   struct CadetPeer *dest_peer;
1947   struct CadetPeer *orig_peer;
1948   struct CadetConnection *c;
1949   unsigned int own_pos;
1950   uint16_t size;
1951
1952   GCC_check_connections ();
1953   /* Check size */
1954   size = ntohs (message->size);
1955   if (size < sizeof (struct GNUNET_CADET_ConnectionCreate))
1956   {
1957     GNUNET_break_op (0);
1958     return GNUNET_OK;
1959   }
1960
1961   /* Calculate hops */
1962   size -= sizeof (struct GNUNET_CADET_ConnectionCreate);
1963   if (size % sizeof (struct GNUNET_PeerIdentity))
1964   {
1965     GNUNET_break_op (0);
1966     return GNUNET_OK;
1967   }
1968   if (0 != size % sizeof (struct GNUNET_PeerIdentity))
1969   {
1970     GNUNET_break_op (0);
1971     return GNUNET_OK;
1972   }
1973   size /= sizeof (struct GNUNET_PeerIdentity);
1974   if (1 > size)
1975   {
1976     GNUNET_break_op (0);
1977     return GNUNET_OK;
1978   }
1979   LOG (GNUNET_ERROR_TYPE_DEBUG, "    path has %u hops.\n", size);
1980
1981   /* Get parameters */
1982   msg = (struct GNUNET_CADET_ConnectionCreate *) message;
1983   cid = &msg->cid;
1984   log_message (message, peer, cid);
1985   id = (struct GNUNET_PeerIdentity *) &msg[1];
1986   LOG (GNUNET_ERROR_TYPE_DEBUG, "    origin: %s\n", GNUNET_i2s (id));
1987
1988   /* Create connection */
1989   c = connection_get (cid);
1990   if (NULL == c)
1991   {
1992     path = path_build_from_peer_ids ((struct GNUNET_PeerIdentity *) &msg[1],
1993                                      size, myid, &own_pos);
1994     if (NULL == path)
1995     {
1996       /* Path was malformed, probably our own ID was not in it. */
1997       GNUNET_STATISTICS_update (stats, "# malformed paths", 1, GNUNET_NO);
1998       GNUNET_break_op (0);
1999       return GNUNET_OK;
2000     }
2001
2002     if (0 == own_pos)
2003     {
2004       /* We received this request from a neighbor, we cannot be origin */
2005       GNUNET_STATISTICS_update (stats, "# fake paths", 1, GNUNET_NO);
2006       GNUNET_break_op (0);
2007       path_destroy (path);
2008       return GNUNET_OK;
2009     }
2010
2011     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Own position: %u\n", own_pos);
2012     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Creating connection\n");
2013     c = GCC_new (cid, NULL, path, own_pos);
2014     if (NULL == c)
2015     {
2016       if (path->length - 1 == own_pos)
2017       {
2018         /* If we are destination, why did the creation fail? */
2019         GNUNET_break (0);
2020         path_destroy (path);
2021         GCC_check_connections ();
2022         return GNUNET_OK;
2023       }
2024       send_broken_unknown (cid, &my_full_id,
2025                            GNUNET_PEER_resolve2 (path->peers[own_pos + 1]),
2026                            peer);
2027       path_destroy (path);
2028       GCC_check_connections ();
2029       return GNUNET_OK;
2030     }
2031     GCP_add_path_to_all (path, GNUNET_NO);
2032     connection_reset_timeout (c, GNUNET_YES);
2033   }
2034   else
2035   {
2036     path = path_duplicate (c->path);
2037   }
2038   if (CADET_CONNECTION_NEW == c->state)
2039     connection_change_state (c, CADET_CONNECTION_SENT);
2040
2041   /* Remember peers */
2042   dest_peer = GCP_get (&id[size - 1], GNUNET_YES);
2043   orig_peer = GCP_get (&id[0], GNUNET_YES);
2044
2045   /* Is it a connection to us? */
2046   if (c->own_pos == path->length - 1)
2047   {
2048     LOG (GNUNET_ERROR_TYPE_DEBUG, "  It's for us!\n");
2049     GCP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_YES);
2050
2051     add_to_peer (c, orig_peer);
2052     if (GNUNET_YES == does_connection_exist (c))
2053     {
2054       /* Peer created a connection equal to one we think exists
2055        * and is fine.
2056        * Solution: Keep both and postpone disambiguation. In the meantime
2057        * the connection will time out or peer will inform us it is broken.
2058        *
2059        * Other options:
2060        * - Use explicit duplicate.
2061        * - Accept new conn and destroy the old. (interruption in higher level)
2062        * - Keep the one with higher ID / created by peer with higher ID. */
2063        schedule_check_duplicates (c);
2064     }
2065
2066     if (CADET_TUNNEL_NEW == GCT_get_cstate (c->t))
2067       GCT_change_cstate (c->t,  CADET_TUNNEL_WAITING);
2068
2069     send_connection_ack (c, GNUNET_NO);
2070     if (CADET_CONNECTION_SENT == c->state)
2071       connection_change_state (c, CADET_CONNECTION_ACK);
2072   }
2073   else
2074   {
2075     /* It's for somebody else! Retransmit. */
2076     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Retransmitting.\n");
2077     GCP_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
2078     GCP_add_path_to_origin (orig_peer, path_duplicate (path), GNUNET_NO);
2079     GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c,
2080                                                       GNUNET_YES, GNUNET_YES,
2081                                                       NULL, NULL));
2082   }
2083   path_destroy (path);
2084   GCC_check_connections ();
2085   return GNUNET_OK;
2086 }
2087
2088
2089 /**
2090  * Core handler for path confirmations.
2091  *
2092  * @param cls closure
2093  * @param message message
2094  * @param peer peer identity this notification is about
2095  * @return #GNUNET_OK to keep the connection open,
2096  *         #GNUNET_SYSERR to close it (signal serious error)
2097  */
2098 int
2099 GCC_handle_confirm (void *cls,
2100                     const struct GNUNET_PeerIdentity *peer,
2101                     const struct GNUNET_MessageHeader *message)
2102 {
2103   struct GNUNET_CADET_ConnectionACK *msg;
2104   struct CadetConnection *c;
2105   struct CadetPeerPath *p;
2106   struct CadetPeer *pi;
2107   enum CadetConnectionState oldstate;
2108   int fwd;
2109
2110   GCC_check_connections ();
2111   msg = (struct GNUNET_CADET_ConnectionACK *) message;
2112   log_message (message, peer, &msg->cid);
2113   c = connection_get (&msg->cid);
2114   if (NULL == c)
2115   {
2116     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
2117                               1, GNUNET_NO);
2118     LOG (GNUNET_ERROR_TYPE_DEBUG,
2119          "  don't know the connection!\n");
2120     send_broken_unknown (&msg->cid, &my_full_id, NULL, peer);
2121     GCC_check_connections ();
2122     return GNUNET_OK;
2123   }
2124
2125   oldstate = c->state;
2126   LOG (GNUNET_ERROR_TYPE_DEBUG, "  via peer %s\n", GNUNET_i2s (peer));
2127   pi = GCP_get (peer, GNUNET_YES);
2128   if (get_next_hop (c) == pi)
2129   {
2130     LOG (GNUNET_ERROR_TYPE_DEBUG, "  SYNACK\n");
2131     fwd = GNUNET_NO;
2132     if (CADET_CONNECTION_SENT == oldstate)
2133       connection_change_state (c, CADET_CONNECTION_ACK);
2134   }
2135   else if (get_prev_hop (c) == pi)
2136   {
2137     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FINAL ACK\n");
2138     fwd = GNUNET_YES;
2139     connection_change_state (c, CADET_CONNECTION_READY);
2140   }
2141   else
2142   {
2143     GNUNET_break_op (0);
2144     return GNUNET_OK;
2145   }
2146
2147   connection_reset_timeout (c, fwd);
2148
2149   /* Add path to peers? */
2150   p = c->path;
2151   if (NULL != p)
2152   {
2153     GCP_add_path_to_all (p, GNUNET_YES);
2154   }
2155   else
2156   {
2157     GNUNET_break (0);
2158   }
2159
2160   /* Message for us as creator? */
2161   if (GCC_is_origin (c, GNUNET_YES))
2162   {
2163     if (GNUNET_NO != fwd)
2164     {
2165       GNUNET_break_op (0);
2166       return GNUNET_OK;
2167     }
2168     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection (SYN)ACK for us!\n");
2169
2170     /* If just created, cancel the short timeout and start a long one */
2171     if (CADET_CONNECTION_SENT == oldstate)
2172       connection_reset_timeout (c, GNUNET_YES);
2173
2174     /* Change connection state */
2175     connection_change_state (c, CADET_CONNECTION_READY);
2176     send_connection_ack (c, GNUNET_YES);
2177
2178     /* Change tunnel state, trigger KX */
2179     if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2180       GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2181     GCC_check_connections ();
2182     return GNUNET_OK;
2183   }
2184
2185   /* Message for us as destination? */
2186   if (GCC_is_terminal (c, GNUNET_YES))
2187   {
2188     if (GNUNET_YES != fwd)
2189     {
2190       GNUNET_break_op (0);
2191       return GNUNET_OK;
2192     }
2193     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Connection ACK for us!\n");
2194
2195     /* If just created, cancel the short timeout and start a long one */
2196     if (CADET_CONNECTION_ACK == oldstate)
2197       connection_reset_timeout (c, GNUNET_NO);
2198
2199     /* Change tunnel state */
2200     if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2201       GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2202     GCC_check_connections ();
2203     return GNUNET_OK;
2204   }
2205
2206   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2207   GNUNET_assert (NULL ==
2208                  GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2209                                             GNUNET_YES, NULL, NULL));
2210   GCC_check_connections ();
2211   return GNUNET_OK;
2212 }
2213
2214
2215 /**
2216  * Core handler for notifications of broken connections.
2217  *
2218  * @param cls Closure (unused).
2219  * @param id Peer identity of sending neighbor.
2220  * @param message Message.
2221  * @return #GNUNET_OK to keep the connection open,
2222  *         #GNUNET_SYSERR to close it (signal serious error)
2223  */
2224 int
2225 GCC_handle_broken (void* cls,
2226                    const struct GNUNET_PeerIdentity* id,
2227                    const struct GNUNET_MessageHeader* message)
2228 {
2229   struct GNUNET_CADET_ConnectionBroken *msg;
2230   struct CadetConnection *c;
2231   struct CadetTunnel *t;
2232   int pending;
2233   int fwd;
2234
2235   GCC_check_connections ();
2236   msg = (struct GNUNET_CADET_ConnectionBroken *) message;
2237   log_message (message, id, &msg->cid);
2238   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
2239               GNUNET_i2s (&msg->peer1));
2240   LOG (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
2241               GNUNET_i2s (&msg->peer2));
2242   c = connection_get (&msg->cid);
2243   if (NULL == c)
2244   {
2245     LOG (GNUNET_ERROR_TYPE_DEBUG, "  duplicate CONNECTION_BROKEN\n");
2246     GCC_check_connections ();
2247     return GNUNET_OK;
2248   }
2249
2250   t = c->t;
2251
2252   fwd = is_fwd (c, id);
2253   c->destroy = GNUNET_YES;
2254   if (GCC_is_terminal (c, fwd))
2255   {
2256     struct CadetPeer *endpoint;
2257
2258     if (NULL == t)
2259     {
2260       /* A terminal connection should not have 't' set to NULL. */
2261       GNUNET_break (0);
2262       GCC_debug (c, GNUNET_ERROR_TYPE_ERROR);
2263       return GNUNET_OK;
2264     }
2265     endpoint = GCP_get_short (c->path->peers[c->path->length - 1], GNUNET_YES);
2266     if (2 < c->path->length)
2267       path_invalidate (c->path);
2268     GCP_notify_broken_link (endpoint, &msg->peer1, &msg->peer2);
2269
2270     c->state = CADET_CONNECTION_BROKEN;
2271     GCT_remove_connection (t, c);
2272     c->t = NULL;
2273
2274     pending = c->pending_messages;
2275     if (0 < pending)
2276       resend_messages_and_destroy (c, !fwd);
2277     else
2278       GCC_destroy (c);
2279   }
2280   else
2281   {
2282     GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2283                                                       GNUNET_YES, NULL, NULL));
2284     connection_cancel_queues (c, !fwd);
2285   }
2286   GCC_check_connections ();
2287   return GNUNET_OK;
2288 }
2289
2290
2291 /**
2292  * Core handler for tunnel destruction
2293  *
2294  * @param cls Closure (unused).
2295  * @param peer Peer identity of sending neighbor.
2296  * @param message Message.
2297  * @return #GNUNET_OK to keep the connection open,
2298  *         #GNUNET_SYSERR to close it (signal serious error)
2299  */
2300 int
2301 GCC_handle_destroy (void *cls,
2302                     const struct GNUNET_PeerIdentity *peer,
2303                     const struct GNUNET_MessageHeader *message)
2304 {
2305   const struct GNUNET_CADET_ConnectionDestroy *msg;
2306   struct CadetConnection *c;
2307   int fwd;
2308
2309   GCC_check_connections ();
2310   msg = (const struct GNUNET_CADET_ConnectionDestroy *) message;
2311   log_message (message, peer, &msg->cid);
2312   c = connection_get (&msg->cid);
2313   if (NULL == c)
2314   {
2315     /* Probably already got the message from another path,
2316      * destroyed the tunnel and retransmitted to children.
2317      * Safe to ignore.
2318      */
2319     GNUNET_STATISTICS_update (stats,
2320                               "# control on unknown connection",
2321                               1, GNUNET_NO);
2322     LOG (GNUNET_ERROR_TYPE_DEBUG,
2323          "  connection unknown: already destroyed?\n");
2324     GCC_check_connections ();
2325     return GNUNET_OK;
2326   }
2327   fwd = is_fwd (c, peer);
2328   if (GNUNET_SYSERR == fwd)
2329   {
2330     GNUNET_break_op (0); /* FIXME */
2331     return GNUNET_OK;
2332   }
2333   if (GNUNET_NO == GCC_is_terminal (c, fwd))
2334   {
2335     GNUNET_assert (NULL ==
2336                    GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2337                                               GNUNET_YES, NULL, NULL));
2338   }
2339   else if (0 == c->pending_messages)
2340   {
2341     LOG (GNUNET_ERROR_TYPE_DEBUG,
2342          "  directly destroying connection!\n");
2343     GCC_destroy (c);
2344     GCC_check_connections ();
2345     return GNUNET_OK;
2346   }
2347   c->destroy = GNUNET_YES;
2348   c->state = CADET_CONNECTION_DESTROYED;
2349   if (NULL != c->t)
2350   {
2351     GCT_remove_connection (c->t, c);
2352     c->t = NULL;
2353   }
2354   GCC_check_connections ();
2355   return GNUNET_OK;
2356 }
2357
2358
2359 /**
2360  * Check the message against internal state and test if it goes FWD or BCK.
2361  *
2362  * Updates the PID, state and timeout values for the connection.
2363  *
2364  * @param message Message to check. It must belong to an existing connection.
2365  * @param minimum_size The message cannot be smaller than this value.
2366  * @param cid Connection ID (even if @a c is NULL, the ID is still needed).
2367  * @param c Connection this message should belong. If NULL, check fails.
2368  * @param neighbor Neighbor that sent the message.
2369  */
2370 static int
2371 check_message (const struct GNUNET_MessageHeader *message,
2372                size_t minimum_size,
2373                const struct GNUNET_CADET_Hash* cid,
2374                struct CadetConnection *c,
2375                const struct GNUNET_PeerIdentity *neighbor,
2376                uint32_t pid)
2377 {
2378   GNUNET_PEER_Id neighbor_id;
2379   struct CadetFlowControl *fc;
2380   struct CadetPeer *hop;
2381   int fwd;
2382   uint16_t type;
2383
2384   /* Check size */
2385   if (ntohs (message->size) < minimum_size)
2386   {
2387     GNUNET_break_op (0);
2388     LOG (GNUNET_ERROR_TYPE_WARNING, "Size %u < %u\n",
2389          ntohs (message->size), minimum_size);
2390     return GNUNET_SYSERR;
2391   }
2392
2393   /* Check connection */
2394   if (NULL == c)
2395   {
2396     GNUNET_STATISTICS_update (stats,
2397                               "# unknown connection",
2398                               1, GNUNET_NO);
2399     LOG (GNUNET_ERROR_TYPE_DEBUG,
2400          "%s on unknown connection %s\n",
2401          GC_m2s (ntohs (message->type)),
2402          GNUNET_h2s (GC_h2hc (cid)));
2403     send_broken_unknown (cid,
2404                          &my_full_id,
2405                          NULL,
2406                          neighbor);
2407     return GNUNET_SYSERR;
2408   }
2409
2410   /* Check if origin is as expected */
2411   neighbor_id = GNUNET_PEER_search (neighbor);
2412   hop = get_prev_hop (c);
2413   if (neighbor_id == GCP_get_short_id (hop))
2414   {
2415     fwd = GNUNET_YES;
2416   }
2417   else
2418   {
2419     hop = get_next_hop (c);
2420     GNUNET_break (hop == c->next_peer);
2421     if (neighbor_id == GCP_get_short_id (hop))
2422     {
2423       fwd = GNUNET_NO;
2424     }
2425     else
2426     {
2427       /* Unexpected peer sending traffic on a connection. */
2428       GNUNET_break_op (0);
2429       return GNUNET_SYSERR;
2430     }
2431   }
2432
2433   /* Check PID for payload messages */
2434   type = ntohs (message->type);
2435   if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type
2436       || GNUNET_MESSAGE_TYPE_CADET_AX == type)
2437   {
2438     fc = fwd ? &c->bck_fc : &c->fwd_fc;
2439     LOG (GNUNET_ERROR_TYPE_DEBUG, " PID %u (expected %u - %u)\n",
2440          pid, fc->last_pid_recv + 1, fc->last_ack_sent);
2441     if (GC_is_pid_bigger (pid, fc->last_ack_sent))
2442     {
2443       GNUNET_break_op (0);
2444       GNUNET_STATISTICS_update (stats, "# unsolicited message", 1, GNUNET_NO);
2445       LOG (GNUNET_ERROR_TYPE_WARNING, "Received PID %u, (prev %u), ACK %u\n",
2446           pid, fc->last_pid_recv, fc->last_ack_sent);
2447       return GNUNET_SYSERR;
2448     }
2449     if (GC_is_pid_bigger (pid, fc->last_pid_recv))
2450     {
2451       unsigned int delta;
2452
2453       delta = pid - fc->last_pid_recv;
2454       fc->last_pid_recv = pid;
2455       fc->recv_bitmap <<= delta;
2456       fc->recv_bitmap |= 1;
2457     }
2458     else
2459     {
2460       GNUNET_STATISTICS_update (stats, "# out of order PID", 1, GNUNET_NO);
2461       if (GNUNET_NO == is_ooo_ok (fc->last_pid_recv, pid, fc->recv_bitmap))
2462       {
2463         LOG (GNUNET_ERROR_TYPE_WARNING, "PID %u unexpected (%u+), dropping!\n",
2464              pid, fc->last_pid_recv - 31);
2465         return GNUNET_SYSERR;
2466       }
2467       fc->recv_bitmap |= get_recv_bitmask (fc->last_pid_recv, pid);
2468     }
2469   }
2470
2471   /* Count as connection confirmation. */
2472   if (CADET_CONNECTION_SENT == c->state || CADET_CONNECTION_ACK == c->state)
2473   {
2474     connection_change_state (c, CADET_CONNECTION_READY);
2475     if (NULL != c->t)
2476     {
2477       if (CADET_TUNNEL_WAITING == GCT_get_cstate (c->t))
2478         GCT_change_cstate (c->t, CADET_TUNNEL_READY);
2479     }
2480   }
2481   connection_reset_timeout (c, fwd);
2482
2483   return fwd;
2484 }
2485
2486
2487 /**
2488  * Generic handler for cadet network encrypted traffic.
2489  *
2490  * @param peer Peer identity this notification is about.
2491  * @param msg Encrypted message.
2492  * @return #GNUNET_OK to keep the connection open,
2493  *         #GNUNET_SYSERR to close it (signal serious error)
2494  */
2495 static int
2496 handle_cadet_encrypted (const struct GNUNET_PeerIdentity *peer,
2497                         const struct GNUNET_MessageHeader *message)
2498 {
2499   const struct GNUNET_CADET_Encrypted *otr_msg;
2500   const struct GNUNET_CADET_AX *ax_msg;
2501   const struct GNUNET_CADET_Hash* cid;
2502   struct CadetConnection *c;
2503   size_t minimum_size;
2504   size_t overhead;
2505   uint32_t pid;
2506   uint32_t ttl;
2507   int fwd;
2508
2509   GCC_check_connections ();
2510   if (GNUNET_MESSAGE_TYPE_CADET_AX == ntohs (message->type))
2511   {
2512     overhead = sizeof (struct GNUNET_CADET_AX);
2513     ax_msg = (const struct GNUNET_CADET_AX *) message;
2514     cid = &ax_msg->cid;
2515     pid = ntohl (ax_msg->pid);
2516     otr_msg = NULL;
2517   }
2518   else
2519   {
2520     overhead = sizeof (struct GNUNET_CADET_Encrypted);
2521     otr_msg = (const struct GNUNET_CADET_Encrypted *) message;
2522     cid = &otr_msg->cid;
2523     pid = ntohl (otr_msg->pid);
2524   }
2525
2526   log_message (message, peer, cid);
2527
2528   minimum_size = sizeof (struct GNUNET_MessageHeader) + overhead;
2529   c = connection_get (cid);
2530   fwd = check_message (message,
2531                        minimum_size,
2532                        cid,
2533                        c,
2534                        peer,
2535                        pid);
2536
2537   /* If something went wrong, discard message. */
2538   if (GNUNET_SYSERR == fwd)
2539   {
2540     GCC_check_connections ();
2541     return GNUNET_OK;
2542   }
2543
2544   /* Is this message for us? */
2545   if (GCC_is_terminal (c, fwd))
2546   {
2547     GNUNET_STATISTICS_update (stats, "# received encrypted", 1, GNUNET_NO);
2548
2549     if (NULL == c->t)
2550     {
2551       GNUNET_break (GNUNET_NO != c->destroy);
2552       return GNUNET_OK;
2553     }
2554     GCT_handle_encrypted (c->t, message);
2555     GCC_send_ack (c, fwd, GNUNET_NO);
2556     GCC_check_connections ();
2557     return GNUNET_OK;
2558   }
2559
2560   /* Message not for us: forward to next hop */
2561   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2562   if (NULL != otr_msg) /* only otr has ttl */
2563   {
2564     ttl = ntohl (otr_msg->ttl);
2565     LOG (GNUNET_ERROR_TYPE_DEBUG, "   ttl: %u\n", ttl);
2566     if (ttl == 0)
2567     {
2568       GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
2569       LOG (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
2570       GCC_send_ack (c, fwd, GNUNET_NO);
2571       GCC_check_connections ();
2572       return GNUNET_OK;
2573     }
2574   }
2575
2576   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2577   GNUNET_assert (NULL == GCC_send_prebuilt_message (message, 0, 0, c, fwd,
2578                                                     GNUNET_NO, NULL, NULL));
2579   GCC_check_connections ();
2580   return GNUNET_OK;
2581 }
2582
2583
2584 /**
2585  * Generic handler for cadet network encrypted traffic.
2586  *
2587  * @param peer Peer identity this notification is about.
2588  * @param msg Encrypted message.
2589  * @return #GNUNET_OK to keep the connection open,
2590  *         #GNUNET_SYSERR to close it (signal serious error)
2591  */
2592 static int
2593 handle_cadet_kx (const struct GNUNET_PeerIdentity *peer,
2594                  const struct GNUNET_CADET_KX *msg)
2595 {
2596   const struct GNUNET_CADET_Hash* cid;
2597   struct CadetConnection *c;
2598   size_t expected_size;
2599   int fwd;
2600
2601   GCC_check_connections ();
2602   cid = &msg->cid;
2603   log_message (&msg->header, peer, cid);
2604
2605   expected_size = sizeof (struct GNUNET_CADET_KX)
2606                   + sizeof (struct GNUNET_MessageHeader);
2607   c = connection_get (cid);
2608   fwd = check_message (&msg->header,
2609                        expected_size,
2610                        cid,
2611                        c,
2612                        peer,
2613                        0);
2614
2615   /* If something went wrong, discard message. */
2616   if (GNUNET_SYSERR == fwd)
2617     return GNUNET_OK;
2618
2619   /* Is this message for us? */
2620   if (GCC_is_terminal (c, fwd))
2621   {
2622     LOG (GNUNET_ERROR_TYPE_DEBUG, "  message for us!\n");
2623     GNUNET_STATISTICS_update (stats, "# received KX", 1, GNUNET_NO);
2624     if (NULL == c->t)
2625     {
2626       GNUNET_break (0);
2627       return GNUNET_OK;
2628     }
2629     GCT_handle_kx (c->t, &msg[1].header);
2630     GCC_check_connections ();
2631     return GNUNET_OK;
2632   }
2633
2634   /* Message not for us: forward to next hop */
2635   LOG (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2636   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2637   GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg->header, 0, 0, c, fwd,
2638                                                     GNUNET_NO, NULL, NULL));
2639   GCC_check_connections ();
2640   return GNUNET_OK;
2641 }
2642
2643
2644 /**
2645  * Core handler for key exchange traffic (ephemeral key, ping, pong).
2646  *
2647  * @param cls Closure (unused).
2648  * @param message Message received.
2649  * @param peer Peer who sent the message.
2650  * @return #GNUNET_OK to keep the connection open,
2651  *         #GNUNET_SYSERR to close it (signal serious error)
2652  */
2653 int
2654 GCC_handle_kx (void *cls,
2655                const struct GNUNET_PeerIdentity *peer,
2656                const struct GNUNET_MessageHeader *message)
2657 {
2658   GCC_check_connections ();
2659   return handle_cadet_kx (peer, (struct GNUNET_CADET_KX *) message);
2660 }
2661
2662
2663 /**
2664  * Core handler for encrypted cadet network traffic (channel mgmt, data).
2665  *
2666  * @param cls Closure (unused).
2667  * @param message Message received.
2668  * @param peer Peer who sent the message.
2669  * @return #GNUNET_OK to keep the connection open,
2670  *         #GNUNET_SYSERR to close it (signal serious error)
2671  */
2672 int
2673 GCC_handle_encrypted (void *cls, const struct GNUNET_PeerIdentity *peer,
2674                       const struct GNUNET_MessageHeader *message)
2675 {
2676   GCC_check_connections ();
2677   return handle_cadet_encrypted (peer, message);
2678 }
2679
2680
2681 /**
2682  * Core handler for cadet network traffic point-to-point acks.
2683  *
2684  * @param cls closure
2685  * @param message message
2686  * @param peer peer identity this notification is about
2687  * @return #GNUNET_OK to keep the connection open,
2688  *         #GNUNET_SYSERR to close it (signal serious error)
2689  */
2690 int
2691 GCC_handle_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
2692                 const struct GNUNET_MessageHeader *message)
2693 {
2694   struct GNUNET_CADET_ACK *msg;
2695   struct CadetConnection *c;
2696   struct CadetFlowControl *fc;
2697   GNUNET_PEER_Id id;
2698   uint32_t ack;
2699   int fwd;
2700
2701   GCC_check_connections ();
2702   msg = (struct GNUNET_CADET_ACK *) message;
2703   log_message (message, peer, &msg->cid);
2704   c = connection_get (&msg->cid);
2705   if (NULL == c)
2706   {
2707     GNUNET_STATISTICS_update (stats,
2708                               "# ack on unknown connection",
2709                               1,
2710                               GNUNET_NO);
2711     send_broken_unknown (&msg->cid,
2712                          &my_full_id,
2713                          NULL,
2714                          peer);
2715     GCC_check_connections ();
2716     return GNUNET_OK;
2717   }
2718
2719   /* Is this a forward or backward ACK? */
2720   id = GNUNET_PEER_search (peer);
2721   if (GCP_get_short_id (get_next_hop (c)) == id)
2722   {
2723     fc = &c->fwd_fc;
2724     fwd = GNUNET_YES;
2725   }
2726   else if (GCP_get_short_id (get_prev_hop (c)) == id)
2727   {
2728     fc = &c->bck_fc;
2729     fwd = GNUNET_NO;
2730   }
2731   else
2732   {
2733     GNUNET_break_op (0);
2734     return GNUNET_OK;
2735   }
2736
2737   ack = ntohl (msg->ack);
2738   LOG (GNUNET_ERROR_TYPE_DEBUG, " %s ACK %u (was %u)\n",
2739        GC_f2s (fwd), ack, fc->last_ack_recv);
2740   if (GC_is_pid_bigger (ack, fc->last_ack_recv))
2741     fc->last_ack_recv = ack;
2742
2743   /* Cancel polling if the ACK is big enough. */
2744   if (NULL != fc->poll_task &&
2745       GC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
2746   {
2747     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Cancel poll\n");
2748     GNUNET_SCHEDULER_cancel (fc->poll_task);
2749     fc->poll_task = NULL;
2750     fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
2751   }
2752
2753   connection_unlock_queue (c, fwd);
2754   GCC_check_connections ();
2755   return GNUNET_OK;
2756 }
2757
2758
2759 /**
2760  * Core handler for cadet network traffic point-to-point ack polls.
2761  *
2762  * @param cls closure
2763  * @param message message
2764  * @param peer peer identity this notification is about
2765  * @return #GNUNET_OK to keep the connection open,
2766  *         #GNUNET_SYSERR to close it (signal serious error)
2767  */
2768 int
2769 GCC_handle_poll (void *cls,
2770                  const struct GNUNET_PeerIdentity *peer,
2771                  const struct GNUNET_MessageHeader *message)
2772 {
2773   struct GNUNET_CADET_Poll *msg;
2774   struct CadetConnection *c;
2775   struct CadetFlowControl *fc;
2776   GNUNET_PEER_Id id;
2777   uint32_t pid;
2778   int fwd;
2779
2780   GCC_check_connections ();
2781   msg = (struct GNUNET_CADET_Poll *) message;
2782   log_message (message, peer, &msg->cid);
2783   c = connection_get (&msg->cid);
2784   if (NULL == c)
2785   {
2786     GNUNET_STATISTICS_update (stats, "# poll on unknown connection", 1,
2787                               GNUNET_NO);
2788     LOG (GNUNET_ERROR_TYPE_DEBUG,
2789          "POLL message on unknown connection %s!\n",
2790          GNUNET_h2s (GC_h2hc (&msg->cid)));
2791     send_broken_unknown (&msg->cid,
2792                          &my_full_id,
2793                          NULL,
2794                          peer);
2795     GCC_check_connections ();
2796     return GNUNET_OK;
2797   }
2798
2799   /* Is this a forward or backward ACK?
2800    * Note: a poll should never be needed in a loopback case,
2801    * since there is no possiblility of packet loss there, so
2802    * this way of discerining FWD/BCK should not be a problem.
2803    */
2804   id = GNUNET_PEER_search (peer);
2805   if (GCP_get_short_id (get_next_hop (c)) == id)
2806   {
2807     LOG (GNUNET_ERROR_TYPE_DEBUG, "  FWD FC\n");
2808     fc = &c->fwd_fc;
2809   }
2810   else if (GCP_get_short_id (get_prev_hop (c)) == id)
2811   {
2812     LOG (GNUNET_ERROR_TYPE_DEBUG, "  BCK FC\n");
2813     fc = &c->bck_fc;
2814   }
2815   else
2816   {
2817     GNUNET_break_op (0);
2818     return GNUNET_OK;
2819   }
2820
2821   pid = ntohl (msg->pid);
2822   LOG (GNUNET_ERROR_TYPE_DEBUG, "  PID %u, OLD %u\n", pid, fc->last_pid_recv);
2823   fc->last_pid_recv = pid;
2824   fwd = fc == &c->bck_fc;
2825   GCC_send_ack (c, fwd, GNUNET_YES);
2826   GCC_check_connections ();
2827
2828   return GNUNET_OK;
2829 }
2830
2831
2832 /**
2833  * Send an ACK on the appropriate connection/channel, depending on
2834  * the direction and the position of the peer.
2835  *
2836  * @param c Which connection to send the hop-by-hop ACK.
2837  * @param fwd Is this a fwd ACK? (will go dest->root).
2838  * @param force Send the ACK even if suboptimal (e.g. requested by POLL).
2839  */
2840 void
2841 GCC_send_ack (struct CadetConnection *c, int fwd, int force)
2842 {
2843   unsigned int buffer;
2844
2845   GCC_check_connections ();
2846   LOG (GNUNET_ERROR_TYPE_DEBUG, "GCC send %s ACK on %s\n",
2847        GC_f2s (fwd), GCC_2s (c));
2848
2849   if (NULL == c)
2850   {
2851     GNUNET_break (0);
2852     return;
2853   }
2854
2855   if (GNUNET_NO != c->destroy)
2856   {
2857     LOG (GNUNET_ERROR_TYPE_DEBUG, "  being destroyed, why bother...\n");
2858     GCC_check_connections ();
2859     return;
2860   }
2861
2862   /* Get available buffer space */
2863   if (GCC_is_terminal (c, fwd))
2864   {
2865     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from all channels\n");
2866     buffer = GCT_get_channels_buffer (c->t);
2867   }
2868   else
2869   {
2870     LOG (GNUNET_ERROR_TYPE_DEBUG, "  getting from one connection\n");
2871     buffer = GCC_get_buffer (c, fwd);
2872   }
2873   LOG (GNUNET_ERROR_TYPE_DEBUG, "  buffer available: %u\n", buffer);
2874   if (0 == buffer && GNUNET_NO == force)
2875   {
2876     GCC_check_connections ();
2877     return;
2878   }
2879
2880   /* Send available buffer space */
2881   if (GCC_is_origin (c, fwd))
2882   {
2883     GNUNET_assert (NULL != c->t);
2884     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on channels...\n");
2885     GCT_unchoke_channels (c->t);
2886   }
2887   else
2888   {
2889     LOG (GNUNET_ERROR_TYPE_DEBUG, "  sending on connection\n");
2890     send_ack (c, buffer, fwd, force);
2891   }
2892   GCC_check_connections ();
2893 }
2894
2895
2896 /**
2897  * Initialize the connections subsystem
2898  *
2899  * @param c Configuration handle.
2900  */
2901 void
2902 GCC_init (const struct GNUNET_CONFIGURATION_Handle *c)
2903 {
2904   LOG (GNUNET_ERROR_TYPE_DEBUG, "init\n");
2905   if (GNUNET_OK !=
2906       GNUNET_CONFIGURATION_get_value_number (c, "CADET", "MAX_MSGS_QUEUE",
2907                                              &max_msgs_queue))
2908   {
2909     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2910                                "CADET", "MAX_MSGS_QUEUE", "MISSING");
2911     GNUNET_SCHEDULER_shutdown ();
2912     return;
2913   }
2914
2915   if (GNUNET_OK !=
2916       GNUNET_CONFIGURATION_get_value_number (c, "CADET", "MAX_CONNECTIONS",
2917                                              &max_connections))
2918   {
2919     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2920                                "CADET", "MAX_CONNECTIONS", "MISSING");
2921     GNUNET_SCHEDULER_shutdown ();
2922     return;
2923   }
2924
2925   if (GNUNET_OK !=
2926       GNUNET_CONFIGURATION_get_value_time (c, "CADET", "REFRESH_CONNECTION_TIME",
2927                                            &refresh_connection_time))
2928   {
2929     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2930                                "CADET", "REFRESH_CONNECTION_TIME", "MISSING");
2931     GNUNET_SCHEDULER_shutdown ();
2932     return;
2933   }
2934   create_connection_time = GNUNET_TIME_UNIT_SECONDS;
2935   connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_NO);
2936 }
2937
2938
2939 /**
2940  * Destroy each connection on shutdown.
2941  *
2942  * @param cls Closure (unused).
2943  * @param key Current key code (CID, unused).
2944  * @param value Value in the hash map (`struct CadetConnection`)
2945  *
2946  * @return #GNUNET_YES, because we should continue to iterate
2947  */
2948 static int
2949 shutdown_iterator (void *cls,
2950                    const struct GNUNET_HashCode *key,
2951                    void *value)
2952 {
2953   struct CadetConnection *c = value;
2954
2955   c->state = CADET_CONNECTION_DESTROYED;
2956   GCC_destroy (c);
2957   return GNUNET_YES;
2958 }
2959
2960
2961 /**
2962  * Shut down the connections subsystem.
2963  */
2964 void
2965 GCC_shutdown (void)
2966 {
2967   LOG (GNUNET_ERROR_TYPE_DEBUG, "Shutting down connections\n");
2968   GCC_check_connections ();
2969   GNUNET_CONTAINER_multihashmap_iterate (connections,
2970                                          &shutdown_iterator,
2971                                          NULL);
2972   GNUNET_CONTAINER_multihashmap_destroy (connections);
2973   connections = NULL;
2974 }
2975
2976
2977 /**
2978  * Create a connection.
2979  *
2980  * @param cid Connection ID (either created locally or imposed remotely).
2981  * @param t Tunnel this connection belongs to (or NULL);
2982  * @param path Path this connection has to use (copy is made).
2983  * @param own_pos Own position in the @c path path.
2984  *
2985  * @return Newly created connection, NULL in case of error (own id not in path).
2986  */
2987 struct CadetConnection *
2988 GCC_new (const struct GNUNET_CADET_Hash *cid,
2989          struct CadetTunnel *t,
2990          struct CadetPeerPath *path,
2991          unsigned int own_pos)
2992 {
2993   struct CadetConnection *c;
2994   struct CadetPeerPath *cpath;
2995
2996   GCC_check_connections ();
2997   cpath = path_duplicate (path);
2998   GNUNET_assert (NULL != cpath);
2999   c = GNUNET_new (struct CadetConnection);
3000   c->id = *cid;
3001   GNUNET_assert (GNUNET_OK ==
3002                  GNUNET_CONTAINER_multihashmap_put (connections,
3003                                                     GCC_get_h (c), c,
3004                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
3005   fc_init (&c->fwd_fc);
3006   fc_init (&c->bck_fc);
3007   c->fwd_fc.c = c;
3008   c->bck_fc.c = c;
3009
3010   c->t = t;
3011   GNUNET_assert (own_pos <= cpath->length - 1);
3012   c->own_pos = own_pos;
3013   c->path = cpath;
3014   cpath->c = c;
3015   if (GNUNET_OK != register_neighbors (c))
3016   {
3017     if (0 == own_pos)
3018     {
3019       /* We were the origin of this request, this means we have invalid
3020        * info about the paths to reach the destination. We must invalidate
3021        * the *original* path to avoid trying it again in the next minute.
3022        */
3023       if (2 < path->length)
3024         path_invalidate (path);
3025       else
3026       {
3027         GNUNET_break (0);
3028         GCT_debug(t, GNUNET_ERROR_TYPE_WARNING);
3029       }
3030       c->t = NULL;
3031     }
3032     path_destroy (c->path);
3033     c->path = NULL;
3034     GCC_destroy (c);
3035     return NULL;
3036   }
3037   LOG (GNUNET_ERROR_TYPE_INFO, "New connection %s\n", GCC_2s (c));
3038   GCC_check_connections ();
3039   return c;
3040 }
3041
3042
3043 void
3044 GCC_destroy (struct CadetConnection *c)
3045 {
3046   GCC_check_connections ();
3047   if (NULL == c)
3048   {
3049     GNUNET_break (0);
3050     return;
3051   }
3052
3053   if (2 == c->destroy) /* cancel queues -> GCP_queue_cancel -> q_destroy -> */
3054     return;            /* -> message_sent -> GCC_destroy. Don't loop. */
3055   c->destroy = 2;
3056
3057   LOG (GNUNET_ERROR_TYPE_DEBUG,
3058        "destroying connection %s\n",
3059        GCC_2s (c));
3060   LOG (GNUNET_ERROR_TYPE_DEBUG,
3061        " fc's f: %p, b: %p\n",
3062        &c->fwd_fc, &c->bck_fc);
3063   LOG (GNUNET_ERROR_TYPE_DEBUG,
3064        " fc tasks f: %u, b: %u\n",
3065        c->fwd_fc.poll_task,
3066        c->bck_fc.poll_task);
3067
3068   /* Cancel all traffic */
3069   if (NULL != c->path)
3070   {
3071     connection_cancel_queues (c, GNUNET_YES);
3072     connection_cancel_queues (c, GNUNET_NO);
3073   }
3074   unregister_neighbors (c);
3075   path_destroy (c->path);
3076   c->path = NULL;
3077
3078   /* Delete from tunnel */
3079   if (NULL != c->t)
3080     GCT_remove_connection (c->t, c);
3081
3082   if (NULL != c->check_duplicates_task)
3083     GNUNET_SCHEDULER_cancel (c->check_duplicates_task);
3084   if (NULL != c->fwd_maintenance_task)
3085     GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
3086   if (NULL != c->bck_maintenance_task)
3087     GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
3088
3089   if (GNUNET_NO == c->was_removed)
3090   {
3091     GNUNET_break (GNUNET_YES ==
3092                   GNUNET_CONTAINER_multihashmap_remove (connections,
3093                                                         GCC_get_h (c),
3094                                                         c));
3095   }
3096   GNUNET_STATISTICS_update (stats,
3097                             "# connections",
3098                             -1,
3099                             GNUNET_NO);
3100   GNUNET_free (c);
3101   GCC_check_connections ();
3102 }
3103
3104
3105 /**
3106  * Get the connection ID.
3107  *
3108  * @param c Connection to get the ID from.
3109  *
3110  * @return ID of the connection.
3111  */
3112 const struct GNUNET_CADET_Hash *
3113 GCC_get_id (const struct CadetConnection *c)
3114 {
3115   return &c->id;
3116 }
3117
3118
3119 /**
3120  * Get the connection ID.
3121  *
3122  * @param c Connection to get the ID from.
3123  *
3124  * @return ID of the connection.
3125  */
3126 const struct GNUNET_HashCode *
3127 GCC_get_h (const struct CadetConnection *c)
3128 {
3129   return GC_h2hc (&c->id);
3130 }
3131
3132
3133 /**
3134  * Get the connection path.
3135  *
3136  * @param c Connection to get the path from.
3137  *
3138  * @return path used by the connection.
3139  */
3140 const struct CadetPeerPath *
3141 GCC_get_path (const struct CadetConnection *c)
3142 {
3143   if (GNUNET_NO == c->destroy)
3144     return c->path;
3145   return NULL;
3146 }
3147
3148
3149 /**
3150  * Get the connection state.
3151  *
3152  * @param c Connection to get the state from.
3153  *
3154  * @return state of the connection.
3155  */
3156 enum CadetConnectionState
3157 GCC_get_state (const struct CadetConnection *c)
3158 {
3159   return c->state;
3160 }
3161
3162 /**
3163  * Get the connection tunnel.
3164  *
3165  * @param c Connection to get the tunnel from.
3166  *
3167  * @return tunnel of the connection.
3168  */
3169 struct CadetTunnel *
3170 GCC_get_tunnel (const struct CadetConnection *c)
3171 {
3172   return c->t;
3173 }
3174
3175
3176 /**
3177  * Get free buffer space in a connection.
3178  *
3179  * @param c Connection.
3180  * @param fwd Is query about FWD traffic?
3181  *
3182  * @return Free buffer space [0 - max_msgs_queue/max_connections]
3183  */
3184 unsigned int
3185 GCC_get_buffer (struct CadetConnection *c, int fwd)
3186 {
3187   struct CadetFlowControl *fc;
3188
3189   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3190
3191   LOG (GNUNET_ERROR_TYPE_DEBUG, "  Get %s buffer on %s: %u - %u\n",
3192        GC_f2s (fwd), GCC_2s (c), fc->queue_max, fc->queue_n);
3193   GCC_debug (c, GNUNET_ERROR_TYPE_DEBUG);
3194
3195   return (fc->queue_max - fc->queue_n);
3196 }
3197
3198
3199 /**
3200  * Get how many messages have we allowed to send to us from a direction.
3201  *
3202  * @param c Connection.
3203  * @param fwd Are we asking about traffic from FWD (BCK messages)?
3204  *
3205  * @return last_ack_sent - last_pid_recv
3206  */
3207 unsigned int
3208 GCC_get_allowed (struct CadetConnection *c, int fwd)
3209 {
3210   struct CadetFlowControl *fc;
3211
3212   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3213   if (CADET_CONNECTION_READY != c->state
3214       || GC_is_pid_bigger (fc->last_pid_recv, fc->last_ack_sent))
3215   {
3216     return 0;
3217   }
3218   return (fc->last_ack_sent - fc->last_pid_recv);
3219 }
3220
3221
3222 /**
3223  * Get messages queued in a connection.
3224  *
3225  * @param c Connection.
3226  * @param fwd Is query about FWD traffic?
3227  *
3228  * @return Number of messages queued.
3229  */
3230 unsigned int
3231 GCC_get_qn (struct CadetConnection *c, int fwd)
3232 {
3233   struct CadetFlowControl *fc;
3234
3235   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3236
3237   return fc->queue_n;
3238 }
3239
3240
3241 /**
3242  * Get next PID to use.
3243  *
3244  * @param c Connection.
3245  * @param fwd Is query about FWD traffic?
3246  *
3247  * @return Last PID used + 1.
3248  */
3249 unsigned int
3250 GCC_get_pid (struct CadetConnection *c, int fwd)
3251 {
3252   struct CadetFlowControl *fc;
3253
3254   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3255
3256   return fc->last_pid_sent + 1;
3257 }
3258
3259
3260 /**
3261  * Allow the connection to advertise a buffer of the given size.
3262  *
3263  * The connection will send an @c fwd ACK message (so: in direction !fwd)
3264  * allowing up to last_pid_recv + buffer.
3265  *
3266  * @param c Connection.
3267  * @param buffer How many more messages the connection can accept.
3268  * @param fwd Is this about FWD traffic? (The ack will go dest->root).
3269  */
3270 void
3271 GCC_allow (struct CadetConnection *c, unsigned int buffer, int fwd)
3272 {
3273   LOG (GNUNET_ERROR_TYPE_DEBUG, "  allowing %s %u messages %s\n",
3274        GCC_2s (c), buffer, GC_f2s (fwd));
3275   send_ack (c, buffer, fwd, GNUNET_NO);
3276 }
3277
3278
3279 /**
3280  * Notify other peers on a connection of a broken link. Mark connections
3281  * to destroy after all traffic has been sent.
3282  *
3283  * @param c Connection on which there has been a disconnection.
3284  * @param peer Peer that disconnected.
3285  */
3286 void
3287 GCC_neighbor_disconnected (struct CadetConnection *c, struct CadetPeer *peer)
3288 {
3289   struct CadetPeer *hop;
3290   char peer_name[16];
3291   int fwd;
3292
3293   GCC_check_connections ();
3294   strncpy (peer_name, GCP_2s (peer), 16);
3295   peer_name[15] = '\0';
3296   LOG (GNUNET_ERROR_TYPE_DEBUG,
3297        "shutting down %s, %s disconnected\n",
3298        GCC_2s (c), peer_name);
3299
3300   invalidate_paths (c, peer);
3301
3302   hop = get_prev_hop (c);
3303   if (NULL == hop)
3304   {
3305     /* Path was NULL, we should have deleted the connection. */
3306     GNUNET_break (0);
3307     return;
3308   }
3309   fwd = (peer == hop);
3310   if ( (GNUNET_YES == GCC_is_terminal (c, fwd)) ||
3311        (GNUNET_NO != c->destroy) )
3312   {
3313     /* Local shutdown, or other peer already down (hence 'c->destroy');
3314        so there is no one to notify about this, just clean up. */
3315     GCC_destroy (c);
3316     GCC_check_connections ();
3317     return;
3318   }
3319   send_broken (c, &my_full_id, GCP_get_id (peer), fwd);
3320   /* Connection will have at least one pending message
3321    * (the one we just scheduled), so delay destruction
3322    * and remove from map so we don't use accidentally. */
3323   c->destroy = GNUNET_YES;
3324   c->state = CADET_CONNECTION_DESTROYED;
3325   GNUNET_assert (GNUNET_NO == c->was_removed);
3326   c->was_removed = GNUNET_YES;
3327   GNUNET_break (GNUNET_YES ==
3328                 GNUNET_CONTAINER_multihashmap_remove (connections,
3329                                                       GCC_get_h (c),
3330                                                       c));
3331   /* Cancel queue in the direction that just died. */
3332   connection_cancel_queues (c, ! fwd);
3333   GCC_stop_poll (c, ! fwd);
3334   unregister_neighbors (c);
3335   GCC_check_connections ();
3336 }
3337
3338
3339 /**
3340  * Is this peer the first one on the connection?
3341  *
3342  * @param c Connection.
3343  * @param fwd Is this about fwd traffic?
3344  *
3345  * @return #GNUNET_YES if origin, #GNUNET_NO if relay/terminal.
3346  */
3347 int
3348 GCC_is_origin (struct CadetConnection *c, int fwd)
3349 {
3350   if (!fwd && c->path->length - 1 == c->own_pos )
3351     return GNUNET_YES;
3352   if (fwd && 0 == c->own_pos)
3353     return GNUNET_YES;
3354   return GNUNET_NO;
3355 }
3356
3357
3358 /**
3359  * Is this peer the last one on the connection?
3360  *
3361  * @param c Connection.
3362  * @param fwd Is this about fwd traffic?
3363  *            Note that the ROOT is the terminal for BCK traffic!
3364  *
3365  * @return #GNUNET_YES if terminal, #GNUNET_NO if relay/origin.
3366  */
3367 int
3368 GCC_is_terminal (struct CadetConnection *c, int fwd)
3369 {
3370   return GCC_is_origin (c, ! fwd);
3371 }
3372
3373
3374 /**
3375  * See if we are allowed to send by the next hop in the given direction.
3376  *
3377  * @param c Connection.
3378  * @param fwd Is this about fwd traffic?
3379  *
3380  * @return #GNUNET_YES in case it's OK to send.
3381  */
3382 int
3383 GCC_is_sendable (struct CadetConnection *c, int fwd)
3384 {
3385   struct CadetFlowControl *fc;
3386
3387   LOG (GNUNET_ERROR_TYPE_DEBUG,
3388        " checking sendability of %s traffic on %s\n",
3389        GC_f2s (fwd), GCC_2s (c));
3390   if (NULL == c)
3391   {
3392     GNUNET_break (0);
3393     return GNUNET_YES;
3394   }
3395   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3396   LOG (GNUNET_ERROR_TYPE_DEBUG,
3397        " last ack recv: %u, last pid sent: %u\n",
3398        fc->last_ack_recv, fc->last_pid_sent);
3399   if (GC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
3400   {
3401     LOG (GNUNET_ERROR_TYPE_DEBUG, " sendable\n");
3402     return GNUNET_YES;
3403   }
3404   LOG (GNUNET_ERROR_TYPE_DEBUG, " not sendable\n");
3405   return GNUNET_NO;
3406 }
3407
3408
3409 /**
3410  * Check if this connection is a direct one (never trim a direct connection).
3411  *
3412  * @param c Connection.
3413  *
3414  * @return #GNUNET_YES in case it's a direct connection, #GNUNET_NO otherwise.
3415  */
3416 int
3417 GCC_is_direct (struct CadetConnection *c)
3418 {
3419   return (c->path->length == 2) ? GNUNET_YES : GNUNET_NO;
3420 }
3421
3422
3423 /**
3424  * Sends an already built message on a connection, properly registering
3425  * all used resources.
3426  *
3427  * @param message Message to send. Function makes a copy of it.
3428  *                If message is not hop-by-hop, decrements TTL of copy.
3429  * @param payload_type Type of payload, in case the message is encrypted.
3430  * @param c Connection on which this message is transmitted.
3431  * @param fwd Is this a fwd message?
3432  * @param force Force the connection to accept the message (buffer overfill).
3433  * @param cont Continuation called once message is sent. Can be NULL.
3434  * @param cont_cls Closure for @c cont.
3435  *
3436  * @return Handle to cancel the message before it's sent.
3437  *         NULL on error or if @c cont is NULL.
3438  *         Invalid on @c cont call.
3439  */
3440 struct CadetConnectionQueue *
3441 GCC_send_prebuilt_message (const struct GNUNET_MessageHeader *message,
3442                            uint16_t payload_type, uint32_t payload_id,
3443                            struct CadetConnection *c, int fwd, int force,
3444                            GCC_sent cont, void *cont_cls)
3445 {
3446   struct CadetFlowControl *fc;
3447   struct CadetConnectionQueue *q;
3448   void *data;
3449   size_t size;
3450   uint16_t type;
3451   int droppable;
3452
3453   GCC_check_connections ();
3454   size = ntohs (message->size);
3455   data = GNUNET_malloc (size);
3456   memcpy (data, message, size);
3457   type = ntohs (message->type);
3458   LOG (GNUNET_ERROR_TYPE_INFO,
3459        "--> %s (%s %4u) on conn %s (%p) %s (%u bytes)\n",
3460        GC_m2s (type), GC_m2s (payload_type), payload_id, GCC_2s (c), c,
3461        GC_f2s(fwd), size);
3462
3463   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3464   droppable = GNUNET_NO == force;
3465   switch (type)
3466   {
3467     struct GNUNET_CADET_AX        *axmsg;
3468     struct GNUNET_CADET_Encrypted *emsg;
3469     struct GNUNET_CADET_KX        *kmsg;
3470     struct GNUNET_CADET_ACK       *amsg;
3471     struct GNUNET_CADET_Poll      *pmsg;
3472     struct GNUNET_CADET_ConnectionDestroy *dmsg;
3473     struct GNUNET_CADET_ConnectionBroken  *bmsg;
3474     uint32_t ttl;
3475
3476     case GNUNET_MESSAGE_TYPE_CADET_AX:
3477     case GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED:
3478       if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type)
3479       {
3480         emsg = (struct GNUNET_CADET_Encrypted *) data;
3481         ttl = ntohl (emsg->ttl);
3482         if (0 == ttl)
3483         {
3484           GNUNET_break_op (0);
3485           GNUNET_free (data);
3486           return NULL;
3487         }
3488         emsg->cid = c->id;
3489         emsg->ttl = htonl (ttl - 1);
3490       }
3491       else
3492       {
3493         axmsg = (struct GNUNET_CADET_AX *) data;
3494         axmsg->cid = c->id;
3495       }
3496       LOG (GNUNET_ERROR_TYPE_DEBUG, "  Q_N+ %p %u\n", fc, fc->queue_n);
3497       LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid sent %u\n", fc->last_pid_sent);
3498       LOG (GNUNET_ERROR_TYPE_DEBUG, "     ack recv %u\n", fc->last_ack_recv);
3499       if (GNUNET_YES == droppable)
3500       {
3501         fc->queue_n++;
3502       }
3503       else
3504       {
3505         LOG (GNUNET_ERROR_TYPE_DEBUG, "  not droppable, Q_N stays the same\n");
3506       }
3507       break;
3508
3509     case GNUNET_MESSAGE_TYPE_CADET_KX:
3510       kmsg = (struct GNUNET_CADET_KX *) data;
3511       kmsg->cid = c->id;
3512       break;
3513
3514     case GNUNET_MESSAGE_TYPE_CADET_ACK:
3515       amsg = (struct GNUNET_CADET_ACK *) data;
3516       amsg->cid = c->id;
3517       LOG (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
3518       droppable = GNUNET_NO;
3519       break;
3520
3521     case GNUNET_MESSAGE_TYPE_CADET_POLL:
3522       pmsg = (struct GNUNET_CADET_Poll *) data;
3523       pmsg->cid = c->id;
3524       LOG (GNUNET_ERROR_TYPE_DEBUG, " POLL %u\n", ntohl (pmsg->pid));
3525       droppable = GNUNET_NO;
3526       break;
3527
3528     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY:
3529       dmsg = (struct GNUNET_CADET_ConnectionDestroy *) data;
3530       dmsg->cid = c->id;
3531       break;
3532
3533     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN:
3534       bmsg = (struct GNUNET_CADET_ConnectionBroken *) data;
3535       bmsg->cid = c->id;
3536       break;
3537
3538     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE:
3539     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_ACK:
3540       break;
3541
3542     default:
3543       GNUNET_break (0);
3544       GNUNET_free (data);
3545       return NULL;
3546   }
3547
3548   if (fc->queue_n > fc->queue_max && droppable)
3549   {
3550     GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
3551                               1, GNUNET_NO);
3552     GNUNET_break (0);
3553     LOG (GNUNET_ERROR_TYPE_DEBUG, "queue full: %u/%u\n",
3554          fc->queue_n, fc->queue_max);
3555     if (GNUNET_MESSAGE_TYPE_CADET_ENCRYPTED == type
3556         || GNUNET_MESSAGE_TYPE_CADET_AX == type)
3557     {
3558       fc->queue_n--;
3559     }
3560     GNUNET_free (data);
3561     return NULL; /* Drop this message */
3562   }
3563
3564   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %s %u\n",
3565        GCC_2s (c), c->pending_messages);
3566   c->pending_messages++;
3567
3568   q = GNUNET_new (struct CadetConnectionQueue);
3569   q->forced = !droppable;
3570   q->q = GCP_queue_add (get_hop (c, fwd), data, type, payload_type, payload_id,
3571                         size, c, fwd, &conn_message_sent, q);
3572   if (NULL == q->q)
3573   {
3574     LOG (GNUNET_ERROR_TYPE_DEBUG, "dropping msg on %s, NULL q\n", GCC_2s (c));
3575     GNUNET_free (data);
3576     GNUNET_free (q);
3577     GCC_check_connections ();
3578     return NULL;
3579   }
3580   q->cont = cont;
3581   q->cont_cls = cont_cls;
3582   GCC_check_connections ();
3583   return (NULL == cont) ? NULL : q;
3584 }
3585
3586
3587 /**
3588  * Cancel a previously sent message while it's in the queue.
3589  *
3590  * ONLY can be called before the continuation given to the send function
3591  * is called. Once the continuation is called, the message is no longer in the
3592  * queue.
3593  *
3594  * @param q Handle to the queue.
3595  */
3596 void
3597 GCC_cancel (struct CadetConnectionQueue *q)
3598 {
3599   LOG (GNUNET_ERROR_TYPE_DEBUG, "!  GCC cancel message\n");
3600
3601   /* queue destroy calls message_sent, which calls q->cont and frees q */
3602   GCP_queue_destroy (q->q, GNUNET_YES, GNUNET_NO, 0);
3603   GCC_check_connections ();
3604 }
3605
3606
3607 /**
3608  * Sends a CREATE CONNECTION message for a path to a peer.
3609  * Changes the connection and tunnel states if necessary.
3610  *
3611  * @param connection Connection to create.
3612  */
3613 void
3614 GCC_send_create (struct CadetConnection *connection)
3615 {
3616   enum CadetTunnelCState state;
3617   size_t size;
3618
3619   GCC_check_connections ();
3620   size = sizeof (struct GNUNET_CADET_ConnectionCreate);
3621   size += connection->path->length * sizeof (struct GNUNET_PeerIdentity);
3622
3623   LOG (GNUNET_ERROR_TYPE_INFO, "--> %s on conn %s  (%u bytes)\n",
3624        GC_m2s (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE),
3625        GCC_2s (connection), size);
3626   LOG (GNUNET_ERROR_TYPE_DEBUG, "  C_P+ %p %u (create)\n",
3627        connection, connection->pending_messages);
3628   connection->pending_messages++;
3629
3630   connection->maintenance_q =
3631     GCP_queue_add (get_next_hop (connection), NULL,
3632                    GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE, 0, 0,
3633                    size, connection, GNUNET_YES, &conn_message_sent, NULL);
3634
3635   state = GCT_get_cstate (connection->t);
3636   if (CADET_TUNNEL_SEARCHING == state || CADET_TUNNEL_NEW == state)
3637     GCT_change_cstate (connection->t, CADET_TUNNEL_WAITING);
3638   if (CADET_CONNECTION_NEW == connection->state)
3639     connection_change_state (connection, CADET_CONNECTION_SENT);
3640   GCC_check_connections ();
3641 }
3642
3643
3644 /**
3645  * Send a message to all peers in this connection that the connection
3646  * is no longer valid.
3647  *
3648  * If some peer should not receive the message, it should be zero'ed out
3649  * before calling this function.
3650  *
3651  * @param c The connection whose peers to notify.
3652  */
3653 void
3654 GCC_send_destroy (struct CadetConnection *c)
3655 {
3656   struct GNUNET_CADET_ConnectionDestroy msg;
3657
3658   if (GNUNET_YES == c->destroy)
3659     return;
3660   GCC_check_connections ();
3661   msg.header.size = htons (sizeof (msg));
3662   msg.header.type = htons (GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY);
3663   msg.cid = c->id;
3664   LOG (GNUNET_ERROR_TYPE_DEBUG,
3665               "  sending connection destroy for connection %s\n",
3666               GCC_2s (c));
3667
3668   if (GNUNET_NO == GCC_is_terminal (c, GNUNET_YES))
3669     GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg.header, 0, 0, c,
3670                                                       GNUNET_YES, GNUNET_YES,
3671                                                       NULL, NULL));
3672   if (GNUNET_NO == GCC_is_terminal (c, GNUNET_NO))
3673     GNUNET_assert (NULL == GCC_send_prebuilt_message (&msg.header, 0, 0, c,
3674                                                       GNUNET_NO, GNUNET_YES,
3675                                                       NULL, NULL));
3676   c->destroy = GNUNET_YES;
3677   c->state = CADET_CONNECTION_DESTROYED;
3678   GCC_check_connections ();
3679 }
3680
3681
3682 /**
3683  * @brief Start a polling timer for the connection.
3684  *
3685  * When a neighbor does not accept more traffic on the connection it could be
3686  * caused by a simple congestion or by a lost ACK. Polling enables to check
3687  * for the lastest ACK status for a connection.
3688  *
3689  * @param c Connection.
3690  * @param fwd Should we poll in the FWD direction?
3691  */
3692 void
3693 GCC_start_poll (struct CadetConnection *c, int fwd)
3694 {
3695   struct CadetFlowControl *fc;
3696
3697   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3698   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL %s requested\n",
3699        GC_f2s (fwd));
3700   if (NULL != fc->poll_task || NULL != fc->poll_msg)
3701   {
3702     LOG (GNUNET_ERROR_TYPE_DEBUG, "  POLL not needed (%p, %p)\n",
3703          fc->poll_task, fc->poll_msg);
3704     return;
3705   }
3706   LOG (GNUNET_ERROR_TYPE_DEBUG, "POLL started on request\n");
3707   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
3708                                                 &connection_poll,
3709                                                 fc);
3710 }
3711
3712
3713 /**
3714  * @brief Stop polling a connection for ACKs.
3715  *
3716  * Once we have enough ACKs for future traffic, polls are no longer necessary.
3717  *
3718  * @param c Connection.
3719  * @param fwd Should we stop the poll in the FWD direction?
3720  */
3721 void
3722 GCC_stop_poll (struct CadetConnection *c, int fwd)
3723 {
3724   struct CadetFlowControl *fc;
3725
3726   fc = fwd ? &c->fwd_fc : &c->bck_fc;
3727   if (NULL != fc->poll_task)
3728   {
3729     GNUNET_SCHEDULER_cancel (fc->poll_task);
3730     fc->poll_task = NULL;
3731   }
3732 }
3733
3734
3735 /**
3736  * Get a (static) string for a connection.
3737  *
3738  * @param c Connection.
3739  */
3740 const char *
3741 GCC_2s (const struct CadetConnection *c)
3742 {
3743   if (NULL == c)
3744     return "NULL";
3745
3746   if (NULL != c->t)
3747   {
3748     static char buf[128];
3749
3750     SPRINTF (buf, "%s (->%s)",
3751              GNUNET_h2s (GC_h2hc (GCC_get_id (c))), GCT_2s (c->t));
3752     return buf;
3753   }
3754   return GNUNET_h2s (GC_h2hc (&c->id));
3755 }
3756
3757
3758 /**
3759  * Log all possible info about the connection state.
3760  *
3761  * @param c Connection to debug.
3762  * @param level Debug level to use.
3763  */
3764 void
3765 GCC_debug (const struct CadetConnection *c, enum GNUNET_ErrorType level)
3766 {
3767   int do_log;
3768   char *s;
3769
3770   do_log = GNUNET_get_log_call_status (level & (~GNUNET_ERROR_TYPE_BULK),
3771                                        "cadet-con",
3772                                        __FILE__, __FUNCTION__, __LINE__);
3773   if (0 == do_log)
3774     return;
3775
3776   if (NULL == c)
3777   {
3778     LOG2 (level, "CCC DEBUG NULL CONNECTION\n");
3779     return;
3780   }
3781
3782   LOG2 (level, "CCC DEBUG CONNECTION %s\n", GCC_2s (c));
3783   s = path_2s (c->path);
3784   LOG2 (level, "CCC  path %s, own pos: %u\n", s, c->own_pos);
3785   GNUNET_free (s);
3786   LOG2 (level, "CCC  state: %s, destroy: %u\n",
3787         GCC_state2s (c->state), c->destroy);
3788   LOG2 (level, "CCC  pending messages: %u\n", c->pending_messages);
3789   if (NULL != c->perf)
3790     LOG2 (level, "CCC  us/byte: %f\n", c->perf->avg);
3791
3792   LOG2 (level, "CCC  FWD flow control:\n");
3793   LOG2 (level, "CCC   queue: %u/%u\n", c->fwd_fc.queue_n, c->fwd_fc.queue_max);
3794   LOG2 (level, "CCC   last PID sent: %5u, recv: %5u\n",
3795         c->fwd_fc.last_pid_sent, c->fwd_fc.last_pid_recv);
3796   LOG2 (level, "CCC   last ACK sent: %5u, recv: %5u\n",
3797         c->fwd_fc.last_ack_sent, c->fwd_fc.last_ack_recv);
3798   LOG2 (level, "CCC   recv PID bitmap: %X\n", c->fwd_fc.recv_bitmap);
3799   LOG2 (level, "CCC   poll: task %d, msg  %p, msg_ack %p)\n",
3800         c->fwd_fc.poll_task, c->fwd_fc.poll_msg, c->fwd_fc.ack_msg);
3801
3802   LOG2 (level, "CCC  BCK flow control:\n");
3803   LOG2 (level, "CCC   queue: %u/%u\n", c->bck_fc.queue_n, c->bck_fc.queue_max);
3804   LOG2 (level, "CCC   last PID sent: %5u, recv: %5u\n",
3805         c->bck_fc.last_pid_sent, c->bck_fc.last_pid_recv);
3806   LOG2 (level, "CCC   last ACK sent: %5u, recv: %5u\n",
3807         c->bck_fc.last_ack_sent, c->bck_fc.last_ack_recv);
3808   LOG2 (level, "CCC   recv PID bitmap: %X\n", c->bck_fc.recv_bitmap);
3809   LOG2 (level, "CCC   poll: task %d, msg  %p, msg_ack %p)\n",
3810         c->bck_fc.poll_task, c->bck_fc.poll_msg, c->bck_fc.ack_msg);
3811
3812   LOG2 (level, "CCC DEBUG CONNECTION END\n");
3813 }