src/mesh/gnunet-service-mesh_peer.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 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 #include "platform.h"
  23 #include "gnunet_util_lib.h"
  24
  25 #include "gnunet_core_service.h"
  26 #include "gnunet_statistics_service.h"
  27
  28 #include "mesh_protocol_enc.h"
  29
  30 #include "gnunet-service-mesh_peer.h"
  31 #include "gnunet-service-mesh_dht.h"
  32 #include "gnunet-service-mesh_connection.h"
  33 #include "gnunet-service-mesh_local.h"
  34 #include "gnunet-service-mesh_tunnel.h"
  35 #include "mesh_path.h"
  36
  37 #define LOG(level, ...) GNUNET_log_from (level,"mesh-p2p",__VA_ARGS__)
  38
  39 /******************************************************************************/
  40 /********************************   STRUCTS  **********************************/
  41 /******************************************************************************/
  42
  43 /**
  44  * Struct containing info about a queued transmission to this peer
  45  */
  46 struct MeshPeerQueue
  47 {
  48     /**
  49       * DLL next
  50       */
  51   struct MeshPeerQueue *next;
  52
  53     /**
  54       * DLL previous
  55       */
  56   struct MeshPeerQueue *prev;
  57
  58     /**
  59      * Peer this transmission is directed to.
  60      */
  61   struct MeshPeer *peer;
  62
  63     /**
  64      * Connection this message belongs to.
  65      */
  66   struct MeshConnection *c;
  67
  68     /**
  69      * Is FWD in c?
  70      */
  71   int fwd;
  72
  73     /**
  74      * Channel this message belongs to, if known.
  75      */
  76   struct MeshChannel *ch;
  77
  78     /**
  79      * Pointer to info stucture used as cls.
  80      */
  81   void *cls;
  82
  83     /**
  84      * Type of message
  85      */
  86   uint16_t type;
  87
  88     /**
  89      * Size of the message
  90      */
  91   size_t size;
  92
  93     /**
  94      * Set when this message starts waiting for CORE.
  95      */
  96   struct GNUNET_TIME_Absolute start_waiting;
  97
  98     /**
  99      * Function to call on sending.
 100      */
 101   GMP_sent callback;
 102
 103     /**
 104      * Closure for callback.
 105      */
 106   void *callback_cls;
 107 };
 108
 109 /**
 110  * Struct containing all information regarding a given peer
 111  */
 112 struct MeshPeer
 113 {
 114     /**
 115      * ID of the peer
 116      */
 117   GNUNET_PEER_Id id;
 118
 119     /**
 120      * Last time we heard from this peer
 121      */
 122   struct GNUNET_TIME_Absolute last_contact;
 123
 124     /**
 125      * Paths to reach the peer, ordered by ascending hop count
 126      */
 127   struct MeshPeerPath *path_head;
 128
 129     /**
 130      * Paths to reach the peer, ordered by ascending hop count
 131      */
 132   struct MeshPeerPath *path_tail;
 133
 134     /**
 135      * Handle to stop the DHT search for paths to this peer
 136      */
 137   struct GMD_search_handle *search_h;
 138
 139     /**
 140      * Tunnel to this peer, if any.
 141      */
 142   struct MeshTunnel2 *tunnel;
 143
 144     /**
 145      * Connections that go through this peer, indexed by tid;
 146      */
 147   struct GNUNET_CONTAINER_MultiHashMap *connections;
 148
 149     /**
 150      * Handle for queued transmissions
 151      */
 152   struct GNUNET_CORE_TransmitHandle *core_transmit;
 153
 154   /**
 155    * Transmission queue to core DLL head
 156    */
 157   struct MeshPeerQueue *queue_head;
 158
 159   /**
 160    * Transmission queue to core DLL tail
 161    */
 162   struct MeshPeerQueue *queue_tail;
 163
 164   /**
 165    * How many messages are in the queue to this peer.
 166    */
 167   unsigned int queue_n;
 168 };
 169
 170
 171 /******************************************************************************/
 172 /*******************************   GLOBALS  ***********************************/
 173 /******************************************************************************/
 174
 175 /**
 176  * Global handle to the statistics service.
 177  */
 178 extern struct GNUNET_STATISTICS_Handle *stats;
 179
 180 /**
 181  * Peers known, indexed by PeerIdentity (MeshPeer).
 182  */
 183 static struct GNUNET_CONTAINER_MultiPeerMap *peers;
 184
 185 /**
 186  * How many peers do we want to remember?
 187  */
 188 static unsigned long long max_peers;
 189
 190 /**
 191  * Percentage of messages that will be dropped (for test purposes only).
 192  */
 193 static unsigned long long drop_percent;
 194
 195 /**
 196  * Handle to communicate with core.
 197  */
 198 static struct GNUNET_CORE_Handle *core_handle;
 199
 200 /**
 201  * Local peer own ID (full value).
 202  */
 203 const static struct GNUNET_PeerIdentity *my_full_id;
 204
 205 /******************************************************************************/
 206 /***************************** CORE CALLBACKS *********************************/
 207 /******************************************************************************/
 208
 209
 210 /**
 211  * Iterator to notify all connections of a broken link. Mark connections
 212  * to destroy after all traffic has been sent.
 213  *
 214  * @param cls Closure (peer disconnected).
 215  * @param key Current key code (peer id).
 216  * @param value Value in the hash map (connection).
 217  *
 218  * @return GNUNET_YES if we should continue to iterate,
 219  *         GNUNET_NO if not.
 220  */
 221 static int
 222 notify_broken (void *cls,
 223                const struct GNUNET_HashCode *key,
 224                void *value)
 225 {
 226   struct MeshPeer *peer = cls;
 227   struct MeshConnection *c = value;
 228
 229   GMC_notify_broken (c, peer, my_full_id);
 230
 231   return GNUNET_YES;
 232 }
 233
 234
 235 /**
 236  * Method called whenever a given peer connects.
 237  *
 238  * @param cls closure
 239  * @param peer peer identity this notification is about
 240  */
 241 static void
 242 core_connect (void *cls, const struct GNUNET_PeerIdentity *peer)
 243 {
 244   struct MeshPeer *pi;
 245   struct MeshPeerPath *path;
 246
 247   LOG ("Peer connected\n");
 248   LOG ("     %s\n", GNUNET_i2s (&my_full_id));
 249   pi = peer_get (peer);
 250   if (myid == pi->id)
 251   {
 252     LOG ("     (self)\n");
 253     path = path_new (1);
 254   }
 255   else
 256   {
 257     LOG ("     %s\n", GNUNET_i2s (peer));
 258     path = path_new (2);
 259     path->peers[1] = pi->id;
 260     GNUNET_PEER_change_rc (pi->id, 1);
 261     GNUNET_STATISTICS_update (stats, "# peers", 1, GNUNET_NO);
 262   }
 263   path->peers[0] = myid;
 264   GNUNET_PEER_change_rc (myid, 1);
 265   peer_add_path (pi, path, GNUNET_YES);
 266
 267   pi->connections = GNUNET_CONTAINER_multihashmap_create (32, GNUNET_YES);
 268   return;
 269 }
 270
 271
 272 /**
 273  * Method called whenever a peer disconnects.
 274  *
 275  * @param cls closure
 276  * @param peer peer identity this notification is about
 277  */
 278 static void
 279 core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
 280 {
 281   struct MeshPeer *pi;
 282
 283   LOG ("Peer disconnected\n");
 284   pi = GNUNET_CONTAINER_multipeermap_get (peers, peer);
 285   if (NULL == pi)
 286   {
 287     GNUNET_break (0);
 288     return;
 289   }
 290
 291   GNUNET_CONTAINER_multihashmap_iterate (pi->connections, &notify_broken, pi);
 292   GNUNET_CONTAINER_multihashmap_destroy (pi->connections);
 293   pi->connections = NULL;
 294   if (NULL != pi->core_transmit)
 295     {
 296       GNUNET_CORE_notify_transmit_ready_cancel (pi->core_transmit);
 297       pi->core_transmit = NULL;
 298     }
 299   if (myid == pi->id)
 300   {
 301     LOG ("     (self)\n");
 302   }
 303   GNUNET_STATISTICS_update (stats, "# peers", -1, GNUNET_NO);
 304
 305   return;
 306 }
 307
 308
 309 /**
 310  * Functions to handle messages from core
 311  */
 312 static struct GNUNET_CORE_MessageHandler core_handlers[] = {
 313   {&GMC_handle_create, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE,
 314     0},
 315   {&GMC_handle_confirm, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK,
 316     sizeof (struct GNUNET_MESH_ConnectionACK)},
 317   {&GMC_handle_broken, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN,
 318     sizeof (struct GNUNET_MESH_ConnectionBroken)},
 319   {&GMC_handle_destroy, GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY,
 320     sizeof (struct GNUNET_MESH_ConnectionDestroy)},
 321   {&GMC_handle_keepalive, GNUNET_MESSAGE_TYPE_MESH_FWD_KEEPALIVE,
 322     sizeof (struct GNUNET_MESH_ConnectionKeepAlive)},
 323   {&GMC_handle_keepalive, GNUNET_MESSAGE_TYPE_MESH_BCK_KEEPALIVE,
 324     sizeof (struct GNUNET_MESH_ConnectionKeepAlive)},
 325   {&GMC_handle_ack, GNUNET_MESSAGE_TYPE_MESH_ACK,
 326     sizeof (struct GNUNET_MESH_ACK)},
 327   {&GMC_handle_poll, GNUNET_MESSAGE_TYPE_MESH_POLL,
 328     sizeof (struct GNUNET_MESH_Poll)},
 329   {&GMC_handle_fwd, GNUNET_MESSAGE_TYPE_MESH_FWD, 0},
 330   {&GMC_handle_bck, GNUNET_MESSAGE_TYPE_MESH_BCK, 0},
 331   {NULL, 0, 0}
 332 };
 333
 334
 335 /**
 336  * To be called on core init/fail.
 337  *
 338  * @param cls Closure (config)
 339  * @param identity the public identity of this peer
 340  */
 341 static void
 342 core_init (void *cls,
 343            const struct GNUNET_PeerIdentity *identity)
 344 {
 345   const struct GNUNET_CONFIGURATION_Handle *c = cls;
 346   static int i = 0;
 347
 348   LOG (GNUNET_ERROR_TYPE_DEBUG, "Core init\n");
 349   if (0 != memcmp (identity, &my_full_id, sizeof (my_full_id)))
 350   {
 351     LOG (GNUNET_ERROR_TYPE_ERROR, _("Wrong CORE service\n"));
 352     LOG (GNUNET_ERROR_TYPE_ERROR, " core id %s\n", GNUNET_i2s (identity));
 353     LOG (GNUNET_ERROR_TYPE_ERROR, " my id %s\n", GNUNET_i2s (&my_full_id));
 354     GNUNET_CORE_disconnect (core_handle);
 355     core_handle = GNUNET_CORE_connect (c, /* Main configuration */
 356                                        NULL,      /* Closure passed to MESH functions */
 357                                        &core_init,        /* Call core_init once connected */
 358                                        &core_connect,     /* Handle connects */
 359                                        &core_disconnect,  /* remove peers on disconnects */
 360                                        NULL,      /* Don't notify about all incoming messages */
 361                                        GNUNET_NO, /* For header only in notification */
 362                                        NULL,      /* Don't notify about all outbound messages */
 363                                        GNUNET_NO, /* For header-only out notification */
 364                                        core_handlers);    /* Register these handlers */
 365     if (10 < i++)
 366       GNUNET_abort();
 367   }
 368   GML_start ();
 369   return;
 370 }
 371
 372 /**
 373   * Core callback to write a pre-constructed data packet to core buffer
 374   *
 375   * @param cls Closure (MeshTransmissionDescriptor with data in "data" member).
 376   * @param size Number of bytes available in buf.
 377   * @param buf Where the to write the message.
 378   *
 379   * @return number of bytes written to buf
 380   */
 381 static size_t
 382 send_core_data_raw (void *cls, size_t size, void *buf)
 383 {
 384   struct GNUNET_MessageHeader *msg = cls;
 385   size_t total_size;
 386
 387   GNUNET_assert (NULL != msg);
 388   total_size = ntohs (msg->size);
 389
 390   if (total_size > size)
 391   {
 392     GNUNET_break (0);
 393     return 0;
 394   }
 395   memcpy (buf, msg, total_size);
 396   GNUNET_free (cls);
 397   return total_size;
 398 }
 399
 400
 401 /**
 402  * Function to send a create connection message to a peer.
 403  *
 404  * @param c Connection to create.
 405  * @param size number of bytes available in buf
 406  * @param buf where the callee should write the message
 407  * @return number of bytes written to buf
 408  */
 409 static size_t
 410 send_core_connection_create (struct MeshConnection *c, size_t size, void *buf)
 411 {
 412   struct GNUNET_MESH_ConnectionCreate *msg;
 413   struct GNUNET_PeerIdentity *peer_ptr;
 414   struct MeshPeerPath *p = c->path;
 415   size_t size_needed;
 416   int i;
 417
 418   LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending CONNECTION CREATE...\n");
 419   size_needed =
 420       sizeof (struct GNUNET_MESH_ConnectionCreate) +
 421       p->length * sizeof (struct GNUNET_PeerIdentity);
 422
 423   if (size < size_needed || NULL == buf)
 424   {
 425     GNUNET_break (0);
 426     return 0;
 427   }
 428   msg = (struct GNUNET_MESH_ConnectionCreate *) buf;
 429   msg->header.size = htons (size_needed);
 430   msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE);
 431   msg->cid = c->id;
 432
 433   peer_ptr = (struct GNUNET_PeerIdentity *) &msg[1];
 434   for (i = 0; i < p->length; i++)
 435   {
 436     GNUNET_PEER_resolve (p->peers[i], peer_ptr++);
 437   }
 438
 439   LOG (GNUNET_ERROR_TYPE_DEBUG,
 440               "CONNECTION CREATE (%u bytes long) sent!\n", size_needed);
 441   return size_needed;
 442 }
 443
 444
 445 /**
 446  * Creates a path ack message in buf and frees all unused resources.
 447  *
 448  * @param c Connection to send an ACK on.
 449  * @param size number of bytes available in buf
 450  * @param buf where the callee should write the message
 451  *
 452  * @return number of bytes written to buf
 453  */
 454 static size_t
 455 send_core_connection_ack (struct MeshConnection *c, size_t size, void *buf)
 456 {
 457   struct GNUNET_MESH_ConnectionACK *msg = buf;
 458   struct MeshTunnel2 *t = c->t;
 459
 460   LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending CONNECTION ACK...\n");
 461   GNUNET_assert (NULL != t);
 462   if (sizeof (struct GNUNET_MESH_ConnectionACK) > size)
 463   {
 464     GNUNET_break (0);
 465     return 0;
 466   }
 467   msg->header.size = htons (sizeof (struct GNUNET_MESH_ConnectionACK));
 468   msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK);
 469   msg->cid = c->id;
 470   msg->reserved = 0;
 471
 472   /* TODO add signature */
 473
 474   LOG (GNUNET_ERROR_TYPE_DEBUG, "CONNECTION ACK sent!\n");
 475   return sizeof (struct GNUNET_MESH_ConnectionACK);
 476 }
 477
 478
 479 /******************************************************************************/
 480 /********************************   STATIC  ***********************************/
 481 /******************************************************************************/
 482
 483 /**
 484  * Iterator over tunnel hash map entries to destroy the tunnel during shutdown.
 485  *
 486  * @param cls closure
 487  * @param key current key code
 488  * @param value value in the hash map
 489  * @return #GNUNET_YES if we should continue to iterate,
 490  *         #GNUNET_NO if not.
 491  */
 492 static int
 493 shutdown_tunnel (void *cls,
 494                  const struct GNUNET_PeerIdentity *key,
 495                  void *value)
 496 {
 497   struct MeshPeer *p = value;
 498   struct MeshTunnel2 *t = p->tunnel;
 499
 500   if (NULL != t)
 501     GMT_destroy (t);
 502   return GNUNET_YES;
 503 }
 504
 505
 506
 507 /**
 508  * Destroy the peer_info and free any allocated resources linked to it
 509  *
 510  * @param peer The peer_info to destroy.
 511  *
 512  * @return GNUNET_OK on success
 513  */
 514 static int
 515 peer_destroy (struct MeshPeer *peer)
 516 {
 517   struct GNUNET_PeerIdentity id;
 518   struct MeshPeerPath *p;
 519   struct MeshPeerPath *nextp;
 520
 521   GNUNET_PEER_resolve (peer->id, &id);
 522   GNUNET_PEER_change_rc (peer->id, -1);
 523
 524   if (GNUNET_YES !=
 525     GNUNET_CONTAINER_multipeermap_remove (peers, &id, peer))
 526   {
 527     GNUNET_break (0);
 528     LOG (GNUNET_ERROR_TYPE_WARNING,
 529                 "removing peer %s, not in peermap\n", GNUNET_i2s (&id));
 530   }
 531     if (NULL != peer->search_h)
 532     {
 533       GMD_search_stop (peer->search_h);
 534     }
 535       p = peer->path_head;
 536       while (NULL != p)
 537       {
 538         nextp = p->next;
 539         GNUNET_CONTAINER_DLL_remove (peer->path_head, peer->path_tail, p);
 540         path_destroy (p);
 541         p = nextp;
 542       }
 543         tunnel_destroy_empty (peer->tunnel);
 544         GNUNET_free (peer);
 545         return GNUNET_OK;
 546 }
 547
 548
 549 /**
 550  * Returns if peer is used (has a tunnel, is neighbor).
 551  *
 552  * @peer Peer to check.
 553  *
 554  * @return GNUNET_YES if peer is in use.
 555  */
 556 static int
 557 peer_is_used (struct MeshPeer *peer)
 558 {
 559   struct MeshPeerPath *p;
 560
 561   if (NULL != peer->tunnel)
 562     return GNUNET_YES;
 563
 564   for (p = peer->path_head; NULL != p; p = p->next)
 565   {
 566     if (p->length < 3)
 567       return GNUNET_YES;
 568   }
 569     return GNUNET_NO;
 570 }
 571
 572
 573 /**
 574  * Iterator over all the peers to get the oldest timestamp.
 575  *
 576  * @param cls Closure (unsued).
 577  * @param key ID of the peer.
 578  * @param value Peer_Info of the peer.
 579  */
 580 static int
 581 peer_get_oldest (void *cls,
 582                  const struct GNUNET_PeerIdentity *key,
 583                  void *value)
 584 {
 585   struct MeshPeer *p = value;
 586   struct GNUNET_TIME_Absolute *abs = cls;
 587
 588   /* Don't count active peers */
 589   if (GNUNET_YES == peer_is_used (p))
 590     return GNUNET_YES;
 591
 592   if (abs->abs_value_us < p->last_contact.abs_value_us)
 593     abs->abs_value_us = p->last_contact.abs_value_us;
 594
 595   return GNUNET_YES;
 596 }
 597
 598
 599 /**
 600  * Iterator over all the peers to remove the oldest entry.
 601  *
 602  * @param cls Closure (unsued).
 603  * @param key ID of the peer.
 604  * @param value Peer_Info of the peer.
 605  */
 606 static int
 607 peer_timeout (void *cls,
 608               const struct GNUNET_PeerIdentity *key,
 609               void *value)
 610 {
 611   struct MeshPeer *p = value;
 612   struct GNUNET_TIME_Absolute *abs = cls;
 613
 614   if (p->last_contact.abs_value_us == abs->abs_value_us &&
 615     GNUNET_NO == peer_is_used (p))
 616   {
 617     peer_destroy (p);
 618     return GNUNET_NO;
 619   }
 620     return GNUNET_YES;
 621 }
 622
 623
 624 /**
 625  * Delete oldest unused peer.
 626  */
 627 static void
 628 peer_delete_oldest (void)
 629 {
 630   struct GNUNET_TIME_Absolute abs;
 631
 632   abs = GNUNET_TIME_UNIT_FOREVER_ABS;
 633
 634   GNUNET_CONTAINER_multipeermap_iterate (peers,
 635                                          &peer_get_oldest,
 636                                          &abs);
 637   GNUNET_CONTAINER_multipeermap_iterate (peers,
 638                                          &peer_timeout,
 639                                          &abs);
 640 }
 641
 642
 643 /**
 644  * Retrieve the MeshPeer stucture associated with the peer, create one
 645  * and insert it in the appropriate structures if the peer is not known yet.
 646  *
 647  * @param peer Full identity of the peer.
 648  *
 649  * @return Existing or newly created peer info.
 650  */
 651 static struct MeshPeer *
 652 peer_get (const struct GNUNET_PeerIdentity *peer_id)
 653 {
 654   struct MeshPeer *peer;
 655
 656   peer = GNUNET_CONTAINER_multipeermap_get (peers, peer_id);
 657   if (NULL == peer)
 658   {
 659     peer = GNUNET_new (struct MeshPeer);
 660     if (GNUNET_CONTAINER_multipeermap_size (peers) > max_peers)
 661     {
 662       peer_delete_oldest ();
 663     }
 664         GNUNET_CONTAINER_multipeermap_put (peers, peer_id, peer,
 665                                            GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 666         peer->id = GNUNET_PEER_intern (peer_id);
 667   }
 668     peer->last_contact = GNUNET_TIME_absolute_get();
 669
 670     return peer;
 671 }
 672
 673
 674 /**
 675  * Retrieve the MeshPeer stucture associated with the peer, create one
 676  * and insert it in the appropriate structures if the peer is not known yet.
 677  *
 678  * @param peer Short identity of the peer.
 679  *
 680  * @return Existing or newly created peer info.
 681  */
 682 static struct MeshPeer *
 683 peer_get_short (const GNUNET_PEER_Id peer)
 684 {
 685   return peer_get (GNUNET_PEER_resolve2 (peer));
 686 }
 687
 688
 689 /**
 690  * Get a cost of a path for a peer considering existing tunnel connections.
 691  *
 692  * @param peer Peer towards which the path is considered.
 693  * @param path Candidate path.
 694  *
 695  * @return Cost of the path (path length + number of overlapping nodes)
 696  */
 697 static unsigned int
 698 peer_get_path_cost (const struct MeshPeer *peer,
 699                     const struct MeshPeerPath *path)
 700 {
 701   struct MeshConnection *c;
 702   unsigned int overlap;
 703   unsigned int i;
 704   unsigned int j;
 705
 706   if (NULL == path)
 707     return 0;
 708
 709   overlap = 0;
 710   GNUNET_assert (NULL != peer->tunnel);
 711
 712   for (i = 0; i < path->length; i++)
 713   {
 714     for (c = peer->tunnel->connection_head; NULL != c; c = c->next)
 715     {
 716       for (j = 0; j < c->path->length; j++)
 717       {
 718         if (path->peers[i] == c->path->peers[j])
 719         {
 720           overlap++;
 721           break;
 722         }
 723       }
 724     }
 725   }
 726   return (path->length + overlap) * (path->score * -1);
 727 }
 728
 729
 730 /**
 731  * Choose the best path towards a peer considering the tunnel properties.
 732  *
 733  * @param peer The destination peer.
 734  *
 735  * @return Best current known path towards the peer, if any.
 736  */
 737 static struct MeshPeerPath *
 738 peer_get_best_path (const struct MeshPeer *peer)
 739 {
 740   struct MeshPeerPath *best_p;
 741   struct MeshPeerPath *p;
 742   struct MeshConnection *c;
 743   unsigned int best_cost;
 744   unsigned int cost;
 745
 746   best_cost = UINT_MAX;
 747   best_p = NULL;
 748   for (p = peer->path_head; NULL != p; p = p->next)
 749   {
 750     for (c = peer->tunnel->connection_head; NULL != c; c = c->next)
 751       if (c->path == p)
 752         break;
 753       if (NULL != c)
 754         continue; /* If path is in use in a connection, skip it. */
 755
 756             if ((cost = peer_get_path_cost (peer, p)) < best_cost)
 757             {
 758               best_cost = cost;
 759               best_p = p;
 760             }
 761   }
 762     return best_p;
 763 }
 764
 765
 766 /**
 767  * Add the path to the peer and update the path used to reach it in case this
 768  * is the shortest.
 769  *
 770  * @param peer_info Destination peer to add the path to.
 771  * @param path New path to add. Last peer must be the peer in arg 1.
 772  *             Path will be either used of freed if already known.
 773  * @param trusted Do we trust that this path is real?
 774  */
 775 void
 776 peer_add_path (struct MeshPeer *peer_info, struct MeshPeerPath *path,
 777                int trusted)
 778 {
 779   struct MeshPeerPath *aux;
 780   unsigned int l;
 781   unsigned int l2;
 782
 783   if ((NULL == peer_info) || (NULL == path))
 784   {
 785     GNUNET_break (0);
 786     path_destroy (path);
 787     return;
 788   }
 789     if (path->peers[path->length - 1] != peer_info->id)
 790     {
 791       GNUNET_break (0);
 792       path_destroy (path);
 793       return;
 794     }
 795       if (2 >= path->length && GNUNET_NO == trusted)
 796       {
 797         /* Only allow CORE to tell us about direct paths */
 798         path_destroy (path);
 799         return;
 800       }
 801         for (l = 1; l < path->length; l++)
 802         {
 803           if (path->peers[l] == myid)
 804           {
 805             LOG (GNUNET_ERROR_TYPE_DEBUG, "shortening path by %u\n", l);
 806             for (l2 = 0; l2 < path->length - l; l2++)
 807             {
 808               path->peers[l2] = path->peers[l + l2];
 809             }
 810                   path->length -= l;
 811                   l = 1;
 812                   path->peers =
 813                             GNUNET_realloc (path->peers, path->length * sizeof (GNUNET_PEER_Id));
 814           }
 815         }
 816
 817           LOG (GNUNET_ERROR_TYPE_DEBUG, "adding path [%u] to peer %s\n",
 818                       path->length, peer2s (peer_info));
 819
 820           l = path_get_length (path);
 821           if (0 == l)
 822           {
 823             path_destroy (path);
 824             return;
 825           }
 826
 827             GNUNET_assert (peer_info->id == path->peers[path->length - 1]);
 828             for (aux = peer_info->path_head; aux != NULL; aux = aux->next)
 829             {
 830               l2 = path_get_length (aux);
 831               if (l2 > l)
 832               {
 833                 GNUNET_CONTAINER_DLL_insert_before (peer_info->path_head,
 834                                                     peer_info->path_tail, aux, path);
 835                 return;
 836               }
 837                   else
 838                   {
 839                     if (l2 == l && memcmp (path->peers, aux->peers, l) == 0)
 840                     {
 841                       path_destroy (path);
 842                       return;
 843                     }
 844                   }
 845             }
 846               GNUNET_CONTAINER_DLL_insert_tail (peer_info->path_head, peer_info->path_tail,
 847                                                 path);
 848               return;
 849 }
 850
 851
 852 /**
 853  * Add the path to the origin peer and update the path used to reach it in case
 854  * this is the shortest.
 855  * The path is given in peer_info -> destination, therefore we turn the path
 856  * upside down first.
 857  *
 858  * @param peer_info Peer to add the path to, being the origin of the path.
 859  * @param path New path to add after being inversed.
 860  *             Path will be either used or freed.
 861  * @param trusted Do we trust that this path is real?
 862  */
 863 static void
 864 peer_add_path_to_origin (struct MeshPeer *peer_info,
 865                          struct MeshPeerPath *path, int trusted)
 866 {
 867   if (NULL == path)
 868     return;
 869   path_invert (path);
 870   peer_add_path (peer_info, path, trusted);
 871 }
 872
 873
 874 /**
 875  * Adds a path to the peer_infos of all the peers in the path
 876  *
 877  * @param p Path to process.
 878  * @param confirmed Whether we know if the path works or not.
 879  */
 880 static void
 881 path_add_to_peers (struct MeshPeerPath *p, int confirmed)
 882 {
 883   unsigned int i;
 884
 885   /* TODO: invert and add */
 886   for (i = 0; i < p->length && p->peers[i] != myid; i++) /* skip'em */ ;
 887   for (i++; i < p->length; i++)
 888   {
 889     struct MeshPeer *aux;
 890     struct MeshPeerPath *copy;
 891
 892     aux = peer_get_short (p->peers[i]);
 893     copy = path_duplicate (p);
 894     copy->length = i + 1;
 895     peer_add_path (aux, copy, p->length < 3 ? GNUNET_NO : confirmed);
 896   }
 897 }
 898
 899
 900 /**
 901  * Function to process paths received for a new peer addition. The recorded
 902  * paths form the initial tunnel, which can be optimized later.
 903  * Called on each result obtained for the DHT search.
 904  *
 905  * @param cls closure
 906  * @param path
 907  */
 908 static void
 909 search_handler (void *cls, struct MeshPeerPath *path)
 910 {
 911   struct MeshPeer *peer = cls;
 912   unsigned int connection_count;
 913
 914   path_add_to_peers (path, GNUNET_NO);
 915
 916   /* Count connections */
 917   connection_count = GMC_count (peer->tunnel->connection_head);
 918
 919   /* If we already have 3 (or more (?!)) connections, it's enough */
 920   if (3 <= connection_count)
 921     return;
 922
 923   if (peer->tunnel->state == MESH_TUNNEL_SEARCHING)
 924   {
 925     LOG (GNUNET_ERROR_TYPE_DEBUG, " ... connect!\n");
 926     GMP_connect (peer);
 927   }
 928   return;
 929 }
 930
 931
 932 /**
 933  * Core callback to write a queued packet to core buffer
 934  *
 935  * @param cls Closure (peer info).
 936  * @param size Number of bytes available in buf.
 937  * @param buf Where the to write the message.
 938  *
 939  * @return number of bytes written to buf
 940  */
 941 static size_t
 942 queue_send (void *cls, size_t size, void *buf)
 943 {
 944   struct MeshPeer *peer = cls;
 945   struct MeshFlowControl *fc;
 946   struct MeshConnection *c;
 947   struct GNUNET_MessageHeader *msg;
 948   struct MeshPeerQueue *queue;
 949   struct MeshTunnel2 *t;
 950   struct MeshChannel *ch;
 951   const struct GNUNET_PeerIdentity *dst_id;
 952   size_t data_size;
 953   uint32_t pid;
 954   uint16_t type;
 955   int fwd;
 956
 957   peer->core_transmit = NULL;
 958   LOG (GNUNET_ERROR_TYPE_DEBUG, "* Queue send (max %u)\n", size);
 959
 960   if (NULL == buf || 0 == size)
 961   {
 962     LOG (GNUNET_ERROR_TYPE_DEBUG, "* Buffer size 0.\n");
 963     return 0;
 964   }
 965
 966   /* Initialize */
 967   queue = peer_get_first_message (peer);
 968   if (NULL == queue)
 969   {
 970     GNUNET_break (0); /* Core tmt_rdy should've been canceled */
 971     return 0;
 972   }
 973   c = queue->c;
 974   fwd = queue->fwd;
 975   fc = fwd ? &c->fwd_fc : &c->bck_fc;
 976
 977   dst_id = GNUNET_PEER_resolve2 (peer->id);
 978   LOG (GNUNET_ERROR_TYPE_DEBUG, "*   towards %s\n", GNUNET_i2s (dst_id));
 979   /* Check if buffer size is enough for the message */
 980   if (queue->size > size)
 981   {
 982       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   not enough room, reissue\n");
 983       peer->core_transmit =
 984           GNUNET_CORE_notify_transmit_ready (core_handle,
 985                                              GNUNET_NO,
 986                                              0,
 987                                              GNUNET_TIME_UNIT_FOREVER_REL,
 988                                              dst_id,
 989                                              queue->size,
 990                                              &queue_send,
 991                                              peer);
 992       return 0;
 993   }
 994   LOG (GNUNET_ERROR_TYPE_DEBUG, "*   size %u ok\n", queue->size);
 995
 996   t = (NULL != c) ? c->t : NULL;
 997   type = 0;
 998
 999   /* Fill buf */
1000   switch (queue->type)
1001   {
1002     case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
1003     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
1004     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
1005     case GNUNET_MESSAGE_TYPE_MESH_FWD:
1006     case GNUNET_MESSAGE_TYPE_MESH_BCK:
1007     case GNUNET_MESSAGE_TYPE_MESH_ACK:
1008     case GNUNET_MESSAGE_TYPE_MESH_POLL:
1009       LOG (GNUNET_ERROR_TYPE_DEBUG,
1010                   "*   raw: %s\n",
1011                   GNUNET_MESH_DEBUG_M2S (queue->type));
1012       data_size = send_core_data_raw (queue->cls, size, buf);
1013       msg = (struct GNUNET_MessageHeader *) buf;
1014       type = ntohs (msg->type);
1015       break;
1016     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
1017       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   path create\n");
1018       if (GMC_is_origin (c, GNUNET_YES))
1019         data_size = send_core_connection_create (queue->c, size, buf);
1020       else
1021         data_size = send_core_data_raw (queue->cls, size, buf);
1022       break;
1023     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
1024       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   path ack\n");
1025       if (GMC_is_origin (c, GNUNET_NO) ||
1026           GMC_is_origin (c, GNUNET_YES))
1027         data_size = send_core_connection_ack (queue->c, size, buf);
1028       else
1029         data_size = send_core_data_raw (queue->cls, size, buf);
1030       break;
1031     case GNUNET_MESSAGE_TYPE_MESH_DATA:
1032     case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
1033     case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
1034       /* This should be encapsulted */
1035       GNUNET_break (0);
1036       data_size = 0;
1037       break;
1038     default:
1039       GNUNET_break (0);
1040       LOG (GNUNET_ERROR_TYPE_WARNING, "*   type unknown: %u\n",
1041                   queue->type);
1042       data_size = 0;
1043   }
1044
1045   if (0 < drop_percent &&
1046       GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 101) < drop_percent)
1047   {
1048     LOG (GNUNET_ERROR_TYPE_WARNING,
1049                 "Dropping message of type %s\n",
1050                 GNUNET_MESH_DEBUG_M2S (queue->type));
1051     data_size = 0;
1052   }
1053
1054   if (NULL != queue->callback)
1055   {
1056     LOG (GNUNET_ERROR_TYPE_DEBUG, "*   Calling callback\n");
1057     queue->callback (queue->callback_cls,
1058                     queue->c,
1059                     GNUNET_TIME_absolute_get_duration (queue->start_waiting));
1060   }
1061
1062   /* Free queue, but cls was freed by send_core_* */
1063   ch = queue->ch;
1064   GMP_queue_destroy (queue, GNUNET_NO);
1065
1066   /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
1067   switch (type)
1068   {
1069     case GNUNET_MESSAGE_TYPE_MESH_FWD:
1070     case GNUNET_MESSAGE_TYPE_MESH_BCK:
1071       pid = ntohl ( ((struct GNUNET_MESH_Encrypted *) buf)->pid );
1072       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   accounting pid %u\n", pid);
1073       fc->last_pid_sent = pid;
1074       send_ack (c, ch, fwd);
1075       break;
1076     default:
1077       break;
1078   }
1079
1080   /* If more data in queue, send next */
1081   queue = peer_get_first_message (peer);
1082   if (NULL != queue)
1083   {
1084     LOG (GNUNET_ERROR_TYPE_DEBUG, "*   more data!\n");
1085     if (NULL == peer->core_transmit) {
1086       peer->core_transmit =
1087           GNUNET_CORE_notify_transmit_ready(core_handle,
1088                                             0,
1089                                             0,
1090                                             GNUNET_TIME_UNIT_FOREVER_REL,
1091                                             dst_id,
1092                                             queue->size,
1093                                             &queue_send,
1094                                             peer);
1095       queue->start_waiting = GNUNET_TIME_absolute_get ();
1096     }
1097     else
1098     {
1099       LOG (GNUNET_ERROR_TYPE_DEBUG,
1100                   "*   tmt rdy called somewhere else\n");
1101     }
1102     if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
1103     {
1104       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   starting poll timeout\n");
1105       fc->poll_task =
1106           GNUNET_SCHEDULER_add_delayed (fc->poll_time, &connection_poll, fc);
1107     }
1108   }
1109   else
1110   {
1111     if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
1112     {
1113       GNUNET_SCHEDULER_cancel (fc->poll_task);
1114       fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
1115     }
1116   }
1117   if (NULL != c)
1118   {
1119     c->pending_messages--;
1120     if (GNUNET_YES == c->destroy && 0 == c->pending_messages)
1121     {
1122       LOG (GNUNET_ERROR_TYPE_DEBUG, "*  destroying connection!\n");
1123       GMC_destroy (c);
1124     }
1125   }
1126
1127   if (NULL != t)
1128   {
1129     t->pending_messages--;
1130     if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
1131     {
1132 //       LOG (GNUNET_ERROR_TYPE_DEBUG, "*  destroying tunnel!\n");
1133       tunnel_destroy (t);
1134     }
1135   }
1136   LOG (GNUNET_ERROR_TYPE_DEBUG, "*  Return %d\n", data_size);
1137   return data_size;
1138 }
1139
1140
1141
1142 /**
1143  * Get first sendable message.
1144  *
1145  * @param peer The destination peer.
1146  *
1147  * @return Best current known path towards the peer, if any.
1148  */
1149 static struct MeshPeerQueue *
1150 peer_get_first_message (const struct MeshPeer *peer)
1151 {
1152   struct MeshPeerQueue *q;
1153
1154   for (q = peer->queue_head; NULL != q; q = q->next)
1155   {
1156     if (queue_is_sendable (q))
1157       return q;
1158   }
1159
1160   return NULL;
1161 }
1162
1163
1164 static int
1165 queue_is_sendable (struct MeshPeerQueue *q)
1166 {
1167   struct MeshFlowControl *fc;
1168
1169   /* Is PID-independent? */
1170   switch (q->type)
1171   {
1172     case GNUNET_MESSAGE_TYPE_MESH_ACK:
1173     case GNUNET_MESSAGE_TYPE_MESH_POLL:
1174       return GNUNET_YES;
1175   }
1176
1177   /* Is PID allowed? */
1178   fc = q->fwd ? &q->c->fwd_fc : &q->c->bck_fc;
1179   if (GMC_is_pid_bigger (fc->last_ack_recv, fc->last_pid_sent))
1180     return GNUNET_YES;
1181
1182   return GNUNET_NO;
1183 }
1184
1185
1186 /******************************************************************************/
1187 /********************************    API    ***********************************/
1188 /******************************************************************************/
1189
1190
1191 /**
1192  * Free a transmission that was already queued with all resources
1193  * associated to the request.
1194  *
1195  * @param queue Queue handler to cancel.
1196  * @param clear_cls Is it necessary to free associated cls?
1197  */
1198 void
1199 GMP_queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
1200 {
1201   struct MeshPeer *peer;
1202   struct MeshFlowControl *fc;
1203   int fwd;
1204
1205   fwd = queue->fwd;
1206   peer = queue->peer;
1207   GNUNET_assert (NULL != queue->c);
1208   fc = fwd ? &queue->c->fwd_fc : &queue->c->bck_fc;
1209
1210   if (GNUNET_YES == clear_cls)
1211   {
1212     LOG (GNUNET_ERROR_TYPE_DEBUG, "   queue destroy type %s\n",
1213                 GNUNET_MESH_DEBUG_M2S (queue->type));
1214     switch (queue->type)
1215     {
1216       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
1217       case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
1218         LOG (GNUNET_ERROR_TYPE_INFO, "destroying a DESTROY message\n");
1219         GNUNET_break (GNUNET_YES == queue->c->destroy);
1220         /* fall through */
1221       case GNUNET_MESSAGE_TYPE_MESH_FWD:
1222       case GNUNET_MESSAGE_TYPE_MESH_BCK:
1223       case GNUNET_MESSAGE_TYPE_MESH_ACK:
1224       case GNUNET_MESSAGE_TYPE_MESH_POLL:
1225       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
1226       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
1227       case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
1228         LOG (GNUNET_ERROR_TYPE_DEBUG, "   prebuilt message\n");;
1229         GNUNET_free_non_null (queue->cls);
1230         break;
1231
1232       default:
1233         GNUNET_break (0);
1234         LOG (GNUNET_ERROR_TYPE_ERROR, "   type %s unknown!\n",
1235                     GNUNET_MESH_DEBUG_M2S (queue->type));
1236     }
1237
1238   }
1239   GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
1240
1241   if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
1242       queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
1243   {
1244     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Q_N- %p %u\n", fc, fc->queue_n);
1245     fc->queue_n--;
1246     peer->queue_n--;
1247   }
1248   if (NULL != queue->c)
1249   {
1250     queue->c->pending_messages--;
1251     if (NULL != queue->c->t)
1252     {
1253       queue->c->t->pending_messages--;
1254     }
1255   }
1256
1257   GNUNET_free (queue);
1258 }
1259
1260
1261 /**
1262  * @brief Queue and pass message to core when possible.
1263  *
1264  * @param cls Closure (@c type dependant). It will be used by queue_send to
1265  *            build the message to be sent if not already prebuilt.
1266  * @param type Type of the message, 0 for a raw message.
1267  * @param size Size of the message.
1268  * @param c Connection this message belongs to (cannot be NULL).
1269  * @param ch Channel this message belongs to, if applicable (otherwise NULL).
1270  * @param fwd Is this a message going root->dest? (FWD ACK are NOT FWD!)
1271  * @param callback Function to be called once CORE has taken the message.
1272  * @param callback_cls Closure for @c callback.
1273  */
1274 void
1275 GMP_queue_add (void *cls, uint16_t type, size_t size,
1276                struct MeshConnection *c,
1277                struct MeshChannel *ch,
1278                int fwd,
1279                GMP_sent callback, void *callback_cls)
1280 {
1281   struct MeshPeerQueue *queue;
1282   struct MeshFlowControl *fc;
1283   struct MeshPeer *peer;
1284   int priority;
1285   int call_core;
1286
1287   LOG (GNUNET_ERROR_TYPE_DEBUG,
1288               "queue add %s %s (%u) on c %p, ch %p\n",
1289               fwd ? "FWD" : "BCK",  GNUNET_MESH_DEBUG_M2S (type), size, c, ch);
1290   GNUNET_assert (NULL != c);
1291
1292   fc   = fwd ? &c->fwd_fc : &c->bck_fc;
1293   peer = fwd ? connection_get_next_hop (c) : connection_get_prev_hop (c);
1294
1295   if (NULL == fc)
1296   {
1297     GNUNET_break (0);
1298     return;
1299   }
1300
1301   if (NULL == peer->connections)
1302   {
1303     /* We are not connected to this peer, ignore request. */
1304     GNUNET_break_op (0);
1305     return;
1306   }
1307
1308   priority = 0;
1309
1310   if (GNUNET_MESSAGE_TYPE_MESH_POLL == type ||
1311       GNUNET_MESSAGE_TYPE_MESH_ACK == type)
1312   {
1313     priority = 100;
1314   }
1315
1316   LOG (GNUNET_ERROR_TYPE_DEBUG, "priority %d\n", priority);
1317   LOG (GNUNET_ERROR_TYPE_DEBUG, "fc %p\n", fc);
1318   if (fc->queue_n >= fc->queue_max && 0 == priority)
1319   {
1320     GNUNET_STATISTICS_update (stats, "# messages dropped (buffer full)",
1321                               1, GNUNET_NO);
1322     GNUNET_break (0);
1323     LOG (GNUNET_ERROR_TYPE_DEBUG,
1324                 "queue full: %u/%u\n",
1325                 fc->queue_n, fc->queue_max);
1326     return; /* Drop this message */
1327   }
1328
1329   LOG (GNUNET_ERROR_TYPE_DEBUG, "last pid %u\n", fc->last_pid_sent);
1330   LOG (GNUNET_ERROR_TYPE_DEBUG, "     ack %u\n", fc->last_ack_recv);
1331   if (GMC_is_pid_bigger (fc->last_pid_sent + 1, fc->last_ack_recv))
1332   {
1333     call_core = GNUNET_NO;
1334     if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task &&
1335         GNUNET_MESSAGE_TYPE_MESH_POLL != type)
1336     {
1337       LOG (GNUNET_ERROR_TYPE_DEBUG,
1338                   "no buffer space (%u > %u): starting poll\n",
1339                   fc->last_pid_sent + 1, fc->last_ack_recv);
1340       fc->poll_task = GNUNET_SCHEDULER_add_delayed (fc->poll_time,
1341                                                     &connection_poll,
1342                                                     fc);
1343     }
1344   }
1345   else
1346     call_core = GNUNET_YES;
1347   queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
1348   queue->cls = cls;
1349   queue->type = type;
1350   queue->size = size;
1351   queue->peer = peer;
1352   queue->c = c;
1353   queue->ch = ch;
1354   queue->fwd = fwd;
1355   queue->callback = callback;
1356   queue->callback_cls = callback_cls;
1357   if (100 <= priority)
1358   {
1359     struct MeshPeerQueue *copy;
1360     struct MeshPeerQueue *next;
1361
1362     for (copy = peer->queue_head; NULL != copy; copy = next)
1363     {
1364       next = copy->next;
1365       if (copy->type == type && copy->c == c && copy->fwd == fwd)
1366       {
1367         /* Example: also a FWD ACK for connection XYZ */
1368         GMP_queue_destroy (copy, GNUNET_YES);
1369       }
1370     }
1371     GNUNET_CONTAINER_DLL_insert (peer->queue_head, peer->queue_tail, queue);
1372   }
1373   else
1374   {
1375     GNUNET_CONTAINER_DLL_insert_tail (peer->queue_head, peer->queue_tail, queue);
1376     LOG (GNUNET_ERROR_TYPE_DEBUG, "  Q_N+ %p %u\n", fc, fc->queue_n);
1377     fc->queue_n++;
1378     peer->queue_n++;
1379   }
1380
1381   if (NULL == peer->core_transmit && GNUNET_YES == call_core)
1382   {
1383     LOG (GNUNET_ERROR_TYPE_DEBUG,
1384                 "calling core tmt rdy towards %s for %u bytes\n",
1385                 peer2s (peer), size);
1386     peer->core_transmit =
1387         GNUNET_CORE_notify_transmit_ready (core_handle,
1388                                            0,
1389                                            0,
1390                                            GNUNET_TIME_UNIT_FOREVER_REL,
1391                                            GNUNET_PEER_resolve2 (peer->id),
1392                                            size,
1393                                            &queue_send,
1394                                            peer);
1395     queue->start_waiting = GNUNET_TIME_absolute_get ();
1396   }
1397   else
1398   {
1399     LOG (GNUNET_ERROR_TYPE_DEBUG,
1400                 "core tmt rdy towards %s already called\n",
1401                 peer2s (peer));
1402
1403   }
1404   c->pending_messages++;
1405   if (NULL != c->t)
1406     c->t->pending_messages++;
1407 }
1408
1409
1410 /**
1411  * Cancel all queued messages to a peer that belong to a certain connection.
1412  *
1413  * @param peer Peer towards whom to cancel.
1414  * @param c Connection whose queued messages to cancel.
1415  */
1416 void
1417 GMP_queue_cancel (struct MeshPeer *peer, struct MeshConnection *c)
1418 {
1419   struct MeshPeerQueue *q;
1420   struct MeshPeerQueue *next;
1421
1422   for (q = peer->queue_head; NULL != q; q = next)
1423   {
1424     next = q->next;
1425     if (q->c == c)
1426     {
1427       LOG (GNUNET_ERROR_TYPE_DEBUG,
1428                   "connection_cancel_queue %s\n",
1429                   GNUNET_MESH_DEBUG_M2S (q->type));
1430       GMP_queue_destroy (q, GNUNET_YES);
1431     }
1432   }
1433   if (NULL == peer->queue_head)
1434   {
1435     if (NULL != peer->core_transmit)
1436     {
1437       GNUNET_CORE_notify_transmit_ready_cancel (peer->core_transmit);
1438       peer->core_transmit = NULL;
1439     }
1440   }
1441 }
1442
1443
1444 /**
1445  * Initialize the peer subsystem.
1446  *
1447  * @param c Configuration.
1448  * @param id Peer identity
1449  */
1450 void
1451 GMP_init (const struct GNUNET_CONFIGURATION_Handle *c,
1452           const struct GNUNET_PeerIdentity *id)
1453 {
1454   my_full_id = id;
1455   peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
1456   if (GNUNET_OK !=
1457       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "MAX_PEERS",
1458                                              &max_peers))
1459   {
1460     LOG_config_invalid (GNUNET_ERROR_TYPE_WARNING,
1461                                "MESH", "MAX_PEERS", "USING DEFAULT");
1462     max_peers = 1000;
1463   }
1464
1465   if (GNUNET_OK !=
1466       GNUNET_CONFIGURATION_get_value_number (c, "MESH", "DROP_PERCENT",
1467                                              &drop_percent))
1468   {
1469     drop_percent = 0;
1470   }
1471   else
1472   {
1473     LOG (GNUNET_ERROR_TYPE_WARNING,
1474                 "\n***************************************\n"
1475                 "Mesh is running with drop mode enabled.\n"
1476                 "This is NOT a good idea!\n"
1477                 "Remove the DROP_PERCENT option from your configuration.\n"
1478                 "***************************************\n");
1479   }
1480
1481   core_handle = GNUNET_CORE_connect (c, /* Main configuration */
1482                                      NULL,      /* Closure passed to MESH functions */
1483                                      &core_init,        /* Call core_init once connected */
1484                                      &core_connect,     /* Handle connects */
1485                                      &core_disconnect,  /* remove peers on disconnects */
1486                                      NULL,      /* Don't notify about all incoming messages */
1487                                      GNUNET_NO, /* For header only in notification */
1488                                      NULL,      /* Don't notify about all outbound messages */
1489                                      GNUNET_NO, /* For header-only out notification */
1490                                      core_handlers);    /* Register these handlers */
1491   if (NULL == core_handle)
1492   {
1493     GNUNET_break (0);
1494     GNUNET_SCHEDULER_shutdown ();
1495     return;
1496   }
1497 }
1498
1499 /**
1500  * Shut down the peer subsystem.
1501  */
1502 void
1503 GMP_shutdown (void)
1504 {
1505   GNUNET_CONTAINER_multipeermap_iterate (peers, &shutdown_tunnel, NULL);
1506
1507   if (core_handle != NULL)
1508   {
1509     GNUNET_CORE_disconnect (core_handle);
1510     core_handle = NULL;
1511   }
1512 }
1513
1514
1515 /**
1516  * Try to establish a new connection to this peer in the given tunnel.
1517  * If the peer doesn't have any path to it yet, try to get one.
1518  * If the peer already has some path, send a CREATE CONNECTION towards it.
1519  *
1520  * @param peer PeerInfo of the peer.
1521  */
1522 void
1523 GMP_connect (struct MeshPeer *peer)
1524 {
1525   struct MeshTunnel2 *t;
1526   struct MeshPeerPath *p;
1527   struct MeshConnection *c;
1528   int rerun_search;
1529
1530   LOG (GNUNET_ERROR_TYPE_DEBUG,
1531               "peer_connect towards %s\n",
1532               peer2s (peer));
1533   t = peer->tunnel;
1534   c = NULL;
1535   rerun_search = GNUNET_NO;
1536
1537   if (NULL != peer->path_head)
1538   {
1539     LOG (GNUNET_ERROR_TYPE_DEBUG, "path exists\n");
1540     p = peer_get_best_path (peer);
1541     if (NULL != p)
1542     {
1543       LOG (GNUNET_ERROR_TYPE_DEBUG, "  %u hops\n", p->length);
1544       c = tunnel_use_path (t, p);
1545       if (NULL == c)
1546       {
1547         /* This case can happen when the path includes a first hop that is
1548          * not yet known to be connected.
1549          *
1550          * This happens quite often during testing when running mesh
1551          * under valgrind: core connect notifications come very late and the
1552          * DHT result has already come and created a valid path.
1553          * In this case, the peer->connections hashmap will be NULL and
1554          * tunnel_use_path will not be able to create a connection from that
1555          * path.
1556          *
1557          * Re-running the DHT GET should give core time to callback.
1558          */
1559         GNUNET_break(0);
1560                 rerun_search = GNUNET_YES;
1561       }
1562       else
1563       {
1564         send_connection_create (c);
1565         return;
1566       }
1567     }
1568   }
1569
1570   if (NULL != peer->search_h && GNUNET_YES == rerun_search)
1571   {
1572     GMD_search_stop (peer->search_h);
1573     peer->search_h = NULL;
1574     LOG (GNUNET_ERROR_TYPE_DEBUG,
1575          "  Stopping DHT GET for peer %s\n",
1576          GMP_2s (peer));
1577   }
1578
1579   if (NULL == peer->search_h)
1580   {
1581     const struct GNUNET_PeerIdentity *id;
1582
1583     id = GNUNET_PEER_resolve2 (peer->id);
1584     LOG (GNUNET_ERROR_TYPE_DEBUG,
1585                 "  Starting DHT GET for peer %s\n", peer2s (peer));
1586     peer->search_h = GMD_search (id, &search_handler, peer);
1587     if (MESH_TUNNEL_NEW == t->state)
1588       GMT_change_state (t, MESH_TUNNEL_SEARCHING);
1589   }
1590 }
1591
1592
1593 /**
1594  * Set tunnel.
1595  *
1596  * @param peer Peer.
1597  * @param t Tunnel.
1598  */
1599 void
1600 GMP_set_tunnel (struct MeshPeer *peer, struct MeshTunnel2 *t)
1601 {
1602   peer->tunnel = t;
1603 }
1604
1605
1606 /**
1607  * Chech whether there is a direct (core level)  connection to peer.
1608  *
1609  * @param peer Peer to check.
1610  *
1611  * @return GNUNET_YES if there is a direct connection.
1612  */
1613 int
1614 GMP_is_neighbor (const struct MeshPeer *peer)
1615 {
1616   struct MeshPeerPath *path;
1617
1618   if (NULL == peer->connections)
1619     return GNUNET_NO;
1620
1621   for (path = peer->path_head; NULL != path; path = path->next)
1622   {
1623     if (3 > path->length)
1624       return GNUNET_YES;
1625   }
1626
1627   GNUNET_break (0); /* Is not a neighbor but connections is not NULL */
1628   return GNUNET_NO;
1629 }
1630
1631
1632 /**
1633  * Add a connection to a neighboring peer.
1634  *
1635  * Store that the peer is the first hop of the connection in one
1636  * direction and that on peer disconnect the connection must be
1637  * notified and destroyed, for it will no longer be valid.
1638  *
1639  * @param peer Peer to add connection to.
1640  * @param c Connection to add.
1641  *
1642  * @return GNUNET_OK on success.
1643  */
1644 int
1645 GMP_add_connection (struct MeshPeer *peer,
1646                     const struct MeshConnection *c)
1647 {
1648   if (NULL == peer->connections)
1649   {
1650     GNUNET_break (0);
1651     return GNUNET_SYSERR;
1652   }
1653   return GNUNET_CONTAINER_multihashmap_put (peer->connections,
1654                                             GMC_get_id (c),
1655                                             c,
1656                                             GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1657 }
1658
1659
1660 /**
1661  * Remove a connection from a neighboring peer.
1662  *
1663  * @param peer Peer to remove connection from.
1664  * @param c Connection to remove.
1665  *
1666  * @return GNUNET_OK on success.
1667  */
1668 int
1669 GMP_remove_connection (struct MeshPeer *peer,
1670                        const struct MeshConnection *c)
1671 {
1672   if (NULL == peer || NULL == peer->connections)
1673   {
1674     GNUNET_break (0);
1675     return GNUNET_SYSERR;
1676   }
1677   return GNUNET_CONTAINER_multihashmap_remove (peer->connections,
1678                                                GMC_get_id (c),
1679                                                c);
1680 }
1681
1682 /**
1683  * Get the Full ID of a peer.
1684  *
1685  * @param peer Peer to get from.
1686  *
1687  * @return Full ID of peer.
1688  */
1689 struct GNUNET_PeerIdentity *
1690 GMP_get_id (const struct MeshPeer *peer)
1691 {
1692   return GNUNET_PEER_resolve2 (peer->id);
1693 }
1694
1695
1696 /**
1697  * Get the Short ID of a peer.
1698  *
1699  * @param peer Peer to get from.
1700  *
1701  * @return Short ID of peer.
1702  */
1703 GNUNET_PEER_Id
1704 GMP_get_short_id (const struct MeshPeer *peer)
1705 {
1706   return peer->id;
1707 }
1708
1709
1710 /**
1711  * Get the static string for a peer ID.
1712  *
1713  * @param peer Peer.
1714  *
1715  * @return Static string for it's ID.
1716  */
1717 const char *
1718 GMP_2s (const struct MeshPeer *peer)
1719 {
1720   if (NULL == peer)
1721     return "(NULL)";
1722   return GNUNET_i2s (GNUNET_PEER_resolve2 (peer->id));
1723 }