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