src/cadet/gnunet-service-cadet_connection.c

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