src/cadet/gnunet-service-cadet_connection.c

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