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