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_core_service.h"
  31
  32 #include "gnunet-service-mesh_connection.h"
  33 #include "gnunet-service-mesh_peer.h"
  34 #include "gnunet-service-mesh_local.h"
  35 #include "mesh_protocol_enc.h"
  36 #include "mesh_path.h"
  37
  38
  39 #define MESH_DEBUG_CONNECTION   GNUNET_NO
  40 #define MESH_MAX_POLL_TIME      GNUNET_TIME_relative_multiply (\
  41                                   GNUNET_TIME_UNIT_MINUTES,\
  42                                   10)
  43 #define MESH_RETRANSMIT_TIME    GNUNET_TIME_UNIT_SECONDS
  44 #define MESH_RETRANSMIT_MARGIN  4
  45
  46
  47 #if MESH_DEBUG_CONNECTION
  48 #define DEBUG_CONN(...) GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
  49 #else
  50 #define DEBUG_CONN(...)
  51 #endif
  52
  53
  54 /**
  55  * All the states a connection can be in.
  56  */
  57 enum MeshConnectionState
  58 {
  59   /**
  60    * Uninitialized status, should never appear in operation.
  61    */
  62   MESH_CONNECTION_NEW,
  63
  64   /**
  65    * Connection create message sent, waiting for ACK.
  66    */
  67   MESH_CONNECTION_SENT,
  68
  69   /**
  70    * Connection ACK sent, waiting for ACK.
  71    */
  72   MESH_CONNECTION_ACK,
  73
  74   /**
  75    * Connection confirmed, ready to carry traffic.
  76    */
  77   MESH_CONNECTION_READY,
  78 };
  79
  80
  81
  82 /**
  83  * Struct containing info about a queued transmission to this peer
  84  */
  85 struct MeshPeerQueue
  86 {
  87     /**
  88       * DLL next
  89       */
  90   struct MeshPeerQueue *next;
  91
  92     /**
  93       * DLL previous
  94       */
  95   struct MeshPeerQueue *prev;
  96
  97     /**
  98      * Peer this transmission is directed to.
  99      */
 100   struct MeshPeer *peer;
 101
 102     /**
 103      * Connection this message belongs to.
 104      */
 105   struct MeshConnection *c;
 106
 107     /**
 108      * Is FWD in c?
 109      */
 110   int fwd;
 111
 112     /**
 113      * Channel this message belongs to, if known.
 114      */
 115   struct MeshChannel *ch;
 116
 117     /**
 118      * Pointer to info stucture used as cls.
 119      */
 120   void *cls;
 121
 122     /**
 123      * Type of message
 124      */
 125   uint16_t type;
 126
 127     /**
 128      * Size of the message
 129      */
 130   size_t size;
 131 };
 132
 133
 134 /**
 135  * Struct to encapsulate all the Flow Control information to a peer to which
 136  * we are directly connected (on a core level).
 137  */
 138 struct MeshFlowControl
 139 {
 140   /**
 141    * Connection this controls.
 142    */
 143   struct MeshConnection *c;
 144
 145   /**
 146    * How many messages are in the queue on this connection.
 147    */
 148   unsigned int queue_n;
 149
 150   /**
 151    * How many messages do we accept in the queue.
 152    */
 153   unsigned int queue_max;
 154
 155   /**
 156    * Next ID to use.
 157    */
 158   uint32_t next_pid;
 159
 160   /**
 161    * ID of the last packet sent towards the peer.
 162    */
 163   uint32_t last_pid_sent;
 164
 165   /**
 166    * ID of the last packet received from the peer.
 167    */
 168   uint32_t last_pid_recv;
 169
 170   /**
 171    * Last ACK sent to the peer (peer can't send more than this PID).
 172    */
 173   uint32_t last_ack_sent;
 174
 175   /**
 176    * Last ACK sent towards the origin (for traffic towards leaf node).
 177    */
 178   uint32_t last_ack_recv;
 179
 180   /**
 181    * Task to poll the peer in case of a lost ACK causes stall.
 182    */
 183   GNUNET_SCHEDULER_TaskIdentifier poll_task;
 184
 185   /**
 186    * How frequently to poll for ACKs.
 187    */
 188   struct GNUNET_TIME_Relative poll_time;
 189 };
 190
 191
 192 /**
 193  * Struct containing all information regarding a connection to a peer.
 194  */
 195 struct MeshConnection
 196 {
 197   /**
 198    * DLL
 199    */
 200   struct MeshConnection *next;
 201   struct MeshConnection *prev;
 202
 203   /**
 204    * Tunnel this connection is part of.
 205    */
 206   struct MeshTunnel2 *t;
 207
 208   /**
 209    * Flow control information for traffic fwd.
 210    */
 211   struct MeshFlowControl fwd_fc;
 212
 213   /**
 214    * Flow control information for traffic bck.
 215    */
 216   struct MeshFlowControl bck_fc;
 217
 218   /**
 219    * ID of the connection.
 220    */
 221   struct GNUNET_HashCode id;
 222
 223   /**
 224    * State of the connection.
 225    */
 226   enum MeshConnectionState state;
 227
 228   /**
 229    * Path being used for the tunnel.
 230    */
 231   struct MeshPeerPath *path;
 232
 233   /**
 234    * Position of the local peer in the path.
 235    */
 236   unsigned int own_pos;
 237
 238   /**
 239    * Task to keep the used paths alive at the owner,
 240    * time tunnel out on all the other peers.
 241    */
 242   GNUNET_SCHEDULER_TaskIdentifier fwd_maintenance_task;
 243
 244   /**
 245    * Task to keep the used paths alive at the destination,
 246    * time tunnel out on all the other peers.
 247    */
 248   GNUNET_SCHEDULER_TaskIdentifier bck_maintenance_task;
 249
 250   /**
 251    * Pending message count.
 252    */
 253   int pending_messages;
 254
 255   /**
 256    * Destroy flag: if true, destroy on last message.
 257    */
 258   int destroy;
 259 };
 260
 261
 262 /**
 263  * Connections known, indexed by cid (MeshConnection).
 264  */
 265 static struct GNUNET_CONTAINER_MultiHashMap *connections;
 266
 267 /**
 268  * How many connections are we willing to maintain.
 269  * Local connections are always allowed, even if there are more connections than max.
 270  */
 271 static unsigned long long max_connections;
 272
 273 /**
 274  * How many messages *in total* are we willing to queue, divide by number of
 275  * connections to get connection queue size.
 276  */
 277 static unsigned long long max_msgs_queue;
 278
 279 /**
 280  * How often to send path keepalives. Paths timeout after 4 missed.
 281  */
 282 static struct GNUNET_TIME_Relative refresh_connection_time;
 283
 284 /**
 285  * Handle to communicate with core.
 286  */
 287 static struct GNUNET_CORE_Handle *core_handle;
 288
 289
 290
 291 /**
 292  * Get string description for tunnel state.
 293  *
 294  * @param s Tunnel state.
 295  *
 296  * @return String representation.
 297  */
 298 static const char *
 299 GMC_DEBUG_state2s (enum MeshTunnelState s)
 300 {
 301   switch (s)
 302   {
 303     case MESH_CONNECTION_NEW:
 304       return "MESH_CONNECTION_NEW";
 305     case MESH_CONNECTION_SENT:
 306       return "MESH_CONNECTION_SENT";
 307     case MESH_CONNECTION_ACK:
 308       return "MESH_CONNECTION_ACK";
 309     case MESH_CONNECTION_READY:
 310       return "MESH_CONNECTION_READY";
 311     default:
 312       return "MESH_CONNECTION_STATE_ERROR";
 313   }
 314 }
 315
 316
 317
 318 /**
 319  * Initialize a Flow Control structure to the initial state.
 320  *
 321  * @param fc Flow Control structure to initialize.
 322  */
 323 static void
 324 fc_init (struct MeshFlowControl *fc)
 325 {
 326   fc->next_pid = 0;
 327   fc->last_pid_sent = (uint32_t) -1; /* Next (expected) = 0 */
 328   fc->last_pid_recv = (uint32_t) -1;
 329   fc->last_ack_sent = (uint32_t) 0;
 330   fc->last_ack_recv = (uint32_t) 0;
 331   fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
 332   fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
 333   fc->queue_n = 0;
 334   fc->queue_max = (max_msgs_queue / max_connections) + 1;
 335 }
 336
 337
 338 /**
 339  * Find a connection.
 340  *
 341  * @param cid Connection ID.
 342  */
 343 static struct MeshConnection *
 344 connection_get (const struct GNUNET_HashCode *cid)
 345 {
 346   return GNUNET_CONTAINER_multihashmap_get (connections, cid);
 347 }
 348
 349
 350 /**
 351  * Get first sendable message.
 352  *
 353  * @param peer The destination peer.
 354  *
 355  * @return Best current known path towards the peer, if any.
 356  */
 357 static struct MeshPeerQueue *
 358 peer_get_first_message (const struct MeshPeer *peer)
 359 {
 360   struct MeshPeerQueue *q;
 361
 362   for (q = peer->queue_head; NULL != q; q = q->next)
 363   {
 364     if (queue_is_sendable (q))
 365       return q;
 366   }
 367
 368   return NULL;
 369 }
 370
 371
 372 static int
 373 queue_is_sendable (struct MeshPeerQueue *q)
 374 {
 375   struct MeshFlowControl *fc;
 376
 377   /* Is PID-independent? */
 378   switch (q->type)
 379   {
 380     case GNUNET_MESSAGE_TYPE_MESH_ACK:
 381     case GNUNET_MESSAGE_TYPE_MESH_POLL:
 382       return GNUNET_YES;
 383   }
 384
 385   /* Is PID allowed? */
 386   fc = q->fwd ? &q->c->fwd_fc : &q->c->bck_fc;
 387   if (GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
 388     return GNUNET_YES;
 389
 390   return GNUNET_NO;
 391 }
 392
 393
 394
 395 /**
 396  * Free a transmission that was already queued with all resources
 397  * associated to the request.
 398  *
 399  * @param queue Queue handler to cancel.
 400  * @param clear_cls Is it necessary to free associated cls?
 401  */
 402 static void
 403 queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
 404 {
 405   struct MeshPeer *peer;
 406   struct MeshFlowControl *fc;
 407   int fwd;
 408
 409   fwd = queue->fwd;
 410   peer = queue->peer;
 411   GNUNET_assert (NULL != queue->c);
 412   fc = fwd ? &queue->c->fwd_fc : &queue->c->bck_fc;
 413
 414   if (GNUNET_YES == clear_cls)
 415   {
 416     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   queue destroy type %s\n",
 417                 GNUNET_MESH_DEBUG_M2S (queue->type));
 418     switch (queue->type)
 419     {
 420       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
 421       case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
 422         GNUNET_log (GNUNET_ERROR_TYPE_INFO, "destroying a DESTROY message\n");
 423         GNUNET_break (GNUNET_YES == queue->c->destroy);
 424         /* fall through */
 425       case GNUNET_MESSAGE_TYPE_MESH_FWD:
 426       case GNUNET_MESSAGE_TYPE_MESH_BCK:
 427       case GNUNET_MESSAGE_TYPE_MESH_ACK:
 428       case GNUNET_MESSAGE_TYPE_MESH_POLL:
 429       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
 430       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
 431       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
 432         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   prebuilt message\n");;
 433         GNUNET_free_non_null (queue->cls);
 434         break;
 435
 436       default:
 437         GNUNET_break (0);
 438         GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "   type %s unknown!\n",
 439                     GNUNET_MESH_DEBUG_M2S (queue->type));
 440     }
 441
 442   }
 443   GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
 444
 445   if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
 446       queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
 447   {
 448     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Q_N- %p %u\n", fc, fc->queue_n);
 449     fc->queue_n--;
 450     peer->queue_n--;
 451   }
 452   if (NULL != queue->c)
 453   {
 454     queue->c->pending_messages--;
 455     if (NULL != queue->c->t)
 456     {
 457       queue->c->t->pending_messages--;
 458     }
 459   }
 460
 461   GNUNET_free (queue);
 462 }
 463
 464
 465
 466 static size_t
 467 queue_send (void *cls, size_t size, void *buf)
 468 {
 469   struct MeshPeer *peer = cls;
 470   struct MeshFlowControl *fc;
 471   struct MeshConnection *c;
 472   struct GNUNET_MessageHeader *msg;
 473   struct MeshPeerQueue *queue;
 474   struct MeshTunnel2 *t;
 475   struct MeshChannel *ch;
 476   const struct GNUNET_PeerIdentity *dst_id;
 477   size_t data_size;
 478   uint32_t pid;
 479   uint16_t type;
 480   int fwd;
 481
 482   peer->core_transmit = NULL;
 483   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Queue send (max %u)\n", size);
 484
 485   if (NULL == buf || 0 == size)
 486   {
 487     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "* Buffer size 0.\n");
 488     return 0;
 489   }
 490
 491   /* Initialize */
 492   queue = peer_get_first_message (peer);
 493   if (NULL == queue)
 494   {
 495     GNUNET_break (0); /* Core tmt_rdy should've been canceled */
 496     return 0;
 497   }
 498   c = queue->c;
 499   fwd = queue->fwd;
 500   fc = fwd ? &c->fwd_fc : &c->bck_fc;
 501
 502   dst_id = GNUNET_PEER_resolve2 (peer->id);
 503   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   towards %s\n", GNUNET_i2s (dst_id));
 504   /* Check if buffer size is enough for the message */
 505   if (queue->size > size)
 506   {
 507       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   not enough room, reissue\n");
 508       peer->core_transmit =
 509           GNUNET_CORE_notify_transmit_ready (core_handle,
 510                                              GNUNET_NO,
 511                                              0,
 512                                              GNUNET_TIME_UNIT_FOREVER_REL,
 513                                              dst_id,
 514                                              queue->size,
 515                                              &queue_send,
 516                                              peer);
 517       return 0;
 518   }
 519   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   size %u ok\n", queue->size);
 520
 521   t = (NULL != c) ? c->t : NULL;
 522   type = 0;
 523
 524   /* Fill buf */
 525   switch (queue->type)
 526   {
 527     case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
 528     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
 529     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
 530     case GNUNET_MESSAGE_TYPE_MESH_FWD:
 531     case GNUNET_MESSAGE_TYPE_MESH_BCK:
 532     case GNUNET_MESSAGE_TYPE_MESH_ACK:
 533     case GNUNET_MESSAGE_TYPE_MESH_POLL:
 534       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 535                   "*   raw: %s\n",
 536                   GNUNET_MESH_DEBUG_M2S (queue->type));
 537       data_size = send_core_data_raw (queue->cls, size, buf);
 538       msg = (struct GNUNET_MessageHeader *) buf;
 539       type = ntohs (msg->type);
 540       break;
 541     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
 542       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   path create\n");
 543       if (GMC_is_origin (c, GNUNET_YES))
 544         data_size = send_core_connection_create (queue->c, size, buf);
 545       else
 546         data_size = send_core_data_raw (queue->cls, size, buf);
 547       break;
 548     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
 549       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   path ack\n");
 550       if (GMC_is_origin (c, GNUNET_NO) ||
 551           GMC_is_origin (c, GNUNET_YES))
 552         data_size = send_core_connection_ack (queue->c, size, buf);
 553       else
 554         data_size = send_core_data_raw (queue->cls, size, buf);
 555       break;
 556     case GNUNET_MESSAGE_TYPE_MESH_DATA:
 557     case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
 558     case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
 559       /* This should be encapsulted */
 560       GNUNET_break (0);
 561       data_size = 0;
 562       break;
 563     default:
 564       GNUNET_break (0);
 565       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "*   type unknown: %u\n",
 566                   queue->type);
 567       data_size = 0;
 568   }
 569
 570   if (0 < drop_percent &&
 571       GNUNET_CRYPTO_random_u32(GNUNET_CRYPTO_QUALITY_WEAK, 101) < drop_percent)
 572   {
 573     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 574                 "Dropping message of type %s\n",
 575                 GNUNET_MESH_DEBUG_M2S (queue->type));
 576     data_size = 0;
 577   }
 578
 579   /* Free queue, but cls was freed by send_core_* */
 580   ch = queue->ch;
 581   queue_destroy (queue, GNUNET_NO);
 582
 583   /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
 584   switch (type)
 585   {
 586     case GNUNET_MESSAGE_TYPE_MESH_FWD:
 587     case GNUNET_MESSAGE_TYPE_MESH_BCK:
 588       pid = ntohl ( ((struct GNUNET_MESH_Encrypted *) buf)->pid );
 589       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   accounting pid %u\n", pid);
 590       fc->last_pid_sent = pid;
 591       send_ack (c, ch, fwd);
 592       break;
 593     default:
 594       break;
 595   }
 596
 597   /* If more data in queue, send next */
 598   queue = peer_get_first_message (peer);
 599   if (NULL != queue)
 600   {
 601     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   more data!\n");
 602     if (NULL == peer->core_transmit) {
 603       peer->core_transmit =
 604           GNUNET_CORE_notify_transmit_ready(core_handle,
 605                                             0,
 606                                             0,
 607                                             GNUNET_TIME_UNIT_FOREVER_REL,
 608                                             dst_id,
 609                                             queue->size,
 610                                             &queue_send,
 611                                             peer);
 612     }
 613     else
 614     {
 615       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 616                   "*   tmt rdy called somewhere else\n");
 617     }
 618     if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
 619     {
 620       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*   starting poll timeout\n");
 621       fc->poll_task =
 622           GNUNET_SCHEDULER_add_delayed (fc->poll_time, &connection_poll, fc);
 623     }
 624   }
 625   else
 626   {
 627     if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
 628     {
 629       GNUNET_SCHEDULER_cancel (fc->poll_task);
 630       fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
 631     }
 632   }
 633   if (NULL != c)
 634   {
 635     c->pending_messages--;
 636     if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
 637     {
 638       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*  destroying connection!\n");
 639       GMC_destroy (c);
 640     }
 641   }
 642
 643   if (NULL != t)
 644   {
 645     t->pending_messages--;
 646     if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
 647     {
 648 //       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*  destroying tunnel!\n");
 649       tunnel_destroy (t);
 650     }
 651   }
 652   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*  Return %d\n", data_size);
 653   return data_size;
 654 }
 655
 656
 657
 658 static void
 659 queue_add (void *cls, uint16_t type, size_t size,
 660            struct MeshConnection *c,
 661            struct MeshChannel *ch,
 662            int fwd)
 663 {
 664   struct MeshPeerQueue *queue;
 665   struct MeshFlowControl *fc;
 666   struct MeshPeer *peer;
 667   int priority;
 668   int call_core;
 669
 670   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 671               "queue add %s %s (%u) on c %p, ch %p\n",
 672               fwd ? "FWD" : "BCK",  GNUNET_MESH_DEBUG_M2S (type), size, c, ch);
 673   GNUNET_assert (NULL != c);
 674
 675   fc   = fwd ? &c->fwd_fc : &c->bck_fc;
 676   peer = fwd ? connection_get_next_hop (c) : connection_get_prev_hop (c);
 677
 678   if (NULL == fc)
 679   {
 680     GNUNET_break (0);
 681     return;
 682   }
 683
 684   if (NULL == peer->connections)
 685   {
 686     /* We are not connected to this peer, ignore request. */
 687     GNUNET_break_op (0);
 688     return;
 689   }
 690
 691   priority = 0;
 692
 693   if (GNUNET_MESSAGE_TYPE_MESH_POLL == type ||
 694       GNUNET_MESSAGE_TYPE_MESH_ACK == type)
 695   {
 696     priority = 100;
 697   }
 698
 699   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "priority %d\n", priority);
 700   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "fc %p\n", fc);
 701   if (fc->queue_n >= fc->queue_max && 0 == priority)
 702   {
 703     GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
 704                               1, GNUNET_NO);
 705     GNUNET_break (0);
 706     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 707                 "queue full: %u/%u\n",
 708                 fc->queue_n, fc->queue_max);
 709     return; /* Drop this message */
 710   }
 711
 712   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "last pid %u\n", fc->last_pid_sent);
 713   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "     ack %u\n", fc->last_ack_recv);
 714   if (GMC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
 715   {
 716     call_core = GNUNET_NO;
 717     if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task &&
 718         GNUNET_MESSAGE_TYPE_MESH_POLL != type)
 719     {
 720       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 721                   "no buffer space (%u > %u): starting poll\n",
 722                   fc->last_pid_sent + 1, fc->last_ack_recv);
 723       fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
 724                                                     &connection_poll,
 725                                                     fc);
 726     }
 727   }
 728   else
 729     call_core = GNUNET_YES;
 730   queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
 731   queue->cls = cls;
 732   queue->type = type;
 733   queue->size = size;
 734   queue->peer = peer;
 735   queue->c = c;
 736   queue->ch = ch;
 737   queue->fwd = fwd;
 738   if (100 <= priority)
 739   {
 740     struct MeshPeerQueue *copy;
 741     struct MeshPeerQueue *next;
 742
 743     for (copy = peer->queue_head; NULL != copy; copy = next)
 744     {
 745       next = copy->next;
 746       if (copy->type == type && copy->c == c && copy->fwd == fwd)
 747       {
 748         /* Example: also a FWD ACK for connection XYZ */
 749         queue_destroy (copy, GNUNET_YES);
 750       }
 751     }
 752     GNUNET_CONTAINER_DLL_insert (peer->queue_head, peer->queue_tail, queue);
 753   }
 754   else
 755   {
 756     GNUNET_CONTAINER_DLL_insert_tail (peer->queue_head, peer->queue_tail, queue);
 757     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Q_N+ %p %u\n", fc, fc->queue_n);
 758     fc->queue_n++;
 759     peer->queue_n++;
 760   }
 761
 762   if (NULL == peer->core_transmit && GNUNET_YES == call_core)
 763   {
 764     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 765                 "calling core tmt rdy towards %s for %u bytes\n",
 766                 peer2s (peer), size);
 767     peer->core_transmit =
 768         GNUNET_CORE_notify_transmit_ready (core_handle,
 769                                            0,
 770                                            0,
 771                                            GNUNET_TIME_UNIT_FOREVER_REL,
 772                                            GNUNET_PEER_resolve2 (peer->id),
 773                                            size,
 774                                            &queue_send,
 775                                            peer);
 776   }
 777   else
 778   {
 779     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 780                 "core tmt rdy towards %s already called\n",
 781                 peer2s (peer));
 782
 783   }
 784   c->pending_messages++;
 785   if (NULL != c->t)
 786     c->t->pending_messages++;
 787 }
 788
 789
 790
 791
 792 /**
 793  * Sends an already built message on a connection, properly registering
 794  * all used resources.
 795  *
 796  * @param message Message to send. Function makes a copy of it.
 797  *                If message is not hop-by-hop, decrements TTL of copy.
 798  * @param c Connection on which this message is transmitted.
 799  * @param ch Channel on which this message is transmitted, or NULL.
 800  * @param fwd Is this a fwd message?
 801  */
 802 static void
 803 send_prebuilt_message_connection (const struct GNUNET_MessageHeader *message,
 804                                   struct MeshConnection *c,
 805                                   struct MeshChannel *ch,
 806                                   int fwd)
 807 {
 808   void *data;
 809   size_t size;
 810   uint16_t type;
 811
 812   size = ntohs (message->size);
 813   data = GNUNET_malloc (size);
 814   memcpy (data, message, size);
 815   type = ntohs (message->type);
 816   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Send %s (%u) on connection %s\n",
 817               GNUNET_MESH_DEBUG_M2S (type), size, GNUNET_h2s (&c->id));
 818
 819   switch (type)
 820   {
 821     struct GNUNET_MESH_Encrypted *emsg;
 822     struct GNUNET_MESH_ACK       *amsg;
 823     struct GNUNET_MESH_Poll      *pmsg;
 824     struct GNUNET_MESH_ConnectionDestroy *dmsg;
 825     struct GNUNET_MESH_ConnectionBroken  *bmsg;
 826     uint32_t ttl;
 827
 828     case GNUNET_MESSAGE_TYPE_MESH_FWD:
 829     case GNUNET_MESSAGE_TYPE_MESH_BCK:
 830       emsg = (struct GNUNET_MESH_Encrypted *) data;
 831       ttl = ntohl (emsg->ttl);
 832       if (0 == ttl)
 833       {
 834         GNUNET_break_op (0);
 835         return;
 836       }
 837       emsg->cid = c->id;
 838       emsg->ttl = htonl (ttl - 1);
 839       emsg->pid = htonl (fwd ? c->fwd_fc.next_pid++ : c->bck_fc.next_pid++);
 840       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " pid %u\n", ntohl (emsg->pid));
 841       break;
 842
 843     case GNUNET_MESSAGE_TYPE_MESH_ACK:
 844       amsg = (struct GNUNET_MESH_ACK *) data;
 845       amsg->cid = c->id;
 846       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ack %u\n", ntohl (amsg->ack));
 847       break;
 848
 849     case GNUNET_MESSAGE_TYPE_MESH_POLL:
 850       pmsg = (struct GNUNET_MESH_Poll *) data;
 851       pmsg->cid = c->id;
 852       pmsg->pid = htonl (fwd ? c->fwd_fc.last_pid_sent : c->bck_fc.last_pid_sent);
 853       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " poll %u\n", ntohl (pmsg->pid));
 854       break;
 855
 856     case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
 857       dmsg = (struct GNUNET_MESH_ConnectionDestroy *) data;
 858       dmsg->cid = c->id;
 859       dmsg->reserved = 0;
 860       break;
 861
 862     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
 863       bmsg = (struct GNUNET_MESH_ConnectionBroken *) data;
 864       bmsg->cid = c->id;
 865       bmsg->reserved = 0;
 866       break;
 867
 868     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
 869     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
 870       break;
 871
 872     default:
 873       GNUNET_break (0);
 874   }
 875
 876   queue_add (data,
 877              type,
 878              size,
 879              c,
 880              ch,
 881              fwd);
 882 }
 883
 884
 885
 886
 887 struct MeshConnection *
 888 GMC_new (const struct GNUNET_HashCode *cid)
 889 {
 890   struct MeshConnection *c;
 891
 892   c = GNUNET_new (struct MeshConnection);
 893   c->id = *cid;
 894   GNUNET_CONTAINER_multihashmap_put (connections, &c->id, c,
 895                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 896   fc_init (&c->fwd_fc);
 897   fc_init (&c->bck_fc);
 898   c->fwd_fc.c = c;
 899   c->bck_fc.c = c;
 900
 901   return c;
 902 }
 903
 904
 905 static void
 906 GMC_destroy (struct MeshConnection *c)
 907 {
 908   struct MeshPeer *peer;
 909
 910   if (NULL == c)
 911     return;
 912
 913   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying connection %s[%X]\n",
 914               peer2s (c->t->peer),
 915               c->id);
 916
 917   /* Cancel all traffic */
 918   connection_cancel_queues (c, GNUNET_YES);
 919   connection_cancel_queues (c, GNUNET_NO);
 920
 921   /* Cancel maintainance task (keepalive/timeout) */
 922   if (GNUNET_SCHEDULER_NO_TASK != c->fwd_maintenance_task)
 923     GNUNET_SCHEDULER_cancel (c->fwd_maintenance_task);
 924   if (GNUNET_SCHEDULER_NO_TASK != c->bck_maintenance_task)
 925     GNUNET_SCHEDULER_cancel (c->bck_maintenance_task);
 926
 927   /* Deregister from neighbors */
 928   peer = connection_get_next_hop (c);
 929   if (NULL != peer && NULL != peer->connections)
 930     GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
 931   peer = connection_get_prev_hop (c);
 932   if (NULL != peer && NULL != peer->connections)
 933     GNUNET_CONTAINER_multihashmap_remove (peer->connections, &c->id, c);
 934
 935   /* Delete */
 936   GNUNET_STATISTICS_update (stats, "# connections", -1, GNUNET_NO);
 937   GNUNET_CONTAINER_DLL_remove (c->t->connection_head, c->t->connection_tail, c);
 938   GNUNET_free (c);
 939 }
 940
 941
 942
 943 /**
 944  * Send an ACK informing the predecessor about the available buffer space.
 945  *
 946  * Note that for fwd ack, the FWD mean forward *traffic* (root->dest),
 947  * the ACK itself goes "back" (dest->root).
 948  *
 949  * @param c Connection on which to send the ACK.
 950  * @param buffer How much space free to advertise?
 951  * @param fwd Is this FWD ACK? (Going dest->owner)
 952  */
 953 static void
 954 connection_send_ack (struct MeshConnection *c, unsigned int buffer, int fwd)
 955 {
 956   struct MeshFlowControl *next_fc;
 957   struct MeshFlowControl *prev_fc;
 958   struct GNUNET_MESH_ACK msg;
 959   uint32_t ack;
 960   int delta;
 961
 962   next_fc = fwd ? &c->fwd_fc : &c->bck_fc;
 963   prev_fc = fwd ? &c->bck_fc : &c->fwd_fc;
 964
 965   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 966               "connection send %s ack on %s\n",
 967               fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
 968
 969   /* Check if we need to transmit the ACK */
 970   if (prev_fc->last_ack_sent - prev_fc->last_pid_recv > 3)
 971   {
 972     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending ACK, buffer > 3\n");
 973     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 974                 "  last pid recv: %u, last ack sent: %u\n",
 975                 prev_fc->last_pid_recv, prev_fc->last_ack_sent);
 976     return;
 977   }
 978
 979   /* Ok, ACK might be necessary, what PID to ACK? */
 980   delta = next_fc->queue_max - next_fc->queue_n;
 981   ack = prev_fc->last_pid_recv + delta;
 982   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ACK %u\n", ack);
 983   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 984               " last pid %u, last ack %u, qmax %u, q %u\n",
 985               prev_fc->last_pid_recv, prev_fc->last_ack_sent,
 986               next_fc->queue_max, next_fc->queue_n);
 987   if (ack == prev_fc->last_ack_sent)
 988   {
 989     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Not sending FWD ACK, not needed\n");
 990     return;
 991   }
 992
 993   prev_fc->last_ack_sent = ack;
 994
 995   /* Build ACK message and send on connection */
 996   msg.header.size = htons (sizeof (msg));
 997   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_ACK);
 998   msg.ack = htonl (ack);
 999   msg.cid = c->id;
