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