src/dv/gnunet-service-dv.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  * @file dv/gnunet-service-dv.c
  23  * @brief the distance vector service, primarily handles gossip of nearby
  24  * peers and sending/receiving DV messages from core and decapsulating
  25  * them
  26  *
  27  * @author Christian Grothoff
  28  * @author Nathan Evans
  29  *
  30  * TODO:
  31  * - routing tables are not exchanged with direct neighbors
  32  * - distance updates are not properly communicate to US by core,
  33  *   and conversely we don't give distance updates properly to the plugin yet
  34  * - even local flow control (send ACK only after core took our message) is
  35  *   not implemented, but should be (easy fix)
  36  * - we send 'ACK' even if a message was dropped due to no route (may
  37  *   be harmless, but should at least be documented)
  38  */
  39 #include "platform.h"
  40 #include "gnunet_util_lib.h"
  41 #include "gnunet_protocols.h"
  42 #include "gnunet_core_service.h"
  43 #include "gnunet_hello_lib.h"
  44 #include "gnunet_peerinfo_service.h"
  45 #include "gnunet_statistics_service.h"
  46 #include "gnunet_consensus_service.h"
  47 #include "dv.h"
  48
  49 /**
  50  * How often do we establish the consensu?
  51  */
  52 #define GNUNET_DV_CONSENSUS_FREQUENCY GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 5)
  53
  54 /**
  55  * Maximum number of messages we queue per peer.
  56  */
  57 #define MAX_QUEUE_SIZE 16
  58
  59 /**
  60  * The default fisheye depth, from how many hops away will
  61  * we keep peers?
  62  */
  63 #define DEFAULT_FISHEYE_DEPTH 3
  64
  65 /**
  66  * How many hops is a direct neighbor away?
  67  */
  68 #define DIRECT_NEIGHBOR_COST 1
  69
  70
  71 GNUNET_NETWORK_STRUCT_BEGIN
  72
  73 /**
  74  * Information about a peer DV can route to.  These entries are what
  75  * we use as the binary format to establish consensus to create our
  76  * routing table and as the address format in the HELLOs.
  77  */
  78 struct Target
  79 {
  80
  81   /**
  82    * Identity of the peer we can reach.
  83    */
  84   struct GNUNET_PeerIdentity peer;
  85
  86   /**
  87    * How many hops (1-3) is this peer away?
  88    */
  89   uint32_t distance GNUNET_PACKED;
  90
  91 };
  92
  93
  94 /**
  95  * Message exchanged between DV services (via core), requesting a
  96  * message to be routed.
  97  */
  98 struct RouteMessage
  99 {
 100   /**
 101    * Type: GNUNET_MESSAGE_TYPE_DV_ROUTE
 102    */
 103   struct GNUNET_MessageHeader header;
 104
 105   /**
 106    * Expected (remaining) distance.  Must be always smaller than
 107    * DEFAULT_FISHEYE_DEPTH, should be zero at the target.  Must
 108    * be decremented by one at each hop.  Peers must not forward
 109    * these messages further once the counter has reached zero.
 110    */
 111   uint32_t distance GNUNET_PACKED;
 112
 113   /**
 114    * The (actual) target of the message (this peer, if distance is zero).
 115    */
 116   struct GNUNET_PeerIdentity target;
 117
 118   /**
 119    * The (actual) sender of the message.
 120    */
 121   struct GNUNET_PeerIdentity sender;
 122
 123 };
 124
 125 GNUNET_NETWORK_STRUCT_END
 126
 127
 128 /**
 129  * Linked list of messages to send to clients.
 130  */
 131 struct PendingMessage
 132 {
 133   /**
 134    * Pointer to next item in the list
 135    */
 136   struct PendingMessage *next;
 137
 138   /**
 139    * Pointer to previous item in the list
 140    */
 141   struct PendingMessage *prev;
 142
 143   /**
 144    * Actual message to be sent, allocated after this struct.
 145    */
 146   const struct GNUNET_MessageHeader *msg;
 147
 148   /**
 149    * Ultimate target for the message.
 150    */
 151   struct GNUNET_PeerIdentity ultimate_target;
 152
 153   /**
 154    * Unique ID of the message.
 155    */
 156   uint32_t uid;
 157
 158 };
 159
 160
 161 /**
 162  * Information about a direct neighbor (core-level, excluding
 163  * DV-links, only DV-enabled peers).
 164  */
 165 struct DirectNeighbor
 166 {
 167
 168   /**
 169    * Identity of the peer.
 170    */
 171   struct GNUNET_PeerIdentity peer;
 172
 173   /**
 174    * Head of linked list of messages to send to this peer.
 175    */
 176   struct PendingMessage *pm_head;
 177
 178   /**
 179    * Tail of linked list of messages to send to this peer.
 180    */
 181   struct PendingMessage *pm_tail;
 182
 183   /**
 184    * Transmit handle to core service.
 185    */
 186   struct GNUNET_CORE_TransmitHandle *cth;
 187
 188   /**
 189    * Routing table of the neighbor, NULL if not yet established.
 190    * Keys are peer identities, values are 'struct Target' entries.
 191    * Note that the distances in the targets are from the point-of-view
 192    * of the peer, not from us!
 193    */
 194   struct GNUNET_CONTAINER_MultiHashMap *neighbor_table;
 195
 196   /**
 197    * Updated routing table of the neighbor, under construction,
 198    * NULL if we are not currently building it.
 199    * Keys are peer identities, values are 'struct Target' entries.
 200    * Note that the distances in the targets are from the point-of-view
 201    * of the peer, not from us!
 202    */
 203   struct GNUNET_CONTAINER_MultiHashMap *neighbor_table_consensus;
 204
 205   /**
 206    * Active consensus, if we are currently synchronizing the
 207    * routing tables.
 208    */
 209   struct GNUNET_CONSENSUS_Handle *consensus;
 210
 211   /**
 212    * ID of the task we use to (periodically) update our consensus
 213    * with this peer.
 214    */
 215   GNUNET_SCHEDULER_TaskIdentifier consensus_task;
 216
 217   /**
 218    * At what offset are we, with respect to inserting our own routes
 219    * into the consensus?
 220    */
 221   unsigned int consensus_insertion_offset;
 222
 223   /**
 224    * At what distance are we, with respect to inserting our own routes
 225    * into the consensus?
 226    */
 227   unsigned int consensus_insertion_distance;
 228
 229   /**
 230    * Number of messages currently in the 'pm_XXXX'-DLL.
 231    */
 232   unsigned int pm_queue_size;
 233
 234 };
 235
 236
 237 /**
 238  * A route includes information about the next hop,
 239  * the target, and the ultimate distance to the
 240  * target.
 241  */
 242 struct Route
 243 {
 244
 245   /**
 246    * Which peer do we need to forward the message to?
 247    */
 248   struct DirectNeighbor *next_hop;
 249
 250   /**
 251    * What would be the target, and how far is it away?
 252    */
 253   struct Target target;
 254
 255   /**
 256    * Offset of this target in the respective consensus set.
 257    */
 258   unsigned int set_offset;
 259
 260 };
 261
 262
 263 /**
 264  * Set of targets we bring to a consensus; all targets in a set have a
 265  * distance equal to the sets distance (which is implied by the array
 266  * index of the set).
 267  */
 268 struct ConsensusSet
 269 {
 270
 271   /**
 272    * Array of targets in the set, may include NULL entries if a
 273    * neighbor has disconnected; the targets are allocated with the
 274    * respective container (all_routes), not here.
 275    */
 276   struct Route **targets;
 277
 278   /**
 279    * Size of the 'targets' array.
 280    */
 281   unsigned int array_length;
 282
 283 };
 284
 285
 286 /**
 287  * Hashmap of all of our direct neighbors (no DV routing).
 288  */
 289 static struct GNUNET_CONTAINER_MultiHashMap *direct_neighbors;
 290
 291 /**
 292  * Hashmap with all routes that we currently support; contains
 293  * routing information for all peers from distance 2
 294  * up to distance DEFAULT_FISHEYE_DEPTH.
 295  */
 296 static struct GNUNET_CONTAINER_MultiHashMap *all_routes;
 297
 298 /**
 299  * Array of consensus sets we expose to the outside world.  Sets
 300  * are structured by the distance to the target.
 301  */
 302 static struct ConsensusSet consensi[DEFAULT_FISHEYE_DEPTH - 1];
 303
 304 /**
 305  * Handle to the core service api.
 306  */
 307 static struct GNUNET_CORE_Handle *core_api;
 308
 309 /**
 310  * The identity of our peer.
 311  */
 312 static struct GNUNET_PeerIdentity my_identity;
 313
 314 /**
 315  * The configuration for this service.
 316  */
 317 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 318
 319 /**
 320  * The client, the DV plugin connected to us.  Hopefully
 321  * this client will never change, although if the plugin dies
 322  * and returns for some reason it may happen.
 323  */
 324 static struct GNUNET_SERVER_Client *client_handle;
 325
 326 /**
 327  * Transmit handle to the plugin.
 328  */
 329 static struct GNUNET_SERVER_TransmitHandle *plugin_transmit_handle;
 330
 331 /**
 332  * Head of DLL for client messages
 333  */
 334 static struct PendingMessage *plugin_pending_head;
 335
 336 /**
 337  * Tail of DLL for client messages
 338  */
 339 static struct PendingMessage *plugin_pending_tail;
 340
 341 /**
 342  * Handle for the statistics service.
 343  */
 344 struct GNUNET_STATISTICS_Handle *stats;
 345
 346
 347 /**
 348  * Get distance information from 'atsi'.
 349  *
 350  * @param atsi performance data
 351  * @param atsi_count number of entries in atsi
 352  * @return connected transport distance
 353  */
 354 static uint32_t
 355 get_atsi_distance (const struct GNUNET_ATS_Information *atsi,
 356                    unsigned int atsi_count)
 357 {
 358   unsigned int i;
 359
 360   for (i = 0; i < atsi_count; i++)
 361     if (ntohl (atsi[i].type) == GNUNET_ATS_QUALITY_NET_DISTANCE)
 362       return ntohl (atsi->value);
 363   /* FIXME: we do not have distance data? Assume direct neighbor. */
 364   return DIRECT_NEIGHBOR_COST;
 365 }
 366
 367
 368 /**
 369  * Function called to notify a client about the socket
 370  * begin ready to queue more data.  "buf" will be
 371  * NULL and "size" zero if the socket was closed for
 372  * writing in the meantime.
 373  *
 374  * @param cls closure
 375  * @param size number of bytes available in buf
 376  * @param buf where the callee should write the message
 377  * @return number of bytes written to buf
 378  */
 379 static size_t
 380 transmit_to_plugin (void *cls, size_t size, void *buf)
 381 {
 382   char *cbuf = buf;
 383   struct PendingMessage *reply;
 384   size_t off;
 385   size_t msize;
 386
 387   plugin_transmit_handle = NULL;
 388   if (NULL == buf)
 389   {
 390     /* client disconnected */
 391     return 0;
 392   }
 393   off = 0;
 394   while ( (NULL != (reply = plugin_pending_head)) &&
 395           (size >= off + (msize = ntohs (reply->msg->size))))
 396   {
 397     GNUNET_CONTAINER_DLL_remove (plugin_pending_head, plugin_pending_tail,
 398                                  reply);
 399     memcpy (&cbuf[off], reply->msg, msize);
 400     GNUNET_free (reply);
 401     off += msize;
 402   }
 403   if (NULL != plugin_pending_head)
 404     plugin_transmit_handle =
 405       GNUNET_SERVER_notify_transmit_ready (client_handle,
 406                                            msize,
 407                                            GNUNET_TIME_UNIT_FOREVER_REL,
 408                                            &transmit_to_plugin, NULL);
 409   return off;
 410 }
 411
 412
 413 /**
 414  * Forward a message from another peer to the plugin.
 415  *
 416  * @param message the message to send to the plugin
 417  * @param distant_neighbor the original sender of the message
 418  * @param distnace distance to the original sender of the message
 419  */
 420 static void
 421 send_data_to_plugin (const struct GNUNET_MessageHeader *message,
 422                      const struct GNUNET_PeerIdentity *distant_neighbor,
 423                      uint32_t distance)
 424 {
 425   struct GNUNET_DV_ReceivedMessage *received_msg;
 426   struct PendingMessage *pending_message;
 427   size_t size;
 428
 429   if (NULL == client_handle)
 430   {
 431     GNUNET_STATISTICS_update (stats,
 432                               "# messages discarded (no plugin)",
 433                               1, GNUNET_NO);
 434     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 435                 _("Refusing to queue messages, DV plugin not active.\n"));
 436     return;
 437   }
 438   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 439               "Delivering message from peer `%s'\n",
 440               GNUNET_i2s (distant_neighbor));
 441   size = sizeof (struct GNUNET_DV_ReceivedMessage) +
 442     ntohs (message->size);
 443   if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 444   {
 445     GNUNET_break (0); /* too big */
 446     return;
 447   }
 448   pending_message = GNUNET_malloc (sizeof (struct PendingMessage) + size);
 449   received_msg = (struct GNUNET_DV_ReceivedMessage *) &pending_message[1];
 450   received_msg->header.size = htons (size);
 451   received_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DV_RECV);
 452   received_msg->distance = htonl (distance);
 453   received_msg->sender = *distant_neighbor;
 454   memcpy (&received_msg[1], message, ntohs (message->size));
 455   GNUNET_CONTAINER_DLL_insert_tail (plugin_pending_head,
 456                                     plugin_pending_tail,
 457                                     pending_message);
 458   if (NULL == plugin_transmit_handle)
 459     plugin_transmit_handle =
 460       GNUNET_SERVER_notify_transmit_ready (client_handle, size,
 461                                            GNUNET_TIME_UNIT_FOREVER_REL,
 462                                            &transmit_to_plugin, NULL);
 463 }
 464
 465
 466 /**
 467  * Forward a control message to the plugin.
 468  *
 469  * @param message the message to send to the plugin
 470  * @param distant_neighbor the original sender of the message
 471  * @param distnace distance to the original sender of the message
 472  */
 473 static void
 474 send_control_to_plugin (const struct GNUNET_MessageHeader *message)
 475 {
 476   struct PendingMessage *pending_message;
 477   size_t size;
 478
 479   if (NULL == client_handle)
 480   {
 481     GNUNET_STATISTICS_update (stats,
 482                               "# control messages discarded (no plugin)",
 483                               1, GNUNET_NO);
 484     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 485                 _("Refusing to queue messages, DV plugin not active.\n"));
 486     return;
 487   }
 488   size = ntohs (message->size);
 489   pending_message = GNUNET_malloc (sizeof (struct PendingMessage) + size);
 490   memcpy (&pending_message[1], message, size);
 491   GNUNET_CONTAINER_DLL_insert_tail (plugin_pending_head,
 492                                     plugin_pending_tail,
 493                                     pending_message);
 494   if (NULL == plugin_transmit_handle)
 495     plugin_transmit_handle =
 496       GNUNET_SERVER_notify_transmit_ready (client_handle, size,
 497                                            GNUNET_TIME_UNIT_FOREVER_REL,
 498                                            &transmit_to_plugin, NULL);
 499 }
 500
 501
 502 /**
 503  * Give an ACK message to the plugin, we transmitted a message for it.
 504  *
 505  * @param target peer that received the message
 506  * @param uid plugin-chosen UID for the message
 507  */
 508 static void
 509 send_ack_to_plugin (const struct GNUNET_PeerIdentity *target,
 510                     uint32_t uid)
 511 {
 512   struct GNUNET_DV_AckMessage ack_msg;
 513
 514   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 515               "Delivering ACK for message to peer `%s'\n",
 516               GNUNET_i2s (target));
 517   ack_msg.header.size = htons (sizeof (ack_msg));
 518   ack_msg.header.type = htons (GNUNET_MESSAGE_TYPE_DV_SEND_ACK);
 519   ack_msg.uid = htonl (uid);
 520   ack_msg.target = *target;
 521   send_control_to_plugin (&ack_msg.header);
 522 }
 523
 524
 525 /**
 526  * Give a CONNECT message to the plugin.
 527  *
 528  * @param target peer that connected
 529  * @param distance distance to the target
 530  */
 531 static void
 532 send_connect_to_plugin (const struct GNUNET_PeerIdentity *target,
 533                         uint32_t distance)
 534 {
 535   struct GNUNET_DV_ConnectMessage cm;
 536
 537   if (NULL == client_handle)
 538     return;
 539   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 540               "Delivering CONNECT about peer `%s'\n",
 541               GNUNET_i2s (target));
 542   cm.header.size = htons (sizeof (cm));
 543   cm.header.type = htons (GNUNET_MESSAGE_TYPE_DV_CONNECT);
 544   cm.distance = htonl (distance);
 545   cm.peer = *target;
 546   send_control_to_plugin (&cm.header);
 547 }
 548
 549
 550 /**
 551  * Give a DISCONNECT message to the plugin.
 552  *
 553  * @param target peer that disconnected
 554  */
 555 static void
 556 send_disconnect_to_plugin (const struct GNUNET_PeerIdentity *target)
 557 {
 558   struct GNUNET_DV_DisconnectMessage dm;
 559
 560   if (NULL == client_handle)
 561     return;
 562   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 563               "Delivering DISCONNECT about peer `%s'\n",
 564               GNUNET_i2s (target));
 565   dm.header.size = htons (sizeof (dm));
 566   dm.header.type = htons (GNUNET_MESSAGE_TYPE_DV_DISCONNECT);
 567   dm.reserved = htonl (0);
 568   dm.peer = *target;
 569   send_control_to_plugin (&dm.header);
 570 }
 571
 572
 573 /**
 574  * Function called to transfer a message to another peer
 575  * via core.
 576  *
 577  * @param cls closure with the direct neighbor
 578  * @param size number of bytes available in buf
 579  * @param buf where the callee should write the message
 580  * @return number of bytes written to buf
 581  */
 582 static size_t
 583 core_transmit_notify (void *cls, size_t size, void *buf)
 584 {
 585   struct DirectNeighbor *dn = cls;
 586   char *cbuf = buf;
 587   struct PendingMessage *pending;
 588   size_t off;
 589   size_t msize;
 590
 591   dn->cth = NULL;
 592   if (NULL == buf)
 593   {
 594     /* peer disconnected */
 595     return 0;
 596   }
 597   off = 0;
 598   pending = dn->pm_head;
 599   off = 0;
 600   while ( (NULL != (pending = dn->pm_head)) &&
 601           (size >= off + (msize = ntohs (pending->msg->size))))
 602   {
 603     dn->pm_queue_size--;
 604     GNUNET_CONTAINER_DLL_remove (dn->pm_head,
 605                                  dn->pm_tail,
 606                                  pending);
 607     memcpy (&cbuf[off], pending->msg, msize);
 608     send_ack_to_plugin (&pending->ultimate_target,
 609                         pending->uid);
 610     GNUNET_free (pending);
 611     off += msize;
 612   }
 613   if (NULL != dn->pm_head)
 614     dn->cth =
 615       GNUNET_CORE_notify_transmit_ready (core_api,
 616                                          GNUNET_YES /* cork */,
 617                                          0 /* priority */,
 618                                          GNUNET_TIME_UNIT_FOREVER_REL,
 619                                          &dn->peer,
 620                                          msize,
 621                                          &core_transmit_notify, dn);
 622   return off;
 623 }
 624
 625
 626 /**
 627  * Forward the given payload to the given target.
 628  *
 629  * @param target where to send the message
 630  * @param distance expected (remaining) distance to the target
 631  * @param sender original sender of the message
 632  * @param payload payload of the message
 633  */
 634 static void
 635 forward_payload (struct DirectNeighbor *target,
 636                  uint32_t distance,
 637                  const struct GNUNET_PeerIdentity *sender,
 638                  const struct GNUNET_MessageHeader *payload)
 639 {
 640   struct PendingMessage *pm;
 641   struct RouteMessage *rm;
 642   size_t msize;
 643
 644   if ( (target->pm_queue_size >= MAX_QUEUE_SIZE) &&
 645        (0 != memcmp (sender,
 646                      &my_identity,
 647                      sizeof (struct GNUNET_PeerIdentity))) )
 648     return;
 649   msize = sizeof (struct RouteMessage) + ntohs (payload->size);
 650   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 651   {
 652     GNUNET_break (0);
 653     return;
 654   }
 655   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
 656   pm->msg = (const struct GNUNET_MessageHeader *) &pm[1];
 657   rm = (struct RouteMessage *) &pm[1];
 658   rm->header.size = htons ((uint16_t) msize);
 659   rm->header.type = htons (GNUNET_MESSAGE_TYPE_DV_ROUTE);
 660   rm->distance = htonl (distance);
 661   rm->target = target->peer;
 662   rm->sender = *sender;
 663   memcpy (&rm[1], payload, ntohs (payload->size));
 664   GNUNET_CONTAINER_DLL_insert_tail (target->pm_head,
 665                                     target->pm_tail,
 666                                     pm);
 667   target->pm_queue_size++;
 668   if (NULL == target->cth)
 669     target->cth = GNUNET_CORE_notify_transmit_ready (core_api,
 670                                                      GNUNET_YES /* cork */,
 671                                                      0 /* priority */,
 672                                                      GNUNET_TIME_UNIT_FOREVER_REL,
 673                                                      &target->peer,
 674                                                      msize,
 675                                                      &core_transmit_notify, target);
 676 }
 677
 678
 679 /**
 680  * Find a free slot for storing a 'route' in the 'consensi'
 681  * set at the given distance.
 682  *
 683  * @param distance distance to use for the set slot
 684  */
 685 static unsigned int
 686 get_consensus_slot (uint32_t distance)
 687 {
 688   struct ConsensusSet *cs;
 689   unsigned int i;
 690
 691   cs = &consensi[distance];
 692   i = 0;
 693   while ( (i < cs->array_length) &&
 694           (NULL != cs->targets[i]) ) i++;
 695   if (i == cs->array_length)
 696     GNUNET_array_grow (cs->targets,
 697                        cs->array_length,
 698                        cs->array_length * 2 + 2);
 699   return i;
 700 }
 701
 702
 703 /**
 704  * Allocate a slot in the consensus set for a route.
 705  *
 706  * @param route route to initialize
 707  * @param distance which consensus set to use
 708  */
 709 static void
 710 allocate_route (struct Route *route,
 711                 uint32_t distance)
 712 {
 713   unsigned int i;
 714
 715   i = get_consensus_slot (distance);
 716   route->set_offset = i;
 717   consensi[distance].targets[i] = route;
 718   route->target.distance = distance;
 719 }
 720
 721
 722 /**
 723  * Release a slot in the consensus set for a route.
 724  *
 725  * @param route route to release the slot from
 726  */
 727 static void
 728 release_route (struct Route *route)
 729 {
 730   consensi[route->target.distance].targets[route->set_offset] = NULL;
 731   route->set_offset = UINT_MAX; /* indicate invalid slot */
 732 }
 733
 734
 735 /**
 736  * Move a route from one consensus set to another.
 737  *
 738  * @param route route to move
 739  * @param new_distance new distance for the route (destination set)
 740  */
 741 static void
 742 move_route (struct Route *route,
 743             uint32_t new_distance)
 744 {
 745   unsigned int i;
 746
 747   release_route (route);
 748   i = get_consensus_slot (new_distance);
 749   route->set_offset = i;
 750   consensi[new_distance].targets[i] = route;
 751   route->target.distance = new_distance;
 752 }
 753
 754
 755 /**
 756  * Start creating a new consensus from scratch.
 757  *
 758  * @param cls the 'struct DirectNeighbor' of the peer we're building
 759  *        a routing consensus with
 760  * @param tc scheduler context
 761  */
 762 static void
 763 start_consensus (void *cls,
 764                  const struct GNUNET_SCHEDULER_TaskContext *tc);
 765
 766
 767 /**
 768  * Method called whenever a peer connects.
 769  *
 770  * @param cls closure
 771  * @param peer peer identity this notification is about
 772  * @param atsi performance data
 773  * @param atsi_count number of entries in atsi
 774  */
 775 static void
 776 handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *peer,
 777                      const struct GNUNET_ATS_Information *atsi,
 778                      unsigned int atsi_count)
 779 {
 780   struct DirectNeighbor *neighbor;
 781   struct Route *route;
 782   uint32_t distance;
 783
 784   /* Check for connect to self message */
 785   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
 786     return;
 787   distance = get_atsi_distance (atsi, atsi_count);
 788   neighbor = GNUNET_CONTAINER_multihashmap_get (direct_neighbors,
 789                                                 &peer->hashPubKey);
 790   if (NULL != neighbor)
 791   {
 792     GNUNET_break (0);
 793     return;
 794   }
 795   if (DIRECT_NEIGHBOR_COST != distance)
 796     return; /* is a DV-neighbor */
 797   neighbor = GNUNET_malloc (sizeof (struct DirectNeighbor));
 798   neighbor->peer = *peer;
 799   GNUNET_assert (GNUNET_YES ==
 800                  GNUNET_CONTAINER_multihashmap_put (direct_neighbors,
 801                                                     &peer->hashPubKey,
 802                                                     neighbor,
 803                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 804   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
 805                                              &peer->hashPubKey);
 806   if (NULL != route)
 807   {
 808     send_disconnect_to_plugin (peer);
 809     release_route (route);
 810     GNUNET_free (route);
 811   }
 812   route->next_hop = neighbor;
 813   neighbor->consensus_task = GNUNET_SCHEDULER_add_now (&start_consensus,
 814                                                        neighbor);
 815 }
 816
 817
 818 /**
 819  * We inserted the last element into the consensus, get ready to
 820  * insert the next element into the consensus or conclude if
 821  * we're done.
 822  *
 823  * @param cls the 'struct DirectNeighbor' of the peer we're building
 824  *        a routing consensus with
 825  * @param success GNUNET_OK if the last element was added successfully,
 826  *                GNUNET_SYSERR if we failed
 827  */
 828 static void
 829 insert_next_element (void *cls,
 830                      int success)
 831 {
 832   struct DirectNeighbor *neighbor = cls;
 833   struct GNUNET_CONSENSUS_Element element;
 834
 835   // FIXME: initialize element...
 836   GNUNET_CONSENSUS_insert (neighbor->consensus,
 837                            &element,
 838                            &insert_next_element,
 839                            neighbor);
 840 }
 841
 842
 843 /**
 844  * We have learned a new route from the other peer.  Add it to the
 845  * route set we're building.
 846  *
 847  * @param cls the 'struct DirectNeighbor' we're building the consensus with
 848  * @param element the new element we have learned
 849  * @return GNUNET_OK if the valid is well-formed and should be added to the consensus,
 850  *         GNUNET_SYSERR if the element should be ignored and not be propagated
 851  */
 852 static int
 853 learn_route_cb (void *cls,
 854                 const struct GNUNET_CONSENSUS_Element *element)
 855 {
 856   GNUNET_break (0); // FIXME
 857   return GNUNET_SYSERR;
 858 }
 859
 860
 861 /**
 862  * Start creating a new consensus from scratch.
 863  *
 864  * @param cls the 'struct DirectNeighbor' of the peer we're building
 865  *        a routing consensus with
 866  * @param tc scheduler context
 867  */
 868 static void
 869 start_consensus (void *cls,
 870                  const struct GNUNET_SCHEDULER_TaskContext *tc)
 871 {
 872   struct DirectNeighbor *neighbor = cls;
 873   struct GNUNET_HashCode session_id;
 874
 875   neighbor->consensus_task = GNUNET_SCHEDULER_NO_TASK;
 876   GNUNET_assert (NULL == neighbor->consensus);
 877   GNUNET_CRYPTO_hash_xor (&session_id, &my_identity.hashPubKey, &neighbor->peer.hashPubKey); // ARG order?
 878   neighbor->consensus = GNUNET_CONSENSUS_create (cfg,
 879                                                  1,
 880                                                  &neighbor->peer,
 881                                                  &session_id,
 882                                                  &learn_route_cb,
 883                                                  neighbor);
 884   if (NULL == neighbor->consensus)
 885   {
 886     neighbor->consensus_task = GNUNET_SCHEDULER_add_delayed (GNUNET_DV_CONSENSUS_FREQUENCY,
 887                                                              &start_consensus,
 888                                                              neighbor);
 889     return;
 890   }
 891   insert_next_element (neighbor, GNUNET_OK);
 892 }
 893
 894
 895 /**
 896  * Core handler for DV data messages.  Whatever this message
 897  * contains all we really have to do is rip it out of its
 898  * DV layering and give it to our pal the DV plugin to report
 899  * in with.
 900  *
 901  * @param cls closure
 902  * @param peer peer which sent the message (immediate sender)
 903  * @param message the message
 904  * @param atsi transport ATS information (latency, distance, etc.)
 905  * @param atsi_count number of entries in atsi
 906  * @return GNUNET_OK on success, GNUNET_SYSERR if the other peer violated the protocol
 907  */
 908 static int
 909 handle_dv_route_message (void *cls, const struct GNUNET_PeerIdentity *peer,
 910                          const struct GNUNET_MessageHeader *message,
 911                          const struct GNUNET_ATS_Information *atsi,
 912                          unsigned int atsi_count)
 913 {
 914   const struct RouteMessage *rm;
 915   const struct GNUNET_MessageHeader *payload;
 916   struct Route *route;
 917
 918   if (ntohs (message->size) < sizeof (struct RouteMessage) + sizeof (struct GNUNET_MessageHeader))
 919   {
 920     GNUNET_break_op (0);
 921     return GNUNET_SYSERR;
 922   }
 923   rm = (const struct RouteMessage *) message;
 924   payload = (const struct GNUNET_MessageHeader *) &rm[1];
 925   if (ntohs (message->size) != sizeof (struct RouteMessage) + ntohs (payload->size))
 926   {
 927     GNUNET_break_op (0);
 928     return GNUNET_SYSERR;
 929   }
 930   if (0 == memcmp (&rm->target,
 931                    &my_identity,
 932                    sizeof (struct GNUNET_PeerIdentity)))
 933   {
 934     /* message is for me, check reverse route! */
 935     route = GNUNET_CONTAINER_multihashmap_get (all_routes,
 936                                                &rm->sender.hashPubKey);
 937     if (NULL == route)
 938     {
 939       /* don't have reverse route, drop */
 940       GNUNET_STATISTICS_update (stats,
 941                                 "# message discarded (no reverse route)",
 942                                 1, GNUNET_NO);
 943       return GNUNET_OK;
 944     }
 945     send_data_to_plugin (payload,
 946                          &rm->sender,
 947                          route->target.distance);
 948     return GNUNET_OK;
 949   }
 950   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
 951                                              &rm->target.hashPubKey);
 952   if (NULL == route)
 953   {
 954     GNUNET_STATISTICS_update (stats,
 955                               "# messages discarded (no route)",
 956                               1, GNUNET_NO);
 957     return GNUNET_OK;
 958   }
 959   if (route->target.distance > ntohl (rm->distance) + 1)
 960   {
 961     GNUNET_STATISTICS_update (stats,
 962                               "# messages discarded (target too far)",
 963                               1, GNUNET_NO);
 964     return GNUNET_OK;
 965   }
 966   forward_payload (route->next_hop,
 967                    route->target.distance,
 968                    &rm->sender,
 969                    payload);
 970   return GNUNET_OK;
 971 }
 972
 973
 974 /**
 975  * Service server's handler for message send requests (which come
 976  * bubbling up to us through the DV plugin).
 977  *
 978  * @param cls closure
 979  * @param client identification of the client
 980  * @param message the actual message
 981  */
 982 static void
 983 handle_dv_send_message (void *cls, struct GNUNET_SERVER_Client *client,
 984                         const struct GNUNET_MessageHeader *message)
 985 {
 986   struct Route *route;
 987   const struct GNUNET_DV_SendMessage *msg;
 988   const struct GNUNET_MessageHeader *payload;
 989
 990   if (ntohs (message->size) < sizeof (struct GNUNET_DV_SendMessage) + sizeof (struct GNUNET_MessageHeader))
 991   {
 992     GNUNET_break (0);
 993     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 994     return;
 995   }
 996   msg = (const struct GNUNET_DV_SendMessage *) message;
 997   payload = (const struct GNUNET_MessageHeader *) &msg[1];
 998   if (ntohs (message->size) != sizeof (struct GNUNET_DV_SendMessage) + ntohs (payload->size))
 999   {
1000     GNUNET_break (0);
1001     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1002     return;
1003   }
1004   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1005                                              &msg->target.hashPubKey);
1006   if (NULL == route)
1007   {
1008     /* got disconnected, send ACK anyway?
1009        FIXME: What we really want is an 'NACK' here... */
1010     GNUNET_STATISTICS_update (stats,
1011                               "# local messages discarded (no route)",
1012                               1, GNUNET_NO);
1013     send_ack_to_plugin (&msg->target, htonl (msg->uid));
1014     GNUNET_SERVER_receive_done (client, GNUNET_OK);
1015     return;
1016   }
1017   // FIXME: flow control (send ACK only once message has left the queue...)
1018   send_ack_to_plugin (&msg->target, htonl (msg->uid));
1019   forward_payload (route->next_hop,
1020                    route->target.distance,
1021                    &my_identity,
1022                    payload);
1023   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1024 }
1025
1026
1027 /**
1028  * Multihashmap iterator for freeing routes that go via a particular
1029  * neighbor that disconnected and is thus no longer available.
1030  *
1031  * @param cls the direct neighbor that is now unavailable
1032  * @param key key value stored under
1033  * @param value a 'struct Route' that may or may not go via neighbor
1034  *
1035  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1036  */
1037 static int
1038 cull_routes (void *cls, const struct GNUNET_HashCode * key, void *value)
1039 {
1040   struct DirectNeighbor *neighbor = cls;
1041   struct Route *route = value;
1042
1043   if (route->next_hop != neighbor)
1044     return GNUNET_YES; /* not affected */
1045   GNUNET_assert (GNUNET_YES ==
1046                  GNUNET_CONTAINER_multihashmap_remove (all_routes, key, value));
1047   release_route (route);
1048   send_disconnect_to_plugin (&route->target.peer);
1049   GNUNET_free (route);
1050   return GNUNET_YES;
1051 }
1052
1053
1054 /**
1055  * Multihashmap iterator for checking if a given route is
1056  * (now) useful to this peer.
1057  *
1058  * @param cls the direct neighbor for the given route
1059  * @param key key value stored under
1060  * @param value a 'struct Target' that may or may not be useful; not that
1061  *        the distance in 'target' does not include the first hop yet
1062  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1063  */
1064 static int
1065 check_possible_route (void *cls, const struct GNUNET_HashCode * key, void *value)
1066 {
1067   struct DirectNeighbor *neighbor = cls;
1068   struct Target *target = value;
1069   struct Route *route;
1070
1071   route = GNUNET_CONTAINER_multihashmap_get (all_routes,
1072                                            key);
1073   if (NULL != route)
1074   {
1075     if (route->target.distance > target->distance + 1)
1076     {
1077       /* this 'target' is cheaper than the existing route; switch to alternative route! */
1078       move_route (route, target->distance + 1);
1079       route->next_hop = neighbor;
1080       // FIXME: notify plugin about distance update?
1081     }
1082     return GNUNET_YES; /* got a route to this target already */
1083   }
1084   route = GNUNET_malloc (sizeof (struct Route));
1085   route->next_hop = neighbor;
1086   route->target.distance = target->distance + 1;
1087   route->target.peer = target->peer;
1088   allocate_route (route, route->target.distance);
1089   GNUNET_assert (GNUNET_YES ==
1090                  GNUNET_CONTAINER_multihashmap_put (all_routes,
1091                                                     &route->target.peer.hashPubKey,
1092                                                     route,
1093                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1094   send_connect_to_plugin (&route->target.peer, target->distance);
1095   return GNUNET_YES;
1096 }
1097
1098
1099 /**
1100  * Multihashmap iterator for finding routes that were previously
1101  * "hidden" due to a better route (called after a disconnect event).
1102  *
1103  * @param cls NULL
1104  * @param key peer identity of the given direct neighbor
1105  * @param value a 'struct DirectNeighbor' to check for additional routes
1106  * @return GNUNET_YES to continue iteration
1107  */
1108 static int
1109 refresh_routes (void *cls, const struct GNUNET_HashCode * key, void *value)
1110 {
1111   struct DirectNeighbor *neighbor = value;
1112
1113   if (NULL != neighbor->neighbor_table)
1114     GNUNET_CONTAINER_multihashmap_iterate (neighbor->neighbor_table,
1115                                            &check_possible_route,
1116                                            neighbor);
1117   return GNUNET_YES;
1118 }
1119
1120
1121 /**
1122  * Cleanup all of the data structures associated with a given neighbor.
1123  *
1124  * @param neighbor neighbor to clean up
1125  */
1126 static void
1127 cleanup_neighbor (struct DirectNeighbor *neighbor)
1128 {
1129   struct PendingMessage *pending;
1130
1131   while (NULL != (pending = neighbor->pm_head))
1132   {
1133     neighbor->pm_queue_size--;
1134     GNUNET_CONTAINER_DLL_remove (neighbor->pm_head,
1135                                  neighbor->pm_tail,
1136                                  pending);
1137     GNUNET_free (pending);
1138   }
1139   GNUNET_CONTAINER_multihashmap_iterate (all_routes,
1140                                          &cull_routes,
1141                                          neighbor);
1142   if (NULL != neighbor->cth)
1143   {
1144     GNUNET_CORE_notify_transmit_ready_cancel (neighbor->cth);
1145     neighbor->cth = NULL;
1146   }
1147   if (GNUNET_SCHEDULER_NO_TASK != neighbor->consensus_task)
1148   {
1149     GNUNET_SCHEDULER_cancel (neighbor->consensus_task);
1150     neighbor->consensus_task = GNUNET_SCHEDULER_NO_TASK;
1151   }
1152   if (NULL != neighbor->consensus)
1153   {
1154     GNUNET_CONSENSUS_destroy (neighbor->consensus);
1155     neighbor->consensus = NULL;
1156   }
1157   GNUNET_assert (GNUNET_YES ==
1158                  GNUNET_CONTAINER_multihashmap_remove (direct_neighbors,
1159                                                        &neighbor->peer.hashPubKey,
1160                                                        neighbor));
1161   GNUNET_free (neighbor);
1162 }
1163
1164
1165 /**
1166  * Method called whenever a given peer disconnects.
1167  *
1168  * @param cls closure
1169  * @param peer peer identity this notification is about
1170  */
1171 static void
1172 handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
1173 {
1174   struct DirectNeighbor *neighbor;
1175
1176   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1177               "Received core peer disconnect message for peer `%s'!\n",
1178               GNUNET_i2s (peer));
1179   /* Check for disconnect from self message */
1180   if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
1181     return;
1182   neighbor =
1183       GNUNET_CONTAINER_multihashmap_get (direct_neighbors, &peer->hashPubKey);
1184   if (NULL == neighbor)
1185   {
1186     /* must have been a DV-neighbor, ignore */
1187     return;
1188   }
1189   cleanup_neighbor (neighbor);
1190   GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1191                                          &refresh_routes,
1192                                          NULL);
1193 }
1194
1195
1196 /**
1197  * Multihashmap iterator for freeing routes.  Should never be called.
1198  *
1199  * @param cls NULL
1200  * @param key key value stored under
1201  * @param value the route to be freed
1202  *
1203  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1204  */
1205 static int
1206 free_route (void *cls, const struct GNUNET_HashCode * key, void *value)
1207 {
1208   struct Route *route = value;
1209
1210   GNUNET_break (0);
1211   GNUNET_assert (GNUNET_YES ==
1212                  GNUNET_CONTAINER_multihashmap_remove (all_routes, key, value));
1213   release_route (route);
1214   send_disconnect_to_plugin (&route->target.peer);
1215   GNUNET_free (route);
1216   return GNUNET_YES;
1217 }
1218
1219
1220 /**
1221  * Multihashmap iterator for freeing direct neighbors. Should never be called.
1222  *
1223  * @param cls NULL
1224  * @param key key value stored under
1225  * @param value the direct neighbor to be freed
1226  *
1227  * @return GNUNET_YES to continue iteration, GNUNET_NO to stop
1228  */
1229 static int
1230 free_direct_neighbors (void *cls, const struct GNUNET_HashCode * key, void *value)
1231 {
1232   struct DirectNeighbor *neighbor = value;
1233
1234   GNUNET_break (0);
1235   cleanup_neighbor (neighbor);
1236   return GNUNET_YES;
1237 }
1238
1239
1240 /**
1241  * Task run during shutdown.
1242  *
1243  * @param cls unused
1244  * @param tc unused
1245  */
1246 static void
1247 shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1248 {
1249   struct PendingMessage *pending;
1250   unsigned int i;
1251
1252   GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors,
1253                                          &free_direct_neighbors, NULL);
1254   GNUNET_CONTAINER_multihashmap_destroy (direct_neighbors);
1255   GNUNET_CONTAINER_multihashmap_iterate (all_routes,
1256                                          &free_route, NULL);
1257   GNUNET_CONTAINER_multihashmap_destroy (all_routes);
1258   GNUNET_CORE_disconnect (core_api);
1259   core_api = NULL;
1260   GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
1261   stats = NULL;
1262   while (NULL != (pending = plugin_pending_head))
1263   {
1264     GNUNET_CONTAINER_DLL_remove (plugin_pending_head,
1265                                  plugin_pending_tail,
1266                                  pending);
1267     GNUNET_free (pending);
1268   }
1269   for (i=0;i<DEFAULT_FISHEYE_DEPTH - 1;i++)
1270     GNUNET_array_grow (consensi[i].targets,
1271                        consensi[i].array_length,
1272                        0);
1273 }
1274
1275
1276 /**
1277  * Handle START-message.  This is the first message sent to us
1278  * by the client (can only be one!).
1279  *
1280  * @param cls closure (always NULL)
1281  * @param client identification of the client
1282  * @param message the actual message
1283  */
1284 static void
1285 handle_start (void *cls, struct GNUNET_SERVER_Client *client,
1286               const struct GNUNET_MessageHeader *message)
1287 {
1288   if (NULL != client_handle)
1289   {
1290     /* forcefully drop old client */
1291     GNUNET_SERVER_client_disconnect (client_handle);
1292     GNUNET_SERVER_client_drop (client_handle);
1293   }
1294   client_handle = client;
1295   GNUNET_SERVER_client_keep (client_handle);
1296   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1297 }
1298
1299
1300 /**
1301  * Called on core init.
1302  *
1303  * @param cls unused
1304  * @param server legacy
1305  * @param identity this peer's identity
1306  */
1307 static void
1308 core_init (void *cls, struct GNUNET_CORE_Handle *server,
1309            const struct GNUNET_PeerIdentity *identity)
1310 {
1311   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1312               "I am peer: %s\n",
1313               GNUNET_i2s (identity));
1314   my_identity = *identity;
1315 }
1316
1317
1318 /**
1319  * Process dv requests.
1320  *
1321  * @param cls closure
1322  * @param server the initialized server
1323  * @param c configuration to use
1324  */
1325 static void
1326 run (void *cls, struct GNUNET_SERVER_Handle *server,
1327      const struct GNUNET_CONFIGURATION_Handle *c)
1328 {
1329   static struct GNUNET_CORE_MessageHandler core_handlers[] = {
1330     {&handle_dv_route_message, GNUNET_MESSAGE_TYPE_DV_ROUTE, 0},
1331     {NULL, 0, 0}
1332   };
1333   static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = {
1334     {&handle_start, NULL,
1335      GNUNET_MESSAGE_TYPE_DV_START,
1336      sizeof (struct GNUNET_MessageHeader) },
1337     { &handle_dv_send_message, NULL,
1338       GNUNET_MESSAGE_TYPE_DV_SEND,
1339       0},
1340     {NULL, NULL, 0, 0}
1341   };
1342
1343   cfg = c;
1344   direct_neighbors = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_NO);
1345   all_routes = GNUNET_CONTAINER_multihashmap_create (65536, GNUNET_NO);
1346   core_api = GNUNET_CORE_connect (cfg, NULL,
1347                                   &core_init,
1348                                   &handle_core_connect,
1349                                   &handle_core_disconnect,
1350                                   NULL, GNUNET_NO,
1351                                   NULL, GNUNET_NO,
1352                                   core_handlers);
1353
1354   if (NULL == core_api)
1355     return;
1356   stats = GNUNET_STATISTICS_create ("dv", cfg);
1357   GNUNET_SERVER_add_handlers (server, plugin_handlers);
1358   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1359                                 &shutdown_task, NULL);
1360 }
1361
1362
1363 /**
1364  * The main function for the dv service.
1365  *
1366  * @param argc number of arguments from the command line
1367  * @param argv command line arguments
1368  * @return 0 ok, 1 on error
1369  */
1370 int
1371 main (int argc, char *const *argv)
1372 {
1373   return (GNUNET_OK ==
1374           GNUNET_SERVICE_run (argc, argv, "dv", GNUNET_SERVICE_OPTION_NONE,
1375                               &run, NULL)) ? 0 : 1;
1376 }
1377
1378 /* end of gnunet-service-dv.c */