src/dht/dht_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010 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 dht/dht_api.c
  23  * @brief library to access the DHT service
  24  * @author Christian Grothoff
  25  * @author Nathan Evans
  26  */
  27
  28 #include "platform.h"
  29 #include "gnunet_bandwidth_lib.h"
  30 #include "gnunet_client_lib.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_container_lib.h"
  33 #include "gnunet_arm_service.h"
  34 #include "gnunet_hello_lib.h"
  35 #include "gnunet_protocols.h"
  36 #include "gnunet_server_lib.h"
  37 #include "gnunet_time_lib.h"
  38 #include "gnunet_dht_service.h"
  39 #include "dht.h"
  40
  41 #define DEBUG_DHT_API GNUNET_NO
  42
  43 /**
  44  * Entry in our list of messages to be (re-)transmitted.
  45  */
  46 struct PendingMessage
  47 {
  48   /**
  49    * This is a doubly-linked list.
  50    */
  51   struct PendingMessage *prev;
  52
  53   /**
  54    * This is a doubly-linked list.
  55    */
  56   struct PendingMessage *next;
  57
  58   /**
  59    * Message that is pending, allocated at the end
  60    * of this struct.
  61    */
  62   const struct GNUNET_MessageHeader *msg;
  63
  64   /**
  65    * Handle to the DHT API context.
  66    */
  67   struct GNUNET_DHT_Handle *handle;
  68
  69   /**
  70    * Continuation to call when the request has been
  71    * transmitted (for the first time) to the service; can be NULL.
  72    */
  73   GNUNET_SCHEDULER_Task cont;
  74
  75   /**
  76    * Closure for 'cont'.
  77    */
  78   void *cont_cls;
  79
  80   /**
  81    * Timeout task for this message
  82    */
  83   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
  84
  85   /**
  86    * Unique ID for this request
  87    */
  88   uint64_t unique_id;
  89
  90   /**
  91    * Free the saved message once sent, set to GNUNET_YES for messages
  92    * that do not receive responses; GNUNET_NO if this pending message
  93    * is aliased from a 'struct GNUNET_DHT_RouteHandle' and will be freed
  94    * from there.
  95    */
  96   int free_on_send;
  97
  98   /**
  99    * GNUNET_YES if this message is in our pending queue right now.
 100    */
 101   int in_pending_queue;
 102
 103 };
 104
 105
 106 /**
 107  * Handle to a route request
 108  */
 109 struct GNUNET_DHT_RouteHandle
 110 {
 111
 112   /**
 113    * Iterator to call on data receipt
 114    */
 115   GNUNET_DHT_ReplyProcessor iter;
 116
 117   /**
 118    * Closure for the iterator callback
 119    */
 120   void *iter_cls;
 121
 122   /**
 123    * Main handle to this DHT api
 124    */
 125   struct GNUNET_DHT_Handle *dht_handle;
 126
 127   /**
 128    * The actual message sent for this request,
 129    * used for retransmitting requests on service
 130    * failure/reconnect.  Freed on route_stop.
 131    */
 132   struct PendingMessage *message;
 133
 134   /**
 135    * Key that this get request is for
 136    */
 137   GNUNET_HashCode key;
 138
 139   /**
 140    * Unique identifier for this request (for key collisions). FIXME: redundant!?
 141    */
 142   uint64_t uid;
 143
 144 };
 145
 146
 147 /**
 148  * Connection to the DHT service.
 149  */
 150 struct GNUNET_DHT_Handle
 151 {
 152
 153   /**
 154    * Configuration to use.
 155    */
 156   const struct GNUNET_CONFIGURATION_Handle *cfg;
 157
 158   /**
 159    * Socket (if available).
 160    */
 161   struct GNUNET_CLIENT_Connection *client;
 162
 163   /**
 164    * Currently pending transmission request (or NULL).
 165    */
 166   struct GNUNET_CLIENT_TransmitHandle *th;
 167
 168   /**
 169    * Head of linked list of messages we would like to transmit.
 170    */
 171   struct PendingMessage *pending_head;
 172
 173   /**
 174    * Tail of linked list of messages we would like to transmit.
 175    */
 176   struct PendingMessage *pending_tail;
 177
 178   /**
 179    * Hash map containing the current outstanding unique requests
 180    * (values are of type 'struct GNUNET_DHT_RouteHandle').
 181    */
 182   struct GNUNET_CONTAINER_MultiHashMap *active_requests;
 183
 184   /**
 185    * Task for trying to reconnect.
 186    */
 187   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 188
 189   /**
 190    * How quickly should we retry?  Used for exponential back-off on
 191    * connect-errors.
 192    */
 193   struct GNUNET_TIME_Relative retry_time;
 194
 195   /**
 196    * Generator for unique ids.
 197    */
 198   uint64_t uid_gen;
 199
 200 };
 201
 202
 203 /**
 204  * Transmit the next pending message, called by notify_transmit_ready
 205  */
 206 static size_t
 207 transmit_pending (void *cls,
 208                   size_t size,
 209                   void *buf);
 210
 211
 212 /**
 213  * Handler for messages received from the DHT service
 214  * a demultiplexer which handles numerous message types
 215  *
 216  */
 217 static void
 218 service_message_handler (void *cls,
 219                          const struct GNUNET_MessageHeader *msg);
 220
 221
 222
 223
 224 /**
 225  * Try to (re)connect to the DHT service.
 226  *
 227  * @return GNUNET_YES on success, GNUNET_NO on failure.
 228  */
 229 static int
 230 try_connect (struct GNUNET_DHT_Handle *handle)
 231 {
 232   if (handle->client != NULL)
 233     return GNUNET_OK;
 234   handle->client = GNUNET_CLIENT_connect ("dht", handle->cfg);
 235   if (handle->client == NULL)
 236     {
 237       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 238                   _("Failed to connect to the DHT service!\n"));
 239       return GNUNET_NO;
 240     }
 241   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 242               "Starting to process replies from DHT\n");
 243   GNUNET_CLIENT_receive (handle->client,
 244                          &service_message_handler,
 245                          handle,
 246                          GNUNET_TIME_UNIT_FOREVER_REL);
 247   return GNUNET_YES;
 248 }
 249
 250
 251 /**
 252  * Add the request corresponding to the given route handle
 253  * to the pending queue (if it is not already in there).
 254  *
 255  * @param cls the 'struct GNUNET_DHT_Handle*'
 256  * @param key key for the request (not used)
 257  * @param value the 'struct GNUNET_DHT_RouteHandle*'
 258  * @return GNUNET_YES (always)
 259  */
 260 static int
 261 add_request_to_pending (void *cls,
 262                         const GNUNET_HashCode *key,
 263                         void *value)
 264 {
 265   struct GNUNET_DHT_Handle *handle = cls;
 266   struct GNUNET_DHT_RouteHandle *rh = value;
 267
 268   if (GNUNET_NO == rh->message->in_pending_queue)
 269     {
 270       GNUNET_CONTAINER_DLL_insert (handle->pending_head,
 271                                    handle->pending_tail,
 272                                    rh->message);
 273       rh->message->in_pending_queue = GNUNET_YES;
 274     }
 275   return GNUNET_YES;
 276 }
 277
 278
 279 /**
 280  * Try reconnecting to the dht service.
 281  *
 282  * @param cls GNUNET_DHT_Handle
 283  * @param tc scheduler context
 284  */
 285 static void
 286 try_reconnect (void *cls,
 287                const struct GNUNET_SCHEDULER_TaskContext *tc)
 288 {
 289   struct GNUNET_DHT_Handle *handle = cls;
 290
 291   if (handle->retry_time.rel_value < GNUNET_CONSTANTS_SERVICE_RETRY.rel_value)
 292     handle->retry_time = GNUNET_CONSTANTS_SERVICE_RETRY;
 293   else
 294     handle->retry_time = GNUNET_TIME_relative_multiply (handle->retry_time, 2);
 295   if (handle->retry_time.rel_value > GNUNET_CONSTANTS_SERVICE_TIMEOUT.rel_value)
 296     handle->retry_time = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
 297   handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 298   handle->client = GNUNET_CLIENT_connect ("dht", handle->cfg);
 299   if (handle->client == NULL)
 300     {
 301       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 302                   "dht reconnect failed(!)\n");
 303       return;
 304     }
 305
 306   GNUNET_CONTAINER_multihashmap_iterate (handle->active_requests,
 307                                          &add_request_to_pending,
 308                                          handle);
 309   if (handle->pending_head == NULL)
 310     return;
 311
 312   GNUNET_CLIENT_notify_transmit_ready (handle->client,
 313                                        ntohs(handle->pending_head->msg->size),
 314                                        GNUNET_TIME_UNIT_FOREVER_REL,
 315                                        GNUNET_NO,
 316                                        &transmit_pending,
 317                                        handle);
 318 }
 319
 320
 321 /**
 322  * Try reconnecting to the DHT service.
 323  *
 324  * @param handle handle to dht to (possibly) disconnect and reconnect
 325  */
 326 static void
 327 do_disconnect (struct GNUNET_DHT_Handle *handle)
 328 {
 329   if (handle->client == NULL)
 330     return;
 331
 332   GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
 333   handle->client = NULL;
 334   handle->reconnect_task = GNUNET_SCHEDULER_add_delayed (handle->retry_time,
 335                                                          &try_reconnect,
 336                                                          handle);
 337 }
 338
 339
 340 /**
 341  * Try to send messages from list of messages to send
 342  */
 343 static void
 344 process_pending_messages (struct GNUNET_DHT_Handle *handle)
 345 {
 346   struct PendingMessage *head;
 347
 348   if (handle->client == NULL)
 349     {
 350       do_disconnect(handle);
 351       return;
 352     }
 353   if (handle->th != NULL)
 354     return;
 355   if (NULL == (head = handle->pending_head))
 356     return;
 357   handle->th = GNUNET_CLIENT_notify_transmit_ready (handle->client,
 358                                                     ntohs (head->msg->size),
 359                                                     GNUNET_TIME_UNIT_FOREVER_REL,
 360                                                     GNUNET_YES,
 361                                                     &transmit_pending,
 362                                                     handle);
 363   if (NULL == handle->th)
 364     {
 365       do_disconnect (handle);
 366       return;
 367     }
 368 }
 369
 370
 371 /**
 372  * Transmit the next pending message, called by notify_transmit_ready
 373  */
 374 static size_t
 375 transmit_pending (void *cls,
 376                   size_t size,
 377                   void *buf)
 378 {
 379   struct GNUNET_DHT_Handle *handle = cls;
 380   struct PendingMessage *head;
 381   size_t tsize;
 382
 383   handle->th = NULL;
 384   if (buf == NULL)
 385     {
 386       GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Transmission to DHT service failed!  Reconnecting!\n");
 387       do_disconnect (handle);
 388       return 0;
 389     }
 390   if (NULL == (head = handle->pending_head))
 391     return 0;
 392
 393   tsize = ntohs (head->msg->size);
 394   if (size < tsize)
 395     {
 396       process_pending_messages (handle);
 397       return 0;
 398     }
 399   memcpy (buf, head->msg, tsize);
 400   GNUNET_CONTAINER_DLL_remove (handle->pending_head,
 401                                handle->pending_tail,
 402                                head);
 403   if (head->timeout_task != GNUNET_SCHEDULER_NO_TASK)
 404     {
 405       GNUNET_SCHEDULER_cancel (head->timeout_task);
 406       head->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 407     }
 408   if (NULL != head->cont)
 409     {
 410       GNUNET_SCHEDULER_add_continuation (head->cont,
 411                                          head->cont_cls,
 412                                          GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 413       head->cont = NULL;
 414       head->cont_cls = NULL;
 415     }
 416   head->in_pending_queue = GNUNET_NO;
 417   if (GNUNET_YES == head->free_on_send)
 418     GNUNET_free (head);
 419   process_pending_messages (handle);
 420   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 421               "Forwarded request of %u bytes to DHT service\n",
 422               (unsigned int) tsize);
 423   return tsize;
 424 }
 425
 426
 427 /**
 428  * Process a given reply that might match the given
 429  * request.
 430  */
 431 static int
 432 process_reply (void *cls,
 433                const GNUNET_HashCode *key,
 434                void *value)
 435 {
 436   const struct GNUNET_DHT_RouteResultMessage *dht_msg = cls;
 437   struct GNUNET_DHT_RouteHandle *rh = value;
 438   const struct GNUNET_MessageHeader *enc_msg;
 439   size_t enc_size;
 440   uint64_t uid;
 441   const struct GNUNET_PeerIdentity **outgoing_path;
 442   const struct GNUNET_PeerIdentity *pos;
 443   uint32_t outgoing_path_length;
 444   unsigned int i;
 445   char *path_offset;
 446
 447   uid = GNUNET_ntohll (dht_msg->unique_id);
 448   if (uid != rh->uid)
 449     {
 450       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 451                   "Reply UID did not match request UID\n");
 452       return GNUNET_YES;
 453     }
 454   enc_msg = (const struct GNUNET_MessageHeader *)&dht_msg[1];
 455   enc_size = ntohs (enc_msg->size);
 456   if (enc_size < sizeof (struct GNUNET_MessageHeader))
 457     {
 458       GNUNET_break (0);
 459       return GNUNET_NO;
 460     }
 461   path_offset = (char *)&dht_msg[1];
 462   path_offset += enc_size;
 463   pos = (const struct GNUNET_PeerIdentity *) path_offset;
 464   outgoing_path_length = ntohl (dht_msg->outgoing_path_length);
 465   if (outgoing_path_length * sizeof (struct GNUNET_PeerIdentity) > ntohs(dht_msg->header.size) - enc_size)
 466     {
 467       GNUNET_break (0);
 468       return GNUNET_NO;
 469     }
 470
 471   if (outgoing_path_length > 0)
 472     {
 473       outgoing_path = GNUNET_malloc ((outgoing_path_length + 1) * sizeof (struct GNUNET_PeerIdentity*));
 474       for (i = 0; i < outgoing_path_length; i++)
 475         {
 476           outgoing_path[i] = pos;
 477           pos++;
 478         }
 479       outgoing_path[outgoing_path_length] = NULL;
 480     }
 481   else
 482     outgoing_path = NULL;
 483
 484   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 485               "Processing reply.\n");
 486   rh->iter (rh->iter_cls,
 487             &rh->key,
 488             outgoing_path,
 489             enc_msg);
 490   GNUNET_free_non_null (outgoing_path);
 491   return GNUNET_YES;
 492 }
 493
 494
 495 /**
 496  * Handler for messages received from the DHT service
 497  * a demultiplexer which handles numerous message types
 498  *
 499  * @param cls the 'struct GNUNET_DHT_Handle'
 500  * @param msg the incoming message
 501  */
 502 static void
 503 service_message_handler (void *cls,
 504                          const struct GNUNET_MessageHeader *msg)
 505 {
 506   struct GNUNET_DHT_Handle *handle = cls;
 507   const struct GNUNET_DHT_RouteResultMessage *dht_msg;
 508
 509   if (msg == NULL)
 510     {
 511       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 512                   "Error receiving data from DHT service, reconnecting\n");
 513       do_disconnect (handle);
 514       return;
 515     }
 516   if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_RESULT)
 517     {
 518       GNUNET_break (0);
 519       do_disconnect (handle);
 520       return;
 521     }
 522   if (ntohs (msg->size) < sizeof (struct GNUNET_DHT_RouteResultMessage))
 523     {
 524       GNUNET_break (0);
 525       do_disconnect (handle);
 526       return;
 527     }
 528   dht_msg = (const struct GNUNET_DHT_RouteResultMessage *) msg;
 529   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 530               "Comparing reply `%s' against %u pending requests.\n",
 531               GNUNET_h2s (&dht_msg->key),
 532               GNUNET_CONTAINER_multihashmap_size (handle->active_requests));
 533   GNUNET_CONTAINER_multihashmap_get_multiple (handle->active_requests,
 534                                               &dht_msg->key,
 535                                               &process_reply,
 536                                               (void*) dht_msg);
 537   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 538               "Continuing to process replies from DHT\n");
 539   GNUNET_CLIENT_receive (handle->client,
 540                          &service_message_handler,
 541                          handle, GNUNET_TIME_UNIT_FOREVER_REL);
 542
 543 }
 544
 545
 546 /**
 547  * Initialize the connection with the DHT service.
 548  *
 549  * @param cfg configuration to use
 550  * @param ht_len size of the internal hash table to use for
 551  *               processing multiple GET/FIND requests in parallel
 552  *
 553  * @return handle to the DHT service, or NULL on error
 554  */
 555 struct GNUNET_DHT_Handle *
 556 GNUNET_DHT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 557                     unsigned int ht_len)
 558 {
 559   struct GNUNET_DHT_Handle *handle;
 560
 561   handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_Handle));
 562   handle->cfg = cfg;
 563   handle->uid_gen = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
 564   handle->active_requests = GNUNET_CONTAINER_multihashmap_create (ht_len);
 565   if (GNUNET_NO == try_connect (handle))
 566     {
 567       GNUNET_DHT_disconnect (handle);
 568       return NULL;
 569     }
 570   return handle;
 571 }
 572
 573
 574 /**
 575  * Shutdown connection with the DHT service.
 576  *
 577  * @param handle handle of the DHT connection to stop
 578  */
 579 void
 580 GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle)
 581 {
 582   struct PendingMessage *pm;
 583   GNUNET_assert(handle != NULL);
 584   GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size(handle->active_requests));
 585   if (handle->th != NULL)
 586     {
 587       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
 588       handle->th = NULL;
 589     }
 590   while (NULL != (pm = handle->pending_head))
 591     {
 592       GNUNET_CONTAINER_DLL_remove (handle->pending_head,
 593                                    handle->pending_tail,
 594                                    pm);
 595       GNUNET_assert (GNUNET_YES == pm->free_on_send);
 596       if (GNUNET_SCHEDULER_NO_TASK != pm->timeout_task)
 597         GNUNET_SCHEDULER_cancel (pm->timeout_task);
 598       if (NULL != pm->cont)
 599         GNUNET_SCHEDULER_add_continuation (pm->cont,
 600                                            pm->cont_cls,
 601                                            GNUNET_SCHEDULER_REASON_TIMEOUT);
 602       pm->in_pending_queue = GNUNET_NO;
 603       GNUNET_free (pm);
 604     }
 605   if (handle->client != NULL)
 606     {
 607       GNUNET_CLIENT_disconnect (handle->client, GNUNET_YES);
 608       handle->client = NULL;
 609     }
 610   GNUNET_CONTAINER_multihashmap_destroy(handle->active_requests);
 611   GNUNET_free (handle);
 612 }
 613
 614
 615
 616
 617 /* ***** Special low-level API providing generic routing abstraction ***** */
 618
 619
 620 /**
 621  * Timeout for the transmission of a fire&forget-request.  Clean it up.
 622  *
 623  * @param cls the 'struct PendingMessage'
 624  * @param tc scheduler context
 625  */
 626 static void
 627 timeout_route_request (void *cls,
 628                        const struct GNUNET_SCHEDULER_TaskContext *tc)
 629 {
 630   struct PendingMessage *pending = cls;
 631   struct GNUNET_DHT_Handle *handle;
 632
 633   if (pending->free_on_send != GNUNET_YES)
 634     {
 635       /* timeouts should only apply to fire & forget requests! */
 636       GNUNET_break (0);
 637       return;
 638     }
 639   handle = pending->handle;
 640   GNUNET_CONTAINER_DLL_remove (handle->pending_head,
 641                                handle->pending_tail,
 642                                pending);
 643   if (pending->cont != NULL)
 644     pending->cont (pending->cont_cls,
 645                    tc);
 646   GNUNET_free (pending);
 647 }
 648
 649
 650 /**
 651  * Initiate a generic DHT route operation.
 652  *
 653  * @param handle handle to the DHT service
 654  * @param key the key to look up
 655  * @param desired_replication_level how many peers should ultimately receive
 656  *                this message (advisory only, target may be too high for the
 657  *                given DHT or not hit exactly).
 658  * @param options options for routing
 659  * @param enc send the encapsulated message to a peer close to the key
 660  * @param iter function to call on each result, NULL if no replies are expected
 661  * @param iter_cls closure for iter
 662  * @param timeout when to abort with an error if we fail to get
 663  *                a confirmation for the request (when necessary) or how long
 664  *                to wait for tramission to the service; only applies
 665  *                if 'iter' is NULL
 666  * @param cont continuation to call when the request has been transmitted
 667  *             the first time to the service
 668  * @param cont_cls closure for cont
 669  * @return handle to stop the request, NULL if the request is "fire and forget"
 670  */
 671 struct GNUNET_DHT_RouteHandle *
 672 GNUNET_DHT_route_start (struct GNUNET_DHT_Handle *handle,
 673                         const GNUNET_HashCode *key,
 674                         uint32_t desired_replication_level,
 675                         enum GNUNET_DHT_RouteOption options,
 676                         const struct GNUNET_MessageHeader *enc,
 677                         struct GNUNET_TIME_Relative timeout,
 678                         GNUNET_DHT_ReplyProcessor iter,
 679                         void *iter_cls,
 680                         GNUNET_SCHEDULER_Task cont,
 681                         void *cont_cls)
 682 {
 683   struct PendingMessage *pending;
 684   struct GNUNET_DHT_RouteMessage *message;
 685   struct GNUNET_DHT_RouteHandle *route_handle;
 686   uint16_t msize;
 687   uint16_t esize;
 688
 689   esize = ntohs (enc->size);
 690   if (sizeof (struct GNUNET_DHT_RouteMessage) + esize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 691     {
 692       GNUNET_break (0);
 693       return NULL;
 694     }
 695   msize = sizeof (struct GNUNET_DHT_RouteMessage) + esize;
 696   pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
 697   message = (struct GNUNET_DHT_RouteMessage*) &pending[1];
 698   pending->msg = &message->header;
 699   pending->handle = handle;
 700   pending->cont = cont;
 701   pending->cont_cls = cont_cls;
 702
 703   message->header.size = htons (msize);
 704   message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE);
 705   message->key = *key;
 706   message->options = htonl ((uint32_t) options);
 707   message->desired_replication_level = htonl (desired_replication_level);
 708   handle->uid_gen++;
 709   message->unique_id = GNUNET_htonll (handle->uid_gen);
 710   memcpy (&message[1], enc, esize);
 711
 712   if (iter != NULL)
 713     {
 714       route_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_RouteHandle));
 715       route_handle->key = *key;
 716       route_handle->iter = iter;
 717       route_handle->iter_cls = iter_cls;
 718       route_handle->dht_handle = handle;
 719       route_handle->uid = handle->uid_gen;
 720       route_handle->message = pending;
 721       GNUNET_CONTAINER_multihashmap_put (handle->active_requests,
 722                                          key,
 723                                          route_handle,
 724                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 725     }
 726   else
 727     {
 728       route_handle = NULL;
 729       pending->free_on_send = GNUNET_YES;
 730       pending->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout,
 731                                                             &timeout_route_request,
 732                                                             pending);
 733     }
 734   GNUNET_CONTAINER_DLL_insert (handle->pending_head,
 735                                handle->pending_tail,
 736                                pending);
 737   pending->in_pending_queue = GNUNET_YES;
 738   process_pending_messages (handle);
 739   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 740               "DHT route start request processed, returning %p\n",
 741               route_handle);
 742   return route_handle;
 743 }
 744
 745
 746 /**
 747  * Stop a previously issued routing request
 748  *
 749  * @param route_handle handle to the request to stop
 750  */
 751 void
 752 GNUNET_DHT_route_stop (struct GNUNET_DHT_RouteHandle *route_handle)
 753 {
 754   struct GNUNET_DHT_Handle *handle;
 755   struct PendingMessage *pending;
 756   struct GNUNET_DHT_StopMessage *message;
 757   size_t msize;
 758
 759   handle = route_handle->dht_handle;
 760   if (GNUNET_NO == route_handle->message->in_pending_queue)
 761     {
 762       /* need to send stop message */
 763       msize = sizeof (struct GNUNET_DHT_StopMessage);
 764       pending = GNUNET_malloc (sizeof (struct PendingMessage) +
 765                                msize);
 766       message = (struct GNUNET_DHT_StopMessage*) &pending[1];
 767       pending->msg = &message->header;
 768       message->header.size = htons (msize);
 769       message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_STOP);
 770       message->unique_id = GNUNET_htonll (route_handle->uid);
 771       message->key = route_handle->key;
 772       pending->handle = handle;
 773       pending->free_on_send = GNUNET_YES;
 774       pending->in_pending_queue = GNUNET_YES;
 775       GNUNET_CONTAINER_DLL_insert (handle->pending_head,
 776                                    handle->pending_tail,
 777                                    pending);
 778       process_pending_messages (handle);
 779     }
 780   else
 781     {
 782       /* simply remove pending request from message queue before
 783          transmission, no need to transmit STOP request! */
 784       GNUNET_CONTAINER_DLL_remove (handle->pending_head,
 785                                    handle->pending_tail,
 786                                    route_handle->message);
 787     }
 788   GNUNET_assert (GNUNET_YES ==
 789                  GNUNET_CONTAINER_multihashmap_remove (route_handle->dht_handle->active_requests,
 790                                                        &route_handle->key,
 791                                                        route_handle));
 792   GNUNET_free(route_handle->message);
 793   GNUNET_free(route_handle);
 794   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 795               "DHT route stop request processed\n");
 796 }
 797
 798
 799
 800 /* ***** Special API for controlling DHT routing maintenance ******* */
 801
 802
 803 /**
 804  * Send a control message to the DHT.
 805  *
 806  * @param handle handle to the DHT service
 807  * @param command command
 808  * @param variable variable to the command
 809  * @param cont continuation to call when done (transmitting request to service)
 810  * @param cont_cls closure for cont
 811  */
 812 static void
 813 send_control_message (struct GNUNET_DHT_Handle *handle,
 814                       uint16_t command,
 815                       uint16_t variable,
 816                       GNUNET_SCHEDULER_Task cont,
 817                       void *cont_cls)
 818 {
 819   struct GNUNET_DHT_ControlMessage *msg;
 820   struct PendingMessage *pending;
 821
 822   pending = GNUNET_malloc (sizeof (struct PendingMessage) +
 823                            sizeof(struct GNUNET_DHT_ControlMessage));
 824   msg = (struct GNUNET_DHT_ControlMessage*) &pending[1];
 825   pending->msg = &msg->header;
 826   msg->header.size = htons (sizeof(struct GNUNET_DHT_ControlMessage));
 827   msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 828   msg->command = htons (command);
 829   msg->variable = htons (variable);
 830   pending->free_on_send = GNUNET_YES;
 831   pending->cont = cont;
 832   pending->cont_cls = cont_cls;
 833   pending->in_pending_queue = GNUNET_YES;
 834   GNUNET_CONTAINER_DLL_insert (handle->pending_head,
 835                                handle->pending_tail,
 836                                pending);
 837   process_pending_messages (handle);
 838 }
 839
 840
 841 /**
 842  * Send a message to the DHT telling it to issue a single find
 843  * peer request using the peers unique identifier as key.  This
 844  * is used to fill the routing table, and is normally controlled
 845  * by the DHT itself.  However, for testing and perhaps more
 846  * close control over the DHT, this can be explicitly managed.
 847  *
 848  * @param handle handle to the DHT service
 849  * @param cont continuation to call when done (transmitting request to service)
 850  * @param cont_cls closure for cont
 851  */
 852 void
 853 GNUNET_DHT_find_peers (struct GNUNET_DHT_Handle *handle,
 854                        GNUNET_SCHEDULER_Task cont,
 855                        void *cont_cls)
 856 {
 857   send_control_message (handle,
 858                         GNUNET_MESSAGE_TYPE_DHT_FIND_PEER, 0,
 859                         cont, cont_cls);
 860 }
 861
 862
 863
 864 #if HAVE_MALICIOUS
 865
 866 /**
 867  * Send a message to the DHT telling it to start issuing random GET
 868  * requests every 'frequency' milliseconds.
 869  *
 870  * @param handle handle to the DHT service
 871  * @param frequency delay between sending malicious messages
 872  */
 873 void
 874 GNUNET_DHT_set_malicious_getter (struct GNUNET_DHT_Handle *handle,
 875                                  struct GNUNET_TIME_Relative frequency)
 876 {
 877   if (frequency.rel_value > UINT16_MAX)
 878     {
 879       GNUNET_break (0);
 880       return;
 881     }
 882   send_control_message (handle,
 883                         GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_GET, frequency.rel_value,
 884                         NULL, NULL);
 885 }
 886
 887 /**
 888  * Send a message to the DHT telling it to start issuing random PUT
 889  * requests every 'frequency' milliseconds.
 890  *
 891  * @param handle handle to the DHT service
 892  * @param frequency delay between sending malicious messages
 893  */
 894 void
 895 GNUNET_DHT_set_malicious_putter (struct GNUNET_DHT_Handle *handle,
 896                                  struct GNUNET_TIME_Relative frequency)
 897 {
 898   if (frequency.rel_value > UINT16_MAX)
 899     {
 900       GNUNET_break (0);
 901       return;
 902     }
 903
 904   send_control_message (handle,
 905                         GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_PUT, frequency.rel_value,
 906                         NULL, NULL);
 907 }
 908
 909
 910 /**
 911  * Send a message to the DHT telling it to start dropping
 912  * all requests received.
 913  *
 914  * @param handle handle to the DHT service
 915  */
 916 void
 917 GNUNET_DHT_set_malicious_dropper (struct GNUNET_DHT_Handle *handle)
 918 {
 919   send_control_message (handle,
 920                         GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_DROP, 0,
 921                         NULL, NULL);
 922 }
 923
 924 #endif
 925
 926 /* end of dht_api.c */