1000
1001   send_prebuilt_message_connection (&msg.header, c, NULL, !fwd);
1002 }
1003
1004
1005 static void
1006 connection_change_state (struct MeshConnection* c,
1007                          enum MeshConnectionState state)
1008 {
1009   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1010               "Connection %s state was %s\n",
1011               GNUNET_h2s (&c->id), GNUNET_MESH_DEBUG_CS2S (c->state));
1012   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1013               "Connection %s state is now %s\n",
1014               GNUNET_h2s (&c->id), GNUNET_MESH_DEBUG_CS2S (state));
1015   c->state = state;
1016 }
1017
1018
1019
1020 /**
1021  * Send keepalive packets for a connection.
1022  *
1023  * @param c Connection to keep alive..
1024  * @param fwd Is this a FWD keepalive? (owner -> dest).
1025  */
1026 static void
1027 connection_keepalive (struct MeshConnection *c, int fwd)
1028 {
1029   struct GNUNET_MESH_ConnectionKeepAlive *msg;
1030   size_t size = sizeof (struct GNUNET_MESH_ConnectionKeepAlive);
1031   char cbuf[size];
1032   uint16_t type;
1033
1034   type = fwd ? GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE :
1035                GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE;
1036
1037   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1038               "sending %s keepalive for connection %s[%d]\n",
1039               fwd ? "FWD" : "BCK",
1040               peer2s (c->t->peer),
1041               c->id);
1042
1043   msg = (struct GNUNET_MESH_ConnectionKeepAlive *) cbuf;
1044   msg->header.size = htons (size);
1045   msg->header.type = htons (type);
1046   msg->cid = c->id;
1047
1048   send_prebuilt_message_connection (&msg->header, c, NULL, fwd);
1049 }
1050
1051
1052 /**
1053  * Send CONNECTION_{CREATE/ACK} packets for a connection.
1054  *
1055  * @param c Connection for which to send the message.
1056  * @param fwd If GNUNET_YES, send CREATE, otherwise send ACK.
1057  */
1058 static void
1059 connection_recreate (struct MeshConnection *c, int fwd)
1060 {
1061   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending connection recreate\n");
1062   if (fwd)
1063     send_connection_create (c);
1064   else
1065     send_connection_ack (c, GNUNET_NO);
1066 }
1067
1068
1069 /**
1070  * Generic connection timer management.
1071  * Depending on the role of the peer in the connection will send the
1072  * appropriate message (build or keepalive)
1073  *
1074  * @param c Conncetion to maintain.
1075  * @param fwd Is FWD?
1076  */
1077 static void
1078 connection_maintain (struct MeshConnection *c, int fwd)
1079 {
1080   if (MESH_TUNNEL_SEARCHING == c->t->state)
1081   {
1082     /* TODO DHT GET with RO_BART */
1083     return;
1084   }
1085   switch (c->state)
1086   {
1087     case MESH_CONNECTION_NEW:
1088       GNUNET_break (0);
1089     case MESH_CONNECTION_SENT:
1090       connection_recreate (c, fwd);
1091       break;
1092     case MESH_CONNECTION_READY:
1093       connection_keepalive (c, fwd);
1094       break;
1095     default:
1096       break;
1097   }
1098 }
1099
1100
1101 static void
1102 connection_fwd_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1103 {
1104   struct MeshConnection *c = cls;
1105
1106   c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1107   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1108     return;
1109
1110   connection_maintain (c, GNUNET_YES);
1111   c->fwd_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
1112                                                           &connection_fwd_keepalive,
1113                                                           c);
1114 }
1115
1116
1117 static void
1118 connection_bck_keepalive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1119 {
1120   struct MeshConnection *c = cls;
1121
1122   c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1123   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1124     return;
1125
1126   connection_maintain (c, GNUNET_NO);
1127   c->bck_maintenance_task = GNUNET_SCHEDULER_add_delayed (refresh_connection_time,
1128                                                           &connection_bck_keepalive,
1129                                                           c);
1130 }
1131
1132
1133 /**
1134  * Send a message to all peers in this connection that the connection
1135  * is no longer valid.
1136  *
1137  * If some peer should not receive the message, it should be zero'ed out
1138  * before calling this function.
1139  *
1140  * @param c The connection whose peers to notify.
1141  */
1142 static void
1143 connection_send_destroy (struct MeshConnection *c)
1144 {
1145   struct GNUNET_MESH_ConnectionDestroy msg;
1146
1147   msg.header.size = htons (sizeof (msg));
1148   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY);;
1149   msg.cid = c->id;
1150   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1151               "  sending connection destroy for connection %s[%X]\n",
1152               peer2s (c->t->peer),
1153               c->id);
1154
1155   if (GNUNET_NO == GMC_is_terminal (c, GNUNET_YES))
1156     send_prebuilt_message_connection (&msg.header, c, NULL, GNUNET_YES);
1157   if (GNUNET_NO == GMC_is_terminal (c, GNUNET_NO))
1158     send_prebuilt_message_connection (&msg.header, c, NULL, GNUNET_NO);
1159   c->destroy = GNUNET_YES;
1160 }
1161
1162
1163 /**
1164  * Get free buffer space in a connection.
1165  *
1166  * @param c Connection.
1167  * @param fwd Is query about FWD traffic?
1168  *
1169  * @return Free buffer space [0 - max_msgs_queue/max_connections]
1170  */
1171 static unsigned int
1172 connection_get_buffer (struct MeshConnection *c, int fwd)
1173 {
1174   struct MeshFlowControl *fc;
1175
1176   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1177
1178   return (fc->queue_max - fc->queue_n);
1179 }
1180
1181
1182 /**
1183  * Get the first transmittable message for a connection.
1184  *
1185  * @param c Connection.
1186  * @param fwd Is this FWD?
1187  *
1188  * @return First transmittable message.
1189  */
1190 static struct MeshPeerQueue *
1191 connection_get_first_message (struct MeshConnection *c, int fwd)
1192 {
1193   struct MeshPeerQueue *q;
1194   struct MeshPeer *p;
1195
1196   p = connection_get_hop (c, fwd);
1197
1198   for (q = p->queue_head; NULL != q; q = q->next)
1199   {
1200     if (q->c != c)
1201       continue;
1202     if (queue_is_sendable (q))
1203       return q;
1204   }
1205
1206   return NULL;
1207 }
1208
1209
1210 /**
1211  * @brief Re-initiate traffic on this connection if necessary.
1212  *
1213  * Check if there is traffic queued towards this peer
1214  * and the core transmit handle is NULL (traffic was stalled).
1215  * If so, call core tmt rdy.
1216  *
1217  * @param c Connection on which initiate traffic.
1218  * @param fwd Is this about fwd traffic?
1219  */
1220 static void
1221 connection_unlock_queue (struct MeshConnection *c, int fwd)
1222 {
1223   struct MeshPeer *peer;
1224   struct MeshPeerQueue *q;
1225   size_t size;
1226
1227   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1228               "connection_unlock_queue %s on %s\n",
1229               fwd ? "FWD" : "BCK", GNUNET_h2s (&c->id));
1230
1231   if (GMC_is_terminal (c, fwd))
1232   {
1233     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " is terminal!\n");
1234     return;
1235   }
1236
1237   peer = connection_get_hop (c, fwd);
1238
1239   if (NULL != peer->core_transmit)
1240   {
1241     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  already unlocked!\n");
1242     return; /* Already unlocked */
1243   }
1244
1245   q = connection_get_first_message (c, fwd);
1246   if (NULL == q)
1247   {
1248     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  queue empty!\n");
1249     return; /* Nothing to transmit */
1250   }
1251
1252   size = q->size;
1253   peer->core_transmit =
1254       GNUNET_CORE_notify_transmit_ready (core_handle,
1255                                          GNUNET_NO,
1256                                          0,
1257                                          GNUNET_TIME_UNIT_FOREVER_REL,
1258                                          GNUNET_PEER_resolve2 (peer->id),
1259                                          size,
1260                                          &queue_send,
1261                                          peer);
1262 }
1263
1264
1265 /**
1266  * Cancel all transmissions that belong to a certain connection.
1267  *
1268  * @param c Connection which to cancel.
1269  * @param fwd Cancel fwd traffic?
1270  */
1271 static void
1272 connection_cancel_queues (struct MeshConnection *c, int fwd)
1273 {
1274   struct MeshPeerQueue *q;
1275   struct MeshPeerQueue *next;
1276   struct MeshFlowControl *fc;
1277   struct MeshPeer *peer;
1278
1279   if (NULL == c)
1280   {
1281     GNUNET_break (0);
1282     return;
1283   }
1284   fc = fwd ? &c->fwd_fc : &c->bck_fc;
1285   peer = connection_get_hop (c, fwd);
1286
1287   for (q = peer->queue_head; NULL != q; q = next)
1288   {
1289     next = q->next;
1290     if (q->c == c)
1291     {
1292       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1293                   "connection_cancel_queue %s\n",
1294                   GNUNET_MESH_DEBUG_M2S (q->type));
1295       queue_destroy (q, GNUNET_YES);
1296     }
1297   }
1298   if (NULL == peer->queue_head)
1299   {
1300     if (NULL != peer->core_transmit)
1301     {
1302       GNUNET_CORE_notify_transmit_ready_cancel (peer->core_transmit);
1303       peer->core_transmit = NULL;
1304     }
1305     if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
1306     {
1307       GNUNET_SCHEDULER_cancel (fc->poll_task);
1308       fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
1309     }
1310   }
1311 }
1312
1313
1314
1315
1316 /**
1317  * Function called if a connection has been stalled for a while,
1318  * possibly due to a missed ACK. Poll the neighbor about its ACK status.
1319  *
1320  * @param cls Closure (poll ctx).
1321  * @param tc TaskContext.
1322  */
1323 static void
1324 connection_poll (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1325 {
1326   struct MeshFlowControl *fc = cls;
1327   struct GNUNET_MESH_Poll msg;
1328   struct MeshConnection *c;
1329
1330   fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
1331   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1332   {
1333     return;
1334   }
1335
1336   c = fc->c;
1337   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** Polling!\n");
1338   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** connection [%X]\n",
1339               GNUNET_h2s (&c->id));
1340   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ***   %s\n",
1341               fc == &c->fwd_fc ? "FWD" : "BCK");
1342
1343   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_POLL);
1344   msg.header.size = htons (sizeof (msg));
1345   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " *** pid (%u)!\n", fc->last_pid_sent);
1346   send_prebuilt_message_connection (&msg.header, c, NULL, fc == &c->fwd_fc);
1347   fc->poll_time = GNUNET_TIME_STD_BACKOFF (fc->poll_time);
1348   fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
1349                                                 &connection_poll, fc);
1350 }
1351
1352
1353
1354
1355 /**
1356  * Get the previous hop in a connection
1357  *
1358  * @param c Connection.
1359  *
1360  * @return Previous peer in the connection.
1361  */
1362 static struct MeshPeer *
1363 connection_get_prev_hop (struct MeshConnection *c)
1364 {
1365   GNUNET_PEER_Id id;
1366
1367   if (0 == c->own_pos || c->path->length < 2)
1368     id = c->path->peers[0];
1369   else
1370     id = c->path->peers[c->own_pos - 1];
1371
1372   return peer_get_short (id);
1373 }
1374
1375
1376 /**
1377  * Get the next hop in a connection
1378  *
1379  * @param c Connection.
1380  *
1381  * @return Next peer in the connection.
1382  */
1383 static struct MeshPeer *
1384 connection_get_next_hop (struct MeshConnection *c)
1385 {
1386   GNUNET_PEER_Id id;
1387
1388   if ((c->path->length - 1) == c->own_pos || c->path->length < 2)
1389     id = c->path->peers[c->path->length - 1];
1390   else
1391     id = c->path->peers[c->own_pos + 1];
1392
1393   return peer_get_short (id);
1394 }
1395
1396
1397 /**
1398  * Get the hop in a connection.
1399  *
1400  * @param c Connection.
1401  * @param fwd Next hop?
1402  *
1403  * @return Next peer in the connection.
1404  */
1405 static struct MeshPeer *
1406 connection_get_hop (struct MeshConnection *c, int fwd)
1407 {
1408   if (fwd)
1409     return connection_get_next_hop (c);
1410   return connection_get_prev_hop (c);
1411 }
1412
1413
1414
1415
1416 /**
1417  * Timeout function due to lack of keepalive/traffic from the owner.
1418  * Destroys connection if called.
1419  *
1420  * @param cls Closure (connection to destroy).
1421  * @param tc TaskContext.
1422  */
1423 static void
1424 connection_fwd_timeout (void *cls,
1425                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1426 {
1427   struct MeshConnection *c = cls;
1428
1429   c->fwd_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1430   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1431     return;
1432   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1433               "Connection %s[%X] FWD timed out. Destroying.\n",
1434               peer2s (c->t->peer),
1435               c->id);
1436
1437   if (GMC_is_origin (c, GNUNET_YES)) /* If local, leave. */
1438     return;
1439
1440   GMC_destroy (c);
1441 }
1442
1443
1444 /**
1445  * Timeout function due to lack of keepalive/traffic from the destination.
1446  * Destroys connection if called.
1447  *
1448  * @param cls Closure (connection to destroy).
1449  * @param tc TaskContext
1450  */
1451 static void
1452 connection_bck_timeout (void *cls,
1453                         const struct GNUNET_SCHEDULER_TaskContext *tc)
1454 {
1455   struct MeshConnection *c = cls;
1456
1457   c->bck_maintenance_task = GNUNET_SCHEDULER_NO_TASK;
1458   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1459     return;
1460
1461   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1462               "Connection %s[%X] FWD timed out. Destroying.\n",
1463               peer2s (c->t->peer),
1464               c->id);
1465
1466   if (GMC_is_origin (c, GNUNET_NO)) /* If local, leave. */
1467     return;
1468
1469   GMC_destroy (c);
1470 }
1471
1472
1473 /**
1474  * Resets the connection timeout task, some other message has done the
1475  * task's job.
1476  * - For the first peer on the direction this means to send
1477  *   a keepalive or a path confirmation message (either create or ACK).
1478  * - For all other peers, this means to destroy the connection,
1479  *   due to lack of activity.
1480  * Starts the tiemout if no timeout was running (connection just created).
1481  *
1482  * @param c Connection whose timeout to reset.
1483  * @param fwd Is this forward?
1484  *
1485  * TODO use heap to improve efficiency of scheduler.
1486  */
1487 static void
1488 connection_reset_timeout (struct MeshConnection *c, int fwd)
1489 {
1490   GNUNET_SCHEDULER_TaskIdentifier *ti;
1491   GNUNET_SCHEDULER_Task f;
1492
1493   ti = fwd ? &c->fwd_maintenance_task : &c->bck_maintenance_task;
1494
1495   if (GNUNET_SCHEDULER_NO_TASK != *ti)
1496     GNUNET_SCHEDULER_cancel (*ti);
1497
1498   if (GMC_is_origin (c, fwd)) /* Endpoint */
1499   {
1500     f  = fwd ? &connection_fwd_keepalive : &connection_bck_keepalive;
1501     *ti = GNUNET_SCHEDULER_add_delayed (refresh_connection_time, f, c);
1502   }
1503   else /* Relay */
1504   {
1505     struct GNUNET_TIME_Relative delay;
1506
1507     delay = GNUNET_TIME_relative_multiply (refresh_connection_time, 4);
1508     f  = fwd ? &connection_fwd_timeout : &connection_bck_timeout;
1509     *ti = GNUNET_SCHEDULER_add_delayed (delay, f, c);
1510   }
1511 }
1512
1513
1514
1515
1516
1517 /**
1518  * Method called whenever a given peer connects.
1519  *
1520  * @param cls closure
1521  * @param peer peer identity this notification is about
1522  */
1523 static void
1524 core_connect (void *cls, const struct GNUNET_PeerIdentity *peer)
1525 {
1526   struct MeshPeer *pi;
1527   struct MeshPeerPath *path;
1528
1529   DEBUG_CONN ("Peer connected\n");
1530   DEBUG_CONN ("     %s\n", GNUNET_i2s (&my_full_id));
1531   pi = peer_get (peer);
1532   if (myid == pi->id)
1533   {
1534     DEBUG_CONN ("     (self)\n");
1535     path = path_new (1);
1536   }
1537   else
1538   {
1539     DEBUG_CONN ("     %s\n", GNUNET_i2s (peer));
1540     path = path_new (2);
1541     path->peers[1] = pi->id;
1542     GNUNET_PEER_change_rc (pi->id, 1);
1543     GNUNET_STATISTICS_update (stats, "# peers", 1, GNUNET_NO);
1544   }
1545   path->peers[0] = myid;
1546   GNUNET_PEER_change_rc (myid, 1);
1547   peer_add_path (pi, path, GNUNET_YES);
1548
1549   pi->connections = GNUNET_CONTAINER_multihashmap_create (32, GNUNET_YES);
1550   return;
1551 }
1552
1553
1554 /**
1555  * Method called whenever a peer disconnects.
1556  *
1557  * @param cls closure
1558  * @param peer peer identity this notification is about
1559  */
1560 static void
1561 core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
1562 {
1563   struct MeshPeer *pi;
1564
1565   DEBUG_CONN ("Peer disconnected\n");
1566   pi = GNUNET_CONTAINER_multipeermap_get (peers, peer);
1567   if (NULL == pi)
1568   {
1569     GNUNET_break (0);
1570     return;
1571   }
1572
1573   GNUNET_CONTAINER_multihashmap_iterate (pi->connections,
1574                                          GMC_notify_broken,
1575                                          pi);
1576   GNUNET_CONTAINER_multihashmap_destroy (pi->connections);
1577   pi->connections = NULL;
1578   if (NULL != pi->core_transmit)
1579     {
1580       GNUNET_CORE_notify_transmit_ready_cancel (pi->core_transmit);
1581       pi->core_transmit = NULL;
1582     }
1583   if (myid == pi->id)
1584   {
1585     DEBUG_CONN ("     (self)\n");
1586   }
1587   GNUNET_STATISTICS_update (stats, "# peers", -1, GNUNET_NO);
1588
1589   return;
1590 }
1591
1592
1593
1594 /**
1595  * To be called on core init/fail.
1596  *
1597  * @param cls Closure (config)
1598  * @param identity the public identity of this peer
1599  */
1600 static void
1601 core_init (void *cls,
1602            const struct GNUNET_PeerIdentity *identity)
1603 {
1604   const struct GNUNET_CONFIGURATION_Handle *c = cls;
1605   static int i = 0;
1606
1607   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core init\n");
1608   if (0 != memcmp (identity, &my_full_id, sizeof (my_full_id)))
1609   {
1610     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Wrong CORE service\n"));
1611     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1612                 " core id %s\n",
1613                 GNUNET_i2s (identity));
1614     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1615                 " my id %s\n",
1616                 GNUNET_i2s (&my_full_id));
1617     GNUNET_CORE_disconnect (core_handle);
1618     core_handle = GNUNET_CORE_connect (c, /* Main configuration */
1619                                        NULL,      /* Closure passed to MESH functions */
1620                                        &core_init,        /* Call core_init once connected */
1621                                        &core_connect,     /* Handle connects */
1622                                        &core_disconnect,  /* remove peers on disconnects */
1623                                        NULL,      /* Don't notify about all incoming messages */
1624                                        GNUNET_NO, /* For header only in notification */
1625                                        NULL,      /* Don't notify about all outbound messages */
1626                                        GNUNET_NO, /* For header-only out notification */
1627                                        core_handlers);    /* Register these handlers */
1628     if (10 < i++)
1629       GNUNET_abort();
1630   }
1631   GML_start ();
1632   return;
1633 }
1634
1635
1636
1637
1638
1639 /**
1640  * Core handler for connection creation.
1641  *
1642  * @param cls Closure (unused).
1643  * @param peer Sender (neighbor).
1644  * @param message Message.
1645  *
1646  * @return GNUNET_OK to keep the connection open,
1647  *         GNUNET_SYSERR to close it (signal serious error)
1648  */
1649 static int
1650 handle_create (void *cls, const struct GNUNET_PeerIdentity *peer,
1651                const struct GNUNET_MessageHeader *message)
1652 {
1653   struct GNUNET_MESH_ConnectionCreate *msg;
1654   struct GNUNET_PeerIdentity *id;
1655   struct GNUNET_HashCode *cid;
1656   struct MeshPeerPath *path;
1657   struct MeshPeer *dest_peer;
1658   struct MeshPeer *orig_peer;
1659   struct MeshConnection *c;
1660   unsigned int own_pos;
1661   uint16_t size;
1662   uint16_t i;
1663
1664   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
1665   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received a connection create msg\n");
1666
1667   /* Check size */
1668   size = ntohs (message->size);
1669   if (size < sizeof (struct GNUNET_MESH_ConnectionCreate))
1670   {
1671     GNUNET_break_op (0);
1672     return GNUNET_OK;
1673   }
1674
1675   /* Calculate hops */
1676   size -= sizeof (struct GNUNET_MESH_ConnectionCreate);
1677   if (size % sizeof (struct GNUNET_PeerIdentity))
1678   {
1679     GNUNET_break_op (0);
1680     return GNUNET_OK;
1681   }
1682   size /= sizeof (struct GNUNET_PeerIdentity);
1683   if (1 > size)
1684   {
1685     GNUNET_break_op (0);
1686     return GNUNET_OK;
1687   }
1688   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "    path has %u hops.\n", size);
1689
1690   /* Get parameters */
1691   msg = (struct GNUNET_MESH_ConnectionCreate *) message;
1692   cid = &msg->cid;
1693   id = (struct GNUNET_PeerIdentity *) &msg[1];
1694   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1695               "    connection %s (%s).\n",
1696               GNUNET_h2s (cid), GNUNET_i2s (id));
1697
1698   /* Create connection */
1699   c = connection_get (cid);
1700   if (NULL == c)
1701   {
1702     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Creating connection\n");
1703     c = connection_new (cid);
1704     if (NULL == c)
1705       return GNUNET_OK;
1706     connection_reset_timeout (c, GNUNET_YES);
1707
1708     /* Create path */
1709     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Creating path...\n");
1710     path = path_new (size);
1711     own_pos = 0;
1712     for (i = 0; i < size; i++)
1713     {
1714       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ... adding %s\n",
1715                   GNUNET_i2s (&id[i]));
1716       path->peers[i] = GNUNET_PEER_intern (&id[i]);
1717       if (path->peers[i] == myid)
1718         own_pos = i;
1719     }
1720     if (own_pos == 0 && path->peers[own_pos] != myid)
1721     {
1722       /* create path: self not found in path through self */
1723       GNUNET_break_op (0);
1724       path_destroy (path);
1725       connection_destroy (c);
1726       return GNUNET_OK;
1727     }
1728     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Own position: %u\n", own_pos);
1729     path_add_to_peers (path, GNUNET_NO);
1730     c->path = path_duplicate (path);
1731     c->own_pos = own_pos;
1732   }
1733   else
1734   {
1735     path = NULL;
1736   }
1737   if (MESH_CONNECTION_NEW == c->state)
1738     connection_change_state (c, MESH_CONNECTION_SENT);
1739
1740   /* Remember peers */
1741   dest_peer = peer_get (&id[size - 1]);
1742   orig_peer = peer_get (&id[0]);
1743
1744   /* Is it a connection to us? */
1745   if (c->own_pos == size - 1)
1746   {
1747     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  It's for us!\n");
1748     peer_add_path_to_origin (orig_peer, path, GNUNET_YES);
1749
1750     if (NULL == orig_peer->tunnel)
1751     {
1752       orig_peer->tunnel = tunnel_new ();
1753       orig_peer->tunnel->peer = orig_peer;
1754     }
1755     tunnel_add_connection (orig_peer->tunnel, c);
1756     if (MESH_TUNNEL_NEW == c->t->state)
1757       tunnel_change_state (c->t,  MESH_TUNNEL_WAITING);
1758
1759     send_connection_ack (c, GNUNET_NO);
1760     if (MESH_CONNECTION_SENT == c->state)
1761       connection_change_state (c, MESH_CONNECTION_ACK);
1762
1763     /* Keep tunnel alive in direction dest->owner*/
1764     connection_reset_timeout (c, GNUNET_NO);
1765   }
1766   else
1767   {
1768     /* It's for somebody else! Retransmit. */
1769     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Retransmitting.\n");
1770     peer_add_path (dest_peer, path_duplicate (path), GNUNET_NO);
1771     peer_add_path_to_origin (orig_peer, path, GNUNET_NO);
1772     send_prebuilt_message_connection (message, c, NULL, GNUNET_YES);
1773   }
1774   return GNUNET_OK;
1775 }
1776
1777
1778 /**
1779  * Core handler for path confirmations.
1780  *
1781  * @param cls closure
1782  * @param message message
1783  * @param peer peer identity this notification is about
1784  *
1785  * @return GNUNET_OK to keep the connection open,
1786  *         GNUNET_SYSERR to close it (signal serious error)
1787  */
1788 static int
1789 handle_confirm (void *cls, const struct GNUNET_PeerIdentity *peer,
1790                 const struct GNUNET_MessageHeader *message)
1791 {
1792   struct GNUNET_MESH_ConnectionACK *msg;
1793   struct MeshConnection *c;
1794   struct MeshPeerPath *p;
1795   struct MeshPeer *pi;
1796   int fwd;
1797
1798   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
1799   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received a connection ACK msg\n");
1800   msg = (struct GNUNET_MESH_ConnectionACK *) message;
1801   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  on connection %s\n",
1802               GNUNET_h2s (&msg->cid));
1803   c = connection_get (&msg->cid);
1804   if (NULL == c)
1805   {
1806     GNUNET_STATISTICS_update (stats, "# control on unknown connection",
1807                               1, GNUNET_NO);
1808     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  don't know the connection!\n");
1809     return GNUNET_OK;
1810   }
1811
1812
1813   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  via peer %s\n",
1814               GNUNET_i2s (peer));
1815   pi = peer_get (peer);
1816   if (connection_get_next_hop (c) == pi)
1817   {
1818     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  SYNACK\n");
1819     fwd = GNUNET_NO;
1820     if (MESH_CONNECTION_SENT == c->state)
1821       connection_change_state (c, MESH_CONNECTION_ACK);
1822   }
1823   else if (connection_get_prev_hop (c) == pi)
1824   {
1825     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ACK\n");
1826     fwd = GNUNET_YES;
1827     connection_change_state (c, MESH_CONNECTION_READY);
1828   }
1829   else
1830   {
1831     GNUNET_break_op (0);
1832     return GNUNET_OK;
1833   }
1834   connection_reset_timeout (c, fwd);
1835
1836   /* Add path to peers? */
1837   p = c->path;
1838   if (NULL != p)
1839   {
1840     path_add_to_peers (p, GNUNET_YES);
1841   }
1842   else
1843   {
1844     GNUNET_break (0);
1845   }
1846
1847   /* Message for us as creator? */
1848   if (connection_is_origin (c, GNUNET_YES))
1849   {
1850     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Connection (SYN)ACK for us!\n");
1851     connection_change_state (c, MESH_CONNECTION_READY);
1852     if (MESH_TUNNEL_READY != c->t->state)
1853       tunnel_change_state (c->t, MESH_TUNNEL_READY);
1854     send_connection_ack (c, GNUNET_YES);
1855     tunnel_send_queued_data (c->t, GNUNET_YES);
1856     if (3 <= tunnel_count_connections (c->t) && NULL != c->t->peer->dhtget)
1857     {
1858       GNUNET_DHT_get_stop (c->t->peer->dhtget);
1859       c->t->peer->dhtget = NULL;
1860     }
1861     return GNUNET_OK;
1862   }
1863
1864   /* Message for us as destination? */
1865   if (GMC_is_terminal (c, GNUNET_YES))
1866   {
1867     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Connection ACK for us!\n");
1868     if (MESH_TUNNEL_READY != c->t->state)
1869       tunnel_change_state (c->t, MESH_TUNNEL_READY);
1870     connection_change_state (c, MESH_CONNECTION_READY);
1871     tunnel_send_queued_data (c->t, GNUNET_NO);
1872     return GNUNET_OK;
1873   }
1874
1875   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
1876   send_prebuilt_message_connection (message, c, NULL, fwd);
1877   return GNUNET_OK;
1878 }
1879
1880
1881 /**
1882  * Core handler for notifications of broken paths
1883  *
1884  * @param cls Closure (unused).
1885  * @param peer Peer identity of sending neighbor.
1886  * @param message Message.
1887  *
1888  * @return GNUNET_OK to keep the connection open,
1889  *         GNUNET_SYSERR to close it (signal serious error)
1890  */
1891 static int
1892 handle_broken (void *cls, const struct GNUNET_PeerIdentity *peer,
1893                const struct GNUNET_MessageHeader *message)
1894 {
1895   struct GNUNET_MESH_ConnectionBroken *msg;
1896   struct MeshConnection *c;
1897
1898   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1899               "Received a CONNECTION BROKEN msg from %s\n", GNUNET_i2s (peer));
1900   msg = (struct GNUNET_MESH_ConnectionBroken *) message;
1901   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
1902               GNUNET_i2s (&msg->peer1));
1903   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  regarding %s\n",
1904               GNUNET_i2s (&msg->peer2));
1905   c = connection_get (&msg->cid);
1906   if (NULL == c)
1907   {
1908     GNUNET_break_op (0);
1909     return GNUNET_OK;
1910   }
1911   tunnel_notify_connection_broken (c->t, GNUNET_PEER_search (&msg->peer1),
1912                                    GNUNET_PEER_search (&msg->peer2));
1913   return GNUNET_OK;
1914
1915 }
1916
1917
1918 /**
1919  * Core handler for tunnel destruction
1920  *
1921  * @param cls Closure (unused).
1922  * @param peer Peer identity of sending neighbor.
1923  * @param message Message.
1924  *
1925  * @return GNUNET_OK to keep the connection open,
1926  *         GNUNET_SYSERR to close it (signal serious error)
1927  */
1928 static int
1929 handle_destroy (void *cls, const struct GNUNET_PeerIdentity *peer,
1930                 const struct GNUNET_MessageHeader *message)
1931 {
1932   struct GNUNET_MESH_ConnectionDestroy *msg;
1933   struct MeshConnection *c;
1934   GNUNET_PEER_Id id;
1935   int fwd;
1936
1937   msg = (struct GNUNET_MESH_ConnectionDestroy *) message;
1938   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1939               "Got a CONNECTION DESTROY message from %s\n",
1940               GNUNET_i2s (peer));
1941   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1942               "  for connection %s\n",
1943               GNUNET_h2s (&msg->cid));
1944   c = connection_get (&msg->cid);
1945   if (NULL == c)
1946   {
1947     /* Probably already got the message from another path,
1948      * destroyed the tunnel and retransmitted to children.
1949      * Safe to ignore.
1950      */
1951     GNUNET_STATISTICS_update (stats, "# control on unknown tunnel",
1952                               1, GNUNET_NO);
1953     return GNUNET_OK;
1954   }
1955   id = GNUNET_PEER_search (peer);
1956   if (id == connection_get_prev_hop (c)->id)
1957     fwd = GNUNET_YES;
1958   else if (id == connection_get_next_hop (c)->id)
1959     fwd = GNUNET_NO;
1960   else
1961   {
1962     GNUNET_break_op (0);
1963     return GNUNET_OK;
1964   }
1965   send_prebuilt_message_connection (message, c, NULL, fwd);
1966   c->destroy = GNUNET_YES;
1967
1968   return GNUNET_OK;
1969 }
1970
1971 /**
1972  * Generic handler for mesh network encrypted traffic.
1973  *
1974  * @param peer Peer identity this notification is about.
1975  * @param message Encrypted message.
1976  * @param fwd Is this FWD traffic? GNUNET_YES : GNUNET_NO;
1977  *
1978  * @return GNUNET_OK to keep the connection open,
1979  *         GNUNET_SYSERR to close it (signal serious error)
1980  */
1981 static int
1982 handle_mesh_encrypted (const struct GNUNET_PeerIdentity *peer,
1983                        const struct GNUNET_MESH_Encrypted *msg,
1984                        int fwd)
1985 {
1986   struct MeshConnection *c;
1987   struct MeshTunnel2 *t;
1988   struct MeshPeer *neighbor;
1989   struct MeshFlowControl *fc;
1990   uint32_t pid;
1991   uint32_t ttl;
1992   uint16_t type;
1993   size_t size;
1994
1995   /* Check size */
1996   size = ntohs (msg->header.size);
1997   if (size <
1998       sizeof (struct GNUNET_MESH_Encrypted) +
1999       sizeof (struct GNUNET_MessageHeader))
2000   {
2001     GNUNET_break_op (0);
2002     return GNUNET_OK;
2003   }
2004   type = ntohs (msg->header.type);
2005   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
2006   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a %s message from %s\n",
2007               GNUNET_MESH_DEBUG_M2S (type), GNUNET_i2s (peer));
2008
2009   /* Check connection */
2010   c = connection_get (&msg->cid);
2011   if (NULL == c)
2012   {
2013     GNUNET_STATISTICS_update (stats, "# unknown connection", 1, GNUNET_NO);
2014     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "WARNING connection unknown\n");
2015     return GNUNET_OK;
2016   }
2017   t = c->t;
2018   fc = fwd ? &c->bck_fc : &c->fwd_fc;
2019
2020   /* Check if origin is as expected */
2021   neighbor = connection_get_hop (c, !fwd);
2022   if (peer_get (peer)->id != neighbor->id)
2023   {
2024     GNUNET_break_op (0);
2025     return GNUNET_OK;
2026   }
2027
2028   /* Check PID */
2029   pid = ntohl (msg->pid);
2030   if (GMC_is_pid_bigger (pid, fc->last_ack_sent))
2031   {
2032     GNUNET_STATISTICS_update (stats, "# unsolicited message", 1, GNUNET_NO);
2033     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2034                 "WARNING Received PID %u, (prev %u), ACK %u\n",
2035                 pid, fc->last_pid_recv, fc->last_ack_sent);
2036     return GNUNET_OK;
2037   }
2038   if (GNUNET_NO == GMC_is_pid_bigger (pid, fc->last_pid_recv))
2039   {
2040     GNUNET_STATISTICS_update (stats, "# duplicate PID", 1, GNUNET_NO);
2041     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2042                 " Pid %u not expected (%u+), dropping!\n",
2043                 pid, fc->last_pid_recv + 1);
2044     return GNUNET_OK;
2045   }
2046   if (MESH_CONNECTION_SENT == c->state)
2047     connection_change_state (c, MESH_CONNECTION_READY);
2048   connection_reset_timeout (c, fwd);
2049   fc->last_pid_recv = pid;
2050
2051   /* Is this message for us? */
2052   if (GMC_is_terminal (c, fwd))
2053   {
2054     size_t dsize = size - sizeof (struct GNUNET_MESH_Encrypted);
2055     char cbuf[dsize];
2056     struct GNUNET_MessageHeader *msgh;
2057     unsigned int off;
2058
2059     /* TODO signature verification */
2060     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  message for us!\n");
2061     GNUNET_STATISTICS_update (stats, "# messages received", 1, GNUNET_NO);
2062
2063     fc->last_pid_recv = pid;
2064     tunnel_decrypt (t, cbuf, &msg[1], dsize, msg->iv, fwd);
2065     off = 0;
2066     while (off < dsize)
2067     {
2068       msgh = (struct GNUNET_MessageHeader *) &cbuf[off];
2069       handle_decrypted (t, msgh, fwd);
2070       off += ntohs (msgh->size);
2071     }
2072     send_ack (c, NULL, fwd);
2073     return GNUNET_OK;
2074   }
2075
2076   /* Message not for us: forward to next hop */
2077   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  not for us, retransmitting...\n");
2078   ttl = ntohl (msg->ttl);
2079   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   ttl: %u\n", ttl);
2080   if (ttl == 0)
2081   {
2082     GNUNET_STATISTICS_update (stats, "# TTL drops", 1, GNUNET_NO);
2083     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, " TTL is 0, DROPPING!\n");
2084     send_ack (c, NULL, fwd);
2085     return GNUNET_OK;
2086   }
2087   GNUNET_STATISTICS_update (stats, "# messages forwarded", 1, GNUNET_NO);
2088
2089   send_prebuilt_message_connection (&msg->header, c, NULL, fwd);
2090
2091   return GNUNET_OK;
2092 }
2093
2094
2095 /**
2096  * Core handler for mesh network traffic going orig->dest.
2097  *
2098  * @param cls Closure (unused).
2099  * @param message Message received.
2100  * @param peer Peer who sent the message.
2101  *
2102  * @return GNUNET_OK to keep the connection open,
2103  *         GNUNET_SYSERR to close it (signal serious error)
2104  */
2105 static int
2106 handle_fwd (void *cls, const struct GNUNET_PeerIdentity *peer,
2107             const struct GNUNET_MessageHeader *message)
2108 {
2109   return handle_mesh_encrypted (peer,
2110                                 (struct GNUNET_MESH_Encrypted *)message,
2111                                 GNUNET_YES);
2112 }
2113
2114 /**
2115  * Core handler for mesh network traffic going dest->orig.
2116  *
2117  * @param cls Closure (unused).
2118  * @param message Message received.
2119  * @param peer Peer who sent the message.
2120  *
2121  * @return GNUNET_OK to keep the connection open,
2122  *         GNUNET_SYSERR to close it (signal serious error)
2123  */
2124 static int
2125 handle_bck (void *cls, const struct GNUNET_PeerIdentity *peer,
2126             const struct GNUNET_MessageHeader *message)
2127 {
2128   return handle_mesh_encrypted (peer,
2129                                 (struct GNUNET_MESH_Encrypted *)message,
2130                                 GNUNET_NO);
2131 }
2132
2133
2134 /**
2135  * Core handler for mesh network traffic point-to-point acks.
2136  *
2137  * @param cls closure
2138  * @param message message
2139  * @param peer peer identity this notification is about
2140  *
2141  * @return GNUNET_OK to keep the connection open,
2142  *         GNUNET_SYSERR to close it (signal serious error)
2143  */
2144 static int
2145 handle_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
2146             const struct GNUNET_MessageHeader *message)
2147 {
2148   struct GNUNET_MESH_ACK *msg;
2149   struct MeshConnection *c;
2150   struct MeshFlowControl *fc;
2151   GNUNET_PEER_Id id;
2152   uint32_t ack;
2153   int fwd;
2154
2155   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
2156   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK packet from %s!\n",
2157               GNUNET_i2s (peer));
2158   msg = (struct GNUNET_MESH_ACK *) message;
2159
2160   c = connection_get (&msg->cid);
2161
2162   if (NULL == c)
2163   {
2164     GNUNET_STATISTICS_update (stats, "# ack on unknown connection", 1,
2165                               GNUNET_NO);
2166     return GNUNET_OK;
2167   }
2168
2169   /* Is this a forward or backward ACK? */
2170   id = GNUNET_PEER_search (peer);
2171   if (connection_get_next_hop (c)->id == id)
2172   {
2173     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  FWD ACK\n");
2174     fc = &c->fwd_fc;
2175     fwd = GNUNET_YES;
2176   }
2177   else if (connection_get_prev_hop (c)->id == id)
2178   {
2179     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  BCK ACK\n");
2180     fc = &c->bck_fc;
2181     fwd = GNUNET_NO;
2182   }
2183   else
2184   {
2185     GNUNET_break_op (0);
2186     return GNUNET_OK;
2187   }
2188
2189   ack = ntohl (msg->ack);
2190   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ACK %u (was %u)\n",
2191               ack, fc->last_ack_recv);
2192   if (GMC_is_pid_bigger (ack, fc->last_ack_recv))
2193     fc->last_ack_recv = ack;
2194
2195   /* Cancel polling if the ACK is big enough. */
2196   if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task &&
2197       GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
2198   {
2199     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Cancel poll\n");
2200     GNUNET_SCHEDULER_cancel (fc->poll_task);
2201     fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
2202     fc->poll_time = GNUNET_TIME_UNIT_SECONDS;
2203   }
2204
2205   connection_unlock_queue (c, fwd);
2206
2207   return GNUNET_OK;
2208 }
2209
2210
2211 /**
2212  * Core handler for mesh network traffic point-to-point ack polls.
2213  *
2214  * @param cls closure
2215  * @param message message
2216  * @param peer peer identity this notification is about
2217  *
2218  * @return GNUNET_OK to keep the connection open,
2219  *         GNUNET_SYSERR to close it (signal serious error)
2220  */
2221 static int
2222 handle_poll (void *cls, const struct GNUNET_PeerIdentity *peer,
2223              const struct GNUNET_MessageHeader *message)
2224 {
2225   struct GNUNET_MESH_Poll *msg;
2226   struct MeshConnection *c;
2227   struct MeshFlowControl *fc;
2228   GNUNET_PEER_Id id;
2229   uint32_t pid;
2230   int fwd;
2231
2232   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\n\n");
2233   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got a POLL packet from %s!\n",
2234               GNUNET_i2s (peer));
2235
2236   msg = (struct GNUNET_MESH_Poll *) message;
2237
2238   c = connection_get (&msg->cid);
2239
2240   if (NULL == c)
2241   {
2242     GNUNET_STATISTICS_update (stats, "# poll on unknown connection", 1,
2243                               GNUNET_NO);
2244     GNUNET_break_op (0);
2245     return GNUNET_OK;
2246   }
2247
2248   /* Is this a forward or backward ACK?
2249    * Note: a poll should never be needed in a loopback case,
2250    * since there is no possiblility of packet loss there, so
2251    * this way of discerining FWD/BCK should not be a problem.
2252    */
2253   id = GNUNET_PEER_search (peer);
2254   if (connection_get_next_hop (c)->id == id)
2255   {
2256     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  FWD ACK\n");
2257     fc = &c->fwd_fc;
2258   }
2259   else if (connection_get_prev_hop (c)->id == id)
2260   {
2261     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  BCK ACK\n");
2262     fc = &c->bck_fc;
2263   }
2264   else
2265   {
2266     GNUNET_break_op (0);
2267     return GNUNET_OK;
2268   }
2269
2270   pid = ntohl (msg->pid);
2271   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  PID %u, OLD %u\n",
2272               pid, fc->last_pid_recv);
2273   fc->last_pid_recv = pid;
2274   fwd = fc == &c->fwd_fc;
2275   send_ack (c, NULL, fwd);
2276
2277   return GNUNET_OK;
2278 }
2279
2280
2281 /**
2282  * Core handler for mesh keepalives.
2283  *
2284  * @param cls closure
2285  * @param message message
2286  * @param peer peer identity this notification is about
2287  * @return GNUNET_OK to keep the connection open,
2288  *         GNUNET_SYSERR to close it (signal serious error)
2289  *
2290  * TODO: Check who we got this from, to validate route.
2291  */
2292 static int
2293 handle_keepalive (void *cls, const struct GNUNET_PeerIdentity *peer,
2294                 const struct GNUNET_MessageHeader *message)
2295 {
2296   struct GNUNET_MESH_ConnectionKeepAlive *msg;
2297   struct MeshConnection *c;
2298   struct MeshPeer *neighbor;
2299   int fwd;
2300
2301   msg = (struct GNUNET_MESH_ConnectionKeepAlive *) message;
2302   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got a keepalive packet from %s\n",
2303               GNUNET_i2s (peer));
2304
2305   c = connection_get (&msg->cid);
2306   if (NULL == c)
2307   {
2308     GNUNET_STATISTICS_update (stats, "# keepalive on unknown connection", 1,
2309                               GNUNET_NO);
2310     return GNUNET_OK;
2311   }
2312
2313   fwd = GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE == ntohs (message->type) ?
2314         GNUNET_YES : GNUNET_NO;
2315
2316   /* Check if origin is as expected */
2317   neighbor = connection_get_hop (c, fwd);
2318   if (peer_get (peer)->id != neighbor->id)
2319   {
2320     GNUNET_break_op (0);
2321     return GNUNET_OK;
2322   }
2323
2324   connection_change_state (c, MESH_CONNECTION_READY);
2325   connection_reset_timeout (c, fwd);
2326
2327   if (GMC_is_terminal (c, fwd))
2328     return GNUNET_OK;
2329
2330   GNUNET_STATISTICS_update (stats, "# keepalives forwarded", 1, GNUNET_NO);
2331   send_prebuilt_message_connection (message, c, NULL, fwd);
2332
2333   return GNUNET_OK;
2334 }
2335
2336
2337 /**
2338  * Functions to handle messages from core
2339  */
2340 static struct GNUNET_CORE_MessageHandler core_handlers[] = {
2341   {&handle_create, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE,
2342     0},
2343   {&handle_confirm, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
2344     sizeof (struct GNUNET_MESH_ConnectionACK)},
2345   {&handle_broken, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN,
2346     sizeof (struct GNUNET_MESH_ConnectionBroken)},
2347   {&handle_destroy, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY,
2348     sizeof (struct GNUNET_MESH_ConnectionDestroy)},
2349   {&handle_keepalive, GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE,
2350     sizeof (struct GNUNET_MESH_ConnectionKeepAlive)},
2351   {&handle_keepalive, GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE,
2352     sizeof (struct GNUNET_MESH_ConnectionKeepAlive)},
2353   {&handle_ack, GNUNET_MESSAGE_TYPE_MESH_ACK,
2354     sizeof (struct GNUNET_MESH_ACK)},
2355   {&handle_poll, GNUNET_MESSAGE_TYPE_MESH_POLL,
2356     sizeof (struct GNUNET_MESH_Poll)},
2357   {&handle_fwd, GNUNET_MESSAGE_TYPE_MESH_FWD, 0},
2358   {&handle_bck, GNUNET_MESSAGE_TYPE_MESH_BCK, 0},
2359   {NULL, 0, 0}
2360 };
2361
2362
2363
2364 /**
2365  * Iterator to notify all connections of a broken link. Mark connections
2366  * to destroy after all traffic has been sent.
2367  *
2368  * @param cls Closure (peer disconnected).
2369  * @param key Current key code (tid).
2370  * @param value Value in the hash map (connection).
2371  *
2372  * @return GNUNET_YES if we should continue to iterate,
2373  *         GNUNET_NO if not.
2374  */
2375 int
2376 GMC_notify_broken (void *cls,
2377                    const struct GNUNET_HashCode *key,
2378                    void *value)
2379 {
2380   struct MeshPeer *peer = cls;
2381   struct MeshConnection *c = value;
2382   struct GNUNET_MESH_ConnectionBroken msg;
2383   int fwd;
2384
2385   fwd = peer == connection_get_prev_hop (c);
2386
2387   connection_cancel_queues (c, !fwd);
2388   if (GMC_is_terminal (c, fwd))
2389   {
2390     /* Local shutdown, no one to notify about this. */
2391     GMC_destroy (c);
2392     return GNUNET_YES;
2393   }
2394
2395   msg.header.size = htons (sizeof (struct GNUNET_MESH_ConnectionBroken));
2396   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN);
2397   msg.cid = c->id;
2398   msg.peer1 = my_full_id;
2399   msg.peer2 = *GNUNET_PEER_resolve2 (peer->id);
2400   send_prebuilt_message_connection (&msg.header, c, NULL, fwd);
2401   c->destroy = GNUNET_YES;
2402
2403   return GNUNET_YES;
2404 }
2405
2406
2407 /**
2408  * Initialize the connections subsystem
2409  *
2410  * @param c Configuration handle.
2411  */
2412 void
2413 GMC_init (const struct GNUNET_CONFIGURATION_Handle *c)
2414 {
2415   if (GNUNET_OK !=
2416       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_MSGS_QUEUE",
2417                                              &max_msgs_queue))
2418   {
2419     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2420                                "MESH", "MAX_MSGS_QUEUE", "MISSING");
2421     GNUNET_SCHEDULER_shutdown ();
2422     return;
2423   }
2424
2425   if (GNUNET_OK !=
2426       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_CONNECTIONS",
2427                                              &max_connections))
2428   {
2429     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2430                                "MESH", "MAX_CONNECTIONS", "MISSING");
2431     GNUNET_SCHEDULER_shutdown ();
2432     return;
2433   }
2434
2435   if (GNUNET_OK !=
2436       GNUNET_CONFIGURATION_get_value_time (c, "MESH", "REFRESH_CONNECTION_TIME",
2437                                            &refresh_connection_time))
2438   {
2439     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
2440                                "MESH", "REFRESH_CONNECTION_TIME", "MISSING");
2441     GNUNET_SCHEDULER_shutdown ();
2442     return;
2443   }
2444   connections = GNUNET_CONTAINER_multihashmap_create (1024, GNUNET_YES);
2445
2446   core_handle = GNUNET_CORE_connect (c, /* Main configuration */
2447                                      NULL,      /* Closure passed to MESH functions */
2448                                      &core_init,        /* Call core_init once connected */
2449                                      &core_connect,     /* Handle connects */
2450                                      &core_disconnect,  /* remove peers on disconnects */
2451                                      NULL,      /* Don't notify about all incoming messages */
2452                                      GNUNET_NO, /* For header only in notification */
2453                                      NULL,      /* Don't notify about all outbound messages */
2454                                      GNUNET_NO, /* For header-only out notification */
2455                                      core_handlers);    /* Register these handlers */
2456   if (NULL == core_handle)
2457   {
2458     GNUNET_break (0);
2459     GNUNET_SCHEDULER_shutdown ();
2460     return;
2461   }
2462 }
2463
2464 /**
2465  * Shut down the connections subsystem.
2466  */
2467 void
2468 GMC_shutdown (void)
2469 {
2470   if (core_handle != NULL)
2471   {
2472     GNUNET_CORE_disconnect (core_handle);
2473     core_handle = NULL;
2474   }
2475 }
2476
2477
2478 /**
2479  * Is this peer the first one on the connection?
2480  *
2481  * @param c Connection.
2482  * @param fwd Is this about fwd traffic?
2483  *
2484  * @return GNUNET_YES if origin, GNUNET_NO if relay/terminal.
2485  */
2486 int
2487 GMC_is_origin (struct MeshConnection *c, int fwd)
2488 {
2489   if (!fwd && c->path->length - 1 == c->own_pos )
2490     return GNUNET_YES;
2491   if (fwd && 0 == c->own_pos)
2492     return GNUNET_YES;
2493   return GNUNET_NO;
2494 }
2495
2496
2497 /**
2498  * Is this peer the last one on the connection?
2499  *
2500  * @param c Connection.
2501  * @param fwd Is this about fwd traffic?
2502  *            Note that the ROOT is the terminal for BCK traffic!
2503  *
2504  * @return GNUNET_YES if terminal, GNUNET_NO if relay/origin.
2505  */
2506 int
2507 GMC_is_terminal (struct MeshConnection *c, int fwd)
2508 {
2509   return GMC_is_origin (c, !fwd);
2510 }