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