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  * TODO: retransmission of pending requests maybe happens now, at least
  28  *       the code is in place to do so.  Need to add checks when api calls
  29  *       happen to check if retransmission is in progress, and if so set
  30  *       the single pending message for transmission once the list of
  31  *       retries are done.
  32  */
  33
  34 #include "platform.h"
  35 #include "gnunet_bandwidth_lib.h"
  36 #include "gnunet_client_lib.h"
  37 #include "gnunet_constants.h"
  38 #include "gnunet_container_lib.h"
  39 #include "gnunet_arm_service.h"
  40 #include "gnunet_hello_lib.h"
  41 #include "gnunet_protocols.h"
  42 #include "gnunet_server_lib.h"
  43 #include "gnunet_time_lib.h"
  44 #include "gnunet_dht_service.h"
  45 #include "dht.h"
  46
  47 #define DEBUG_DHT_API GNUNET_NO
  48
  49 struct PendingMessage
  50 {
  51   /**
  52    * Message that is pending
  53    */
  54   struct GNUNET_MessageHeader *msg;
  55
  56   /**
  57    * Timeout for this message
  58    */
  59   struct GNUNET_TIME_Relative timeout;
  60
  61   /**
  62    * Continuation to call on message send
  63    * or message receipt confirmation
  64    */
  65   GNUNET_SCHEDULER_Task cont;
  66
  67   /**
  68    * Continuation closure
  69    */
  70   void *cont_cls;
  71
  72   /**
  73    * Unique ID for this request
  74    */
  75   uint64_t unique_id;
  76
  77   /**
  78    * Free the saved message once sent, set
  79    * to GNUNET_YES for messages that don't
  80    * receive responses!
  81    */
  82   int free_on_send;
  83
  84 };
  85
  86 struct PendingMessageList
  87 {
  88   /**
  89    * This is a singly linked list.
  90    */
  91   struct PendingMessageList *next;
  92
  93   /**
  94    * The pending message.
  95    */
  96   struct PendingMessage *message;
  97 };
  98
  99 struct GNUNET_DHT_GetContext
 100 {
 101   /**
 102    * Iterator to call on data receipt
 103    */
 104   GNUNET_DHT_GetIterator iter;
 105
 106   /**
 107    * Closure for the iterator callback
 108    */
 109   void *iter_cls;
 110
 111 };
 112
 113 struct GNUNET_DHT_FindPeerContext
 114 {
 115   /**
 116    * Iterator to call on data receipt
 117    */
 118   GNUNET_DHT_FindPeerProcessor proc;
 119
 120   /**
 121    * Closure for the iterator callback
 122    */
 123   void *proc_cls;
 124
 125 };
 126
 127 /**
 128  * Handle to a route request
 129  */
 130 struct GNUNET_DHT_RouteHandle
 131 {
 132
 133   /**
 134    * Unique identifier for this request (for key collisions)
 135    */
 136   uint64_t uid;
 137
 138   /**
 139    * Key that this get request is for
 140    */
 141   GNUNET_HashCode key;
 142
 143   /**
 144    * Iterator to call on data receipt
 145    */
 146   GNUNET_DHT_ReplyProcessor iter;
 147
 148   /**
 149    * Closure for the iterator callback
 150    */
 151   void *iter_cls;
 152
 153   /**
 154    * Main handle to this DHT api
 155    */
 156   struct GNUNET_DHT_Handle *dht_handle;
 157
 158   /**
 159    * The actual message sent for this request,
 160    * used for retransmitting requests on service
 161    * failure/reconnect.  Freed on route_stop.
 162    */
 163   struct GNUNET_DHT_RouteMessage *message;
 164 };
 165
 166
 167 /**
 168  * Handle to control a get operation.
 169  */
 170 struct GNUNET_DHT_GetHandle
 171 {
 172   /**
 173    * Handle to the actual route operation for the get
 174    */
 175   struct GNUNET_DHT_RouteHandle *route_handle;
 176
 177   /**
 178    * The context of the get request
 179    */
 180   struct GNUNET_DHT_GetContext get_context;
 181 };
 182
 183
 184 /**
 185  * Handle to control a find peer operation.
 186  */
 187 struct GNUNET_DHT_FindPeerHandle
 188 {
 189   /**
 190      * Handle to the actual route operation for the request
 191      */
 192   struct GNUNET_DHT_RouteHandle *route_handle;
 193
 194     /**
 195      * The context of the find peer request
 196      */
 197   struct GNUNET_DHT_FindPeerContext find_peer_context;
 198 };
 199
 200
 201 enum DHT_Retransmit_Stage
 202 {
 203   /**
 204    * The API is not retransmitting anything at this time.
 205    */
 206   DHT_NOT_RETRANSMITTING,
 207
 208   /**
 209    * The API is retransmitting, and nothing has been single
 210    * queued for sending.
 211    */
 212   DHT_RETRANSMITTING,
 213
 214   /**
 215    * The API is retransmitting, and a single message has been
 216    * queued for transmission once finished.
 217    */
 218   DHT_RETRANSMITTING_MESSAGE_QUEUED
 219 };
 220
 221
 222 /**
 223  * Connection to the DHT service.
 224  */
 225 struct GNUNET_DHT_Handle
 226 {
 227   /**
 228    * Our scheduler.
 229    */
 230   struct GNUNET_SCHEDULER_Handle *sched;
 231
 232   /**
 233    * Configuration to use.
 234    */
 235   const struct GNUNET_CONFIGURATION_Handle *cfg;
 236
 237   /**
 238    * Socket (if available).
 239    */
 240   struct GNUNET_CLIENT_Connection *client;
 241
 242   /**
 243    * Currently pending transmission request.
 244    */
 245   struct GNUNET_CLIENT_TransmitHandle *th;
 246
 247   /**
 248    * Message we are currently sending, only allow
 249    * a single message to be queued.  If not unique
 250    * (typically a put request), await a confirmation
 251    * from the service that the message was received.
 252    * If unique, just fire and forget.
 253    */
 254   struct PendingMessage *current;
 255
 256   /**
 257    * Hash map containing the current outstanding unique requests
 258    */
 259   struct GNUNET_CONTAINER_MultiHashMap *outstanding_requests;
 260
 261   /**
 262    * Generator for unique ids.
 263    */
 264   uint64_t uid_gen;
 265
 266   /**
 267    * Are we currently retransmitting requests?  If so queue a _single_
 268    * new request when received.
 269    */
 270   enum DHT_Retransmit_Stage retransmit_stage;
 271
 272   /**
 273    * Linked list of retranmissions, to be used in the event
 274    * of a dht service disconnect/reconnect.
 275    */
 276   struct PendingMessageList *retransmissions;
 277
 278   /**
 279    * A single pending message allowed to be scheduled
 280    * during retransmission phase.
 281    */
 282   struct PendingMessage *retransmission_buffer;
 283 };
 284
 285
 286 /**
 287  * Convert unique ID to hash code.
 288  *
 289  * @param uid unique ID to convert
 290  * @param hash set to uid (extended with zeros)
 291  */
 292 static void
 293 hash_from_uid (uint64_t uid,
 294                GNUNET_HashCode *hash)
 295 {
 296   memset (hash, 0, sizeof(GNUNET_HashCode));
 297   *((uint64_t*)hash) = uid;
 298 }
 299
 300 #if RETRANSMIT
 301 /**
 302  * Iterator callback to retransmit each outstanding request
 303  * because the connection to the DHT service went down (and
 304  * came back).
 305  *
 306  *
 307  */
 308 static int retransmit_iterator (void *cls,
 309                                 const GNUNET_HashCode * key,
 310                                 void *value)
 311 {
 312   struct GNUNET_DHT_RouteHandle *route_handle = value;
 313   struct PendingMessageList *pending_message_list;
 314
 315   pending_message_list = GNUNET_malloc(sizeof(struct PendingMessageList) + sizeof(struct PendingMessage));
 316   pending_message_list->message = (struct PendingMessage *)&pending_message_list[1];
 317   pending_message_list->message->msg = &route_handle->message->header;
 318   pending_message_list->message->timeout = GNUNET_TIME_relative_get_forever();
 319   pending_message_list->message->cont = NULL;
 320   pending_message_list->message->cont_cls = NULL;
 321   pending_message_list->message->unique_id = route_handle->uid;
 322   /* Add the new pending message to the front of the retransmission list */
 323   pending_message_list->next = route_handle->dht_handle->retransmissions;
 324   route_handle->dht_handle->retransmissions = pending_message_list;
 325
 326   return GNUNET_OK;
 327 }
 328 #endif
 329
 330 /**
 331  * Try to (re)connect to the dht service.
 332  *
 333  * @return GNUNET_YES on success, GNUNET_NO on failure.
 334  */
 335 static int
 336 try_connect (struct GNUNET_DHT_Handle *handle)
 337 {
 338   if (handle->client != NULL)
 339     return GNUNET_OK;
 340   handle->client = GNUNET_CLIENT_connect (handle->sched, "dht", handle->cfg);
 341   if (handle->client != NULL)
 342     return GNUNET_YES;
 343 #if DEBUG_STATISTICS
 344   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 345               _("Failed to connect to the dht service!\n"));
 346 #endif
 347   return GNUNET_NO;
 348 }
 349
 350 /**
 351  * Send complete (or failed), call continuation if we have one.
 352  */
 353 static void
 354 finish (struct GNUNET_DHT_Handle *handle, int code)
 355 {
 356   struct PendingMessage *pos = handle->current;
 357   GNUNET_HashCode uid_hash;
 358 #if DEBUG_DHT_API
 359   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Finish called!\n", "DHT API");
 360 #endif
 361   GNUNET_assert (pos != NULL);
 362   hash_from_uid (pos->unique_id, &uid_hash);
 363   if (pos->cont != NULL)
 364     {
 365       if (code == GNUNET_SYSERR)
 366         GNUNET_SCHEDULER_add_continuation (handle->sched, pos->cont,
 367                                            pos->cont_cls,
 368                                            GNUNET_SCHEDULER_REASON_TIMEOUT);
 369       else
 370         GNUNET_SCHEDULER_add_continuation (handle->sched, pos->cont,
 371                                            pos->cont_cls,
 372                                            GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 373     }
 374
 375   GNUNET_assert(handle->th == NULL);
 376   if (pos->free_on_send == GNUNET_YES)
 377     GNUNET_free(pos->msg);
 378   GNUNET_free (pos);
 379   handle->current = NULL;
 380 }
 381
 382 /**
 383  * Transmit the next pending message, called by notify_transmit_ready
 384  */
 385 static size_t
 386 transmit_pending (void *cls, size_t size, void *buf)
 387 {
 388   struct GNUNET_DHT_Handle *handle = cls;
 389   size_t tsize;
 390
 391 #if DEBUG_DHT_API
 392   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 393               "`%s': In transmit_pending\n", "DHT API");
 394 #endif
 395   handle->th = NULL;
 396
 397   if (buf == NULL)
 398     {
 399 #if DEBUG_DHT_API
 400       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 401                   "`%s': In transmit_pending buf is NULL\n", "DHT API");
 402 #endif
 403       finish (handle, GNUNET_SYSERR);
 404       return 0;
 405     }
 406
 407   if (handle->current != NULL)
 408     {
 409       tsize = ntohs (handle->current->msg->size);
 410       if (size >= tsize)
 411         {
 412 #if DEBUG_DHT_API
 413           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 414                       "`%s': Sending message size %d\n", "DHT API", tsize);
 415 #endif
 416           memcpy (buf, handle->current->msg, tsize);
 417           finish (handle, GNUNET_OK);
 418           return tsize;
 419         }
 420       else
 421         {
 422           return 0;
 423         }
 424     }
 425   /* Have no pending request */
 426   return 0;
 427 }
 428
 429 /**
 430  * Try to send messages from list of messages to send
 431  */
 432 static void
 433 process_pending_message (struct GNUNET_DHT_Handle *handle)
 434 {
 435
 436   if (handle->current == NULL)
 437     return;                     /* action already pending */
 438   if (GNUNET_YES != try_connect (handle))
 439     {
 440       handle->th = NULL;
 441       finish (handle, GNUNET_SYSERR);
 442       return;
 443     }
 444
 445   if (NULL ==
 446       (handle->th = GNUNET_CLIENT_notify_transmit_ready (handle->client,
 447                                                          ntohs (handle->
 448                                                                 current->msg->
 449                                                                 size),
 450                                                          handle->current->
 451                                                          timeout, GNUNET_YES,
 452                                                          &transmit_pending,
 453                                                          handle)))
 454     {
 455 #if DEBUG_DHT_API
 456       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 457                   "Failed to transmit request to dht service.\n");
 458 #endif
 459       finish (handle, GNUNET_SYSERR);
 460       return;
 461     }
 462 #if DEBUG_DHT_API
 463   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 464               "`%s': Scheduled sending message of size %d to service\n",
 465               "DHT API", ntohs (handle->current->msg->size));
 466 #endif
 467 }
 468
 469 /**
 470  * Send complete (or failed), call continuation if we have one.
 471  * Forward declaration.
 472  */
 473 static void
 474 finish_retransmission (struct GNUNET_DHT_Handle *handle, int code);
 475
 476 /* Forward declaration */
 477 static size_t
 478 transmit_pending_retransmission (void *cls, size_t size, void *buf);
 479
 480 /**
 481  * Try to send messages from list of messages to send
 482  */
 483 static void
 484 process_pending_retransmissions (struct GNUNET_DHT_Handle *handle)
 485 {
 486
 487   if (handle->current == NULL)
 488     return;                     /* action already pending */
 489   if (GNUNET_YES != try_connect (handle))
 490     {
 491       finish_retransmission (handle, GNUNET_SYSERR);
 492       return;
 493     }
 494
 495   if (NULL ==
 496       (handle->th = GNUNET_CLIENT_notify_transmit_ready (handle->client,
 497                                                          ntohs (handle->
 498                                                                 current->msg->
 499                                                                 size),
 500                                                          handle->current->
 501                                                          timeout, GNUNET_YES,
 502                                                          &transmit_pending_retransmission,
 503                                                          handle)))
 504     {
 505 #if DEBUG_DHT_API
 506       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 507                   "Failed to transmit request to dht service.\n");
 508 #endif
 509       finish_retransmission (handle, GNUNET_SYSERR);
 510       return;
 511     }
 512 #if DEBUG_DHT_API
 513   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 514               "`%s': Scheduled sending message of size %d to service\n",
 515               "DHT API", ntohs (handle->current->msg->size));
 516 #endif
 517 }
 518
 519 /**
 520  * Send complete (or failed), call continuation if we have one.
 521  */
 522 static void
 523 finish_retransmission (struct GNUNET_DHT_Handle *handle, int code)
 524 {
 525   struct PendingMessage *pos = handle->current;
 526   struct PendingMessageList *pending_list;
 527 #if DEBUG_DHT_API
 528   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Finish (retransmission) called!\n", "DHT API");
 529 #endif
 530   GNUNET_assert (pos == handle->retransmissions->message);
 531   pending_list = handle->retransmissions;
 532   handle->retransmissions = handle->retransmissions->next;
 533   GNUNET_free (pending_list);
 534
 535   if (handle->retransmissions == NULL)
 536     {
 537       handle->retransmit_stage = DHT_NOT_RETRANSMITTING;
 538     }
 539
 540   if (handle->retransmissions != NULL)
 541     {
 542       handle->current = handle->retransmissions->message;
 543       process_pending_retransmissions(handle);
 544     }
 545   else if (handle->retransmission_buffer != NULL)
 546     {
 547       handle->current = handle->retransmission_buffer;
 548       process_pending_message(handle);
 549     }
 550 }
 551
 552 /**
 553  * Handler for messages received from the DHT service
 554  * a demultiplexer which handles numerous message types
 555  *
 556  */
 557 void
 558 service_message_handler (void *cls,
 559                          const struct GNUNET_MessageHeader *msg)
 560 {
 561   struct GNUNET_DHT_Handle *handle = cls;
 562   struct GNUNET_DHT_RouteResultMessage *dht_msg;
 563   struct GNUNET_MessageHeader *enc_msg;
 564   struct GNUNET_DHT_RouteHandle *route_handle;
 565   uint64_t uid;
 566   GNUNET_HashCode uid_hash;
 567   size_t enc_size;
 568
 569   if (msg == NULL)
 570     {
 571 #if DEBUG_DHT_API
 572       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 573                   "`%s': Received NULL from server, connection down!\n",
 574                   "DHT API");
 575 #endif
 576       GNUNET_CLIENT_disconnect (handle->client, GNUNET_YES);
 577       handle->client = GNUNET_CLIENT_connect (handle->sched,
 578                                               "dht",
 579                                               handle->cfg);
 580       if (handle->current != NULL)
 581         {
 582           handle->th = NULL;
 583           finish(handle, GNUNET_SYSERR); /* If there was a current message, kill it! */
 584         }
 585 #if RETRANSMIT
 586       if ((handle->retransmit_stage != DHT_RETRANSMITTING) && (GNUNET_CONTAINER_multihashmap_iterate(handle->outstanding_requests, &retransmit_iterator, handle) > 0))
 587         {
 588           handle->retransmit_stage = DHT_RETRANSMITTING;
 589           handle->current = handle->retransmissions->message;
 590           process_pending_retransmissions(handle);
 591         }
 592 #endif
 593       return;
 594     }
 595
 596   switch (ntohs (msg->type))
 597     {
 598     case GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_RESULT:
 599       {
 600         dht_msg = (struct GNUNET_DHT_RouteResultMessage *) msg;
 601         uid = GNUNET_ntohll (dht_msg->unique_id);
 602 #if DEBUG_DHT_API
 603         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 604                     "`%s': Received response to message (uid %llu)\n",
 605                     "DHT API", uid);
 606 #endif
 607
 608         hash_from_uid (uid, &uid_hash);
 609         route_handle =
 610           GNUNET_CONTAINER_multihashmap_get (handle->outstanding_requests,
 611                                              &uid_hash);
 612         if (route_handle == NULL)   /* We have no recollection of this request */
 613           {
 614 #if DEBUG_DHT_API
 615             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 616                         "`%s': Received response to message (uid %llu), but have no recollection of it!\n",
 617                         "DHT API", uid);
 618 #endif
 619           }
 620         else
 621           {
 622             enc_size =
 623               ntohs (dht_msg->header.size) -
 624               sizeof (struct GNUNET_DHT_RouteResultMessage);
 625             GNUNET_assert (enc_size > 0);
 626             enc_msg = (struct GNUNET_MessageHeader *) &dht_msg[1];
 627             route_handle->iter (route_handle->iter_cls, enc_msg);
 628           }
 629
 630         break;
 631       }
 632     default:
 633       {
 634         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 635                     "`%s': Received unknown message type %d\n", "DHT API",
 636                     ntohs (msg->type));
 637       }
 638     }
 639   GNUNET_CLIENT_receive (handle->client,
 640                          &service_message_handler,
 641                          handle, GNUNET_TIME_UNIT_FOREVER_REL);
 642
 643 }
 644
 645
 646 /**
 647  * Initialize the connection with the DHT service.
 648  *
 649  * @param sched scheduler to use
 650  * @param cfg configuration to use
 651  * @param ht_len size of the internal hash table to use for
 652  *               processing multiple GET/FIND requests in parallel
 653  *
 654  * @return handle to the DHT service, or NULL on error
 655  */
 656 struct GNUNET_DHT_Handle *
 657 GNUNET_DHT_connect (struct GNUNET_SCHEDULER_Handle *sched,
 658                     const struct GNUNET_CONFIGURATION_Handle *cfg,
 659                     unsigned int ht_len)
 660 {
 661   struct GNUNET_DHT_Handle *handle;
 662
 663   handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_Handle));
 664   handle->cfg = cfg;
 665   handle->sched = sched;
 666   handle->client = GNUNET_CLIENT_connect (sched, "dht", cfg);
 667   handle->uid_gen = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, -1);
 668   if (handle->client == NULL)
 669     {
 670       GNUNET_free (handle);
 671       return NULL;
 672     }
 673   handle->outstanding_requests =
 674     GNUNET_CONTAINER_multihashmap_create (ht_len);
 675 #if DEBUG_DHT_API
 676   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 677               "`%s': Connection to service in progress\n", "DHT API");
 678 #endif
 679   GNUNET_CLIENT_receive (handle->client,
 680                          &service_message_handler,
 681                          handle, GNUNET_TIME_UNIT_FOREVER_REL);
 682   return handle;
 683 }
 684
 685
 686 /**
 687  * Shutdown connection with the DHT service.
 688  *
 689  * @param handle handle of the DHT connection to stop
 690  */
 691 void
 692 GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle)
 693 {
 694 #if DEBUG_DHT_API
 695   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 696               "`%s': Called GNUNET_DHT_disconnect\n", "DHT API");
 697 #endif
 698   GNUNET_assert (handle != NULL);
 699   if (handle->th != NULL)       /* We have a live transmit request */
 700     {
 701       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
 702       handle->th = NULL;
 703     }
 704   if (handle->current != NULL)  /* We are trying to send something now, clean it up */
 705     GNUNET_free (handle->current);
 706
 707   if (handle->client != NULL)   /* Finally, disconnect from the service */
 708     {
 709       GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
 710       handle->client = NULL;
 711     }
 712
 713   GNUNET_assert(GNUNET_CONTAINER_multihashmap_size(handle->outstanding_requests) == 0);
 714   GNUNET_CONTAINER_multihashmap_destroy(handle->outstanding_requests);
 715   GNUNET_free (handle);
 716 }
 717
 718
 719 /**
 720  * Transmit the next pending message, called by notify_transmit_ready
 721  */
 722 static size_t
 723 transmit_pending_retransmission (void *cls, size_t size, void *buf)
 724 {
 725   struct GNUNET_DHT_Handle *handle = cls;
 726   size_t tsize;
 727
 728 #if DEBUG_DHT_API
 729   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 730               "`%s': In transmit_pending\n", "DHT API");
 731 #endif
 732   if (buf == NULL)
 733     {
 734 #if DEBUG_DHT_API
 735       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 736                   "`%s': In transmit_pending buf is NULL\n", "DHT API");
 737 #endif
 738       finish_retransmission (handle, GNUNET_SYSERR);
 739       return 0;
 740     }
 741
 742   handle->th = NULL;
 743
 744   if (handle->current != NULL)
 745     {
 746       tsize = ntohs (handle->current->msg->size);
 747       if (size >= tsize)
 748         {
 749 #if DEBUG_DHT_API
 750           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 751                       "`%s': Sending message size %d\n", "DHT API", tsize);
 752 #endif
 753           memcpy (buf, handle->current->msg, tsize);
 754           finish_retransmission (handle, GNUNET_OK);
 755           return tsize;
 756         }
 757       else
 758         {
 759           return 0;
 760         }
 761     }
 762   /* Have no pending request */
 763   return 0;
 764 }
 765
 766
 767 /**
 768  * Iterator called on each result obtained from a generic route
 769  * operation
 770  */
 771 void
 772 get_reply_iterator (void *cls, const struct GNUNET_MessageHeader *reply)
 773 {
 774   struct GNUNET_DHT_GetHandle *get_handle = cls;
 775   struct GNUNET_DHT_GetResultMessage *result;
 776   size_t data_size;
 777   char *result_data;
 778
 779   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_GET_RESULT)
 780     return;
 781
 782   GNUNET_assert (ntohs (reply->size) >=
 783                  sizeof (struct GNUNET_DHT_GetResultMessage));
 784   result = (struct GNUNET_DHT_GetResultMessage *) reply;
 785   data_size = ntohs (reply->size) - sizeof(struct GNUNET_DHT_GetResultMessage);
 786
 787   result_data = (char *) &result[1];    /* Set data pointer to end of message */
 788
 789   get_handle->get_context.iter (get_handle->get_context.iter_cls,
 790                                 GNUNET_TIME_absolute_ntoh (result->expiration), &get_handle->route_handle->key,
 791                                 ntohs (result->type), data_size, result_data);
 792 }
 793
 794
 795 /**
 796  * Iterator called on each result obtained from a generic route
 797  * operation
 798  */
 799 void
 800 find_peer_reply_iterator (void *cls, const struct GNUNET_MessageHeader *reply)
 801 {
 802   struct GNUNET_DHT_FindPeerHandle *find_peer_handle = cls;
 803   struct GNUNET_MessageHeader *hello;
 804
 805   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_FIND_PEER_RESULT)
 806     {
 807       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 808                   "Received wrong type of response to a find peer request...\n");
 809       return;
 810     }
 811
 812
 813   GNUNET_assert (ntohs (reply->size) >=
 814                  sizeof (struct GNUNET_MessageHeader));
 815   hello = (struct GNUNET_MessageHeader *)&reply[1];
 816
 817   if (ntohs(hello->type) != GNUNET_MESSAGE_TYPE_HELLO)
 818     {
 819       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 820                   "Encapsulated message of type %d, is not a `%s' message!\n", ntohs(hello->type), "HELLO");
 821       return;
 822     }
 823   find_peer_handle->find_peer_context.proc (find_peer_handle->
 824                                             find_peer_context.proc_cls,
 825                                             (struct GNUNET_HELLO_Message *)hello);
 826 }
 827
 828 /**
 829  * Send a message to the DHT telling it to start issuing random GET
 830  * requests every 'frequency' milliseconds.
 831  *
 832  * @param handle handle to the DHT service
 833  * @param frequency delay (in milliseconds) between sending malicious messages
 834  * @param cont continuation to call once the message is sent
 835  * @param cont_cls closure for continuation
 836  *
 837  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 838  */
 839 int GNUNET_DHT_set_malicious_getter (struct GNUNET_DHT_Handle *handle, int frequency, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 840 {
 841   struct GNUNET_DHT_ControlMessage *msg;
 842   struct PendingMessage *pending;
 843
 844   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 845     return GNUNET_NO;
 846
 847   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 848   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 849   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 850   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_GET);
 851   msg->variable = htons(frequency);
 852
 853   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 854   pending->msg = &msg->header;
 855   pending->timeout = GNUNET_TIME_relative_get_forever();
 856   pending->free_on_send = GNUNET_YES;
 857   pending->cont = cont;
 858   pending->cont_cls = cont_cls;
 859   pending->unique_id = 0;
 860
 861   if (handle->current == NULL)
 862     {
 863       handle->current = pending;
 864       process_pending_message (handle);
 865     }
 866   else
 867   {
 868     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 869     handle->retransmission_buffer = pending;
 870   }
 871
 872   return GNUNET_YES;
 873 }
 874
 875 /**
 876  * Send a message to the DHT telling it to issue a single find
 877  * peer request using the peers unique identifier as key.  This
 878  * is used to fill the routing table, and is normally controlled
 879  * by the DHT itself.  However, for testing and perhaps more
 880  * close control over the DHT, this can be explicitly managed.
 881  *
 882  * @param handle handle to the DHT service
 883  * @param cont continuation to call once the message is sent
 884  * @param cont_cls closure for continuation
 885  *
 886  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 887  */
 888 int GNUNET_DHT_find_peers (struct GNUNET_DHT_Handle *handle,
 889                            GNUNET_SCHEDULER_Task cont, void *cont_cls)
 890 {
 891   struct GNUNET_DHT_ControlMessage *msg;
 892   struct PendingMessage *pending;
 893
 894   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 895     return GNUNET_NO;
 896
 897   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 898   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 899   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 900   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_FIND_PEER);
 901
 902   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 903   pending->msg = &msg->header;
 904   pending->timeout = GNUNET_TIME_relative_get_forever();
 905   pending->free_on_send = GNUNET_YES;
 906   pending->cont = cont;
 907   pending->cont_cls = cont_cls;
 908   pending->unique_id = 0;
 909
 910   if (handle->current == NULL)
 911     {
 912       handle->current = pending;
 913       process_pending_message (handle);
 914     }
 915   else
 916   {
 917     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 918     handle->retransmission_buffer = pending;
 919   }
 920
 921   return GNUNET_YES;
 922 }
 923
 924 /**
 925  * Send a message to the DHT telling it to start issuing random PUT
 926  * requests every 'frequency' milliseconds.
 927  *
 928  * @param handle handle to the DHT service
 929  * @param frequency delay (in milliseconds) between sending malicious messages
 930  * @param cont continuation to call once the message is sent
 931  * @param cont_cls closure for continuation
 932  *
 933  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 934  */
 935 int GNUNET_DHT_set_malicious_putter (struct GNUNET_DHT_Handle *handle, int frequency, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 936 {
 937   struct GNUNET_DHT_ControlMessage *msg;
 938   struct PendingMessage *pending;
 939
 940   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 941     return GNUNET_NO;
 942
 943   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 944   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 945   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 946   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_PUT);
 947   msg->variable = htons(frequency);
 948
 949   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 950   pending->msg = &msg->header;
 951   pending->timeout = GNUNET_TIME_relative_get_forever();
 952   pending->free_on_send = GNUNET_YES;
 953   pending->cont = cont;
 954   pending->cont_cls = cont_cls;
 955   pending->unique_id = 0;
 956
 957   if (handle->current == NULL)
 958     {
 959       handle->current = pending;
 960       process_pending_message (handle);
 961     }
 962   else
 963   {
 964     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 965     handle->retransmission_buffer = pending;
 966   }
 967
 968   return GNUNET_YES;
 969 }
 970
 971 /**
 972  * Send a message to the DHT telling it to start dropping
 973  * all requests received.
 974  *
 975  * @param handle handle to the DHT service
 976  * @param cont continuation to call once the message is sent
 977  * @param cont_cls closure for continuation
 978  *
 979  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 980  */
 981 int GNUNET_DHT_set_malicious_dropper (struct GNUNET_DHT_Handle *handle, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 982 {
 983   struct GNUNET_DHT_ControlMessage *msg;
 984   struct PendingMessage *pending;
 985
 986   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 987     return GNUNET_NO;
 988
 989   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 990   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 991   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 992   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_DROP);
 993   msg->variable = htons(0);
 994
 995   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 996   pending->msg = &msg->header;
 997   pending->timeout = GNUNET_TIME_relative_get_forever();
 998   pending->free_on_send = GNUNET_YES;
 999   pending->cont = cont;
1000   pending->cont_cls = cont_cls;
1001   pending->unique_id = 0;
1002
1003   if (handle->current == NULL)
1004     {
1005       handle->current = pending;
1006       process_pending_message (handle);
1007     }
1008   else
1009   {
1010     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
1011     handle->retransmission_buffer = pending;
1012   }
1013
1014   return GNUNET_YES;
1015 }
1016
1017
1018 /**
1019  * Initiate a generic DHT route operation.
1020  *
1021  * @param handle handle to the DHT service
1022  * @param key the key to look up
1023  * @param desired_replication_level how many peers should ultimately receive
1024  *                this message (advisory only, target may be too high for the
1025  *                given DHT or not hit exactly).
1026  * @param options options for routing
1027  * @param enc send the encapsulated message to a peer close to the key
1028  * @param iter function to call on each result, NULL if no replies are expected
1029  * @param iter_cls closure for iter
1030  * @param timeout when to abort with an error if we fail to get
1031  *                a confirmation for the request (when necessary) or how long
1032  *                to wait for tramission to the service
1033  * @param cont continuation to call when done;
1034  *             reason will be TIMEOUT on error,
1035  *             reason will be PREREQ_DONE on success
1036  * @param cont_cls closure for cont
1037  *
1038  * @return handle to stop the request, NULL if the request is "fire and forget"
1039  */
1040 struct GNUNET_DHT_RouteHandle *
1041 GNUNET_DHT_route_start (struct GNUNET_DHT_Handle *handle,
1042                         const GNUNET_HashCode * key,
1043                         unsigned int desired_replication_level,
1044                         enum GNUNET_DHT_RouteOption options,
1045                         const struct GNUNET_MessageHeader *enc,
1046                         struct GNUNET_TIME_Relative timeout,
1047                         GNUNET_DHT_ReplyProcessor iter,
1048                         void *iter_cls,
1049                         GNUNET_SCHEDULER_Task cont, void *cont_cls)
1050 {
1051   struct GNUNET_DHT_RouteHandle *route_handle;
1052   struct PendingMessage *pending;
1053   struct GNUNET_DHT_RouteMessage *message;
1054   uint16_t msize;
1055   GNUNET_HashCode uid_key;
1056
1057   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
1058     return NULL;
1059
1060   if (sizeof (struct GNUNET_DHT_RouteMessage) + ntohs (enc->size) >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1061     {
1062       GNUNET_break (0);
1063       return NULL;
1064     }
1065
1066   route_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_RouteHandle));
1067   memcpy (&route_handle->key, key, sizeof (GNUNET_HashCode));
1068   route_handle->iter = iter;
1069   route_handle->iter_cls = iter_cls;
1070   route_handle->dht_handle = handle;
1071   route_handle->uid = handle->uid_gen++;
1072   if (iter != NULL)
1073     {
1074       hash_from_uid (route_handle->uid, &uid_key);
1075       GNUNET_CONTAINER_multihashmap_put (handle->outstanding_requests,
1076                                          &uid_key, route_handle,
1077                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1078     }
1079
1080 #if DEBUG_DHT_API
1081       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1082                   "`%s': Unique ID is %llu\n", "DHT API", route_handle->uid);
1083 #endif
1084
1085   msize = sizeof (struct GNUNET_DHT_RouteMessage) + ntohs (enc->size);
1086   message = GNUNET_malloc (msize);
1087   message->header.size = htons (msize);
1088   message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE);
1089   memcpy (&message->key, key, sizeof (GNUNET_HashCode));
1090   message->options = htonl (options);
1091   message->desired_replication_level = htonl (desired_replication_level);
1092   message->unique_id = GNUNET_htonll (route_handle->uid);
1093   memcpy (&message[1], enc, ntohs (enc->size));
1094   pending = GNUNET_malloc (sizeof (struct PendingMessage));
1095   pending->msg = &message->header;
1096   pending->timeout = timeout;
1097   if (iter == NULL)
1098     pending->free_on_send = GNUNET_YES;
1099   pending->cont = cont;
1100   pending->cont_cls = cont_cls;
1101   pending->unique_id = route_handle->uid;
1102   if (handle->current == NULL)
1103     {
1104       handle->current = pending;
1105       process_pending_message (handle);
1106     }
1107   else
1108   {
1109     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
1110     handle->retransmission_buffer = pending;
1111   }
1112
1113   route_handle->message = message;
1114   return route_handle;
1115 }
1116
1117
1118 /**
1119  * Perform an asynchronous GET operation on the DHT identified.
1120  *
1121  * @param handle handle to the DHT service
1122  * @param timeout how long to wait for transmission of this request to the service
1123  * @param type expected type of the response object
1124  * @param key the key to look up
1125  * @param iter function to call on each result
1126  * @param iter_cls closure for iter
1127  * @param cont continuation to call once message sent
1128  * @param cont_cls closure for continuation
1129  *
1130  * @return handle to stop the async get
1131  */
1132 struct GNUNET_DHT_GetHandle *
1133 GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
1134                       struct GNUNET_TIME_Relative timeout,
1135                       enum GNUNET_BLOCK_Type type,
1136                       const GNUNET_HashCode * key,
1137                       GNUNET_DHT_GetIterator iter,
1138                       void *iter_cls,
1139                       GNUNET_SCHEDULER_Task cont, void *cont_cls)
1140 {
1141   struct GNUNET_DHT_GetHandle *get_handle;
1142   struct GNUNET_DHT_GetMessage get_msg;
1143
1144   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING)) /* Can't send right now, we have a pending message... */
1145     return NULL;
1146
1147   get_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle));
1148   get_handle->get_context.iter = iter;
1149   get_handle->get_context.iter_cls = iter_cls;
1150
1151 #if DEBUG_DHT_API
1152   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1153               "`%s': Inserting pending get request with key %s\n", "DHT API",
1154               GNUNET_h2s (key));
1155 #endif
1156
1157   get_msg.header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET);
1158   get_msg.header.size = htons (sizeof (struct GNUNET_DHT_GetMessage));
1159   get_msg.type = htons (type);
1160
1161   get_handle->route_handle =
1162     GNUNET_DHT_route_start (handle, key, DEFAULT_GET_REPLICATION, 0, &get_msg.header, timeout,
1163                             &get_reply_iterator, get_handle, cont, cont_cls);
1164
1165   return get_handle;
1166 }
1167
1168
1169 /**
1170  * Stop a previously issued routing request
1171  *
1172  * @param route_handle handle to the request to stop
1173  * @param cont continuation to call once this message is sent to the service or times out
1174  * @param cont_cls closure for the continuation
1175  */
1176 void
1177 GNUNET_DHT_route_stop (struct GNUNET_DHT_RouteHandle *route_handle,
1178                        GNUNET_SCHEDULER_Task cont, void *cont_cls)
1179 {
1180   struct PendingMessage *pending;
1181   struct GNUNET_DHT_StopMessage *message;
1182   size_t msize;
1183   GNUNET_HashCode uid_key;
1184
1185   msize = sizeof (struct GNUNET_DHT_StopMessage);
1186   message = GNUNET_malloc (msize);
1187   message->header.size = htons (msize);
1188   message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_STOP);
1189 #if DEBUG_DHT_API
1190   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1191               "`%s': Remove outstanding request for uid %llu\n", "DHT API",
1192               route_handle->uid);
1193 #endif
1194   message->unique_id = GNUNET_htonll (route_handle->uid);
1195   memcpy(&message->key, &route_handle->key, sizeof(GNUNET_HashCode));
1196   pending = GNUNET_malloc (sizeof (struct PendingMessage));
1197   pending->msg = (struct GNUNET_MessageHeader *) message;
1198   pending->timeout = GNUNET_TIME_relative_get_forever();
1199   pending->cont = cont;
1200   pending->cont_cls = cont_cls;
1201   pending->free_on_send = GNUNET_YES;
1202   pending->unique_id = 0; /* When finished is called, free pending->msg */
1203
1204   if (route_handle->dht_handle->current == NULL)
1205     {
1206       route_handle->dht_handle->current = pending;
1207       process_pending_message (route_handle->dht_handle);
1208     }
1209   else if (route_handle->dht_handle->retransmit_stage == DHT_RETRANSMITTING)
1210     {
1211       route_handle->dht_handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
1212       route_handle->dht_handle->retransmission_buffer = pending;
1213     }
1214   else
1215     {
1216       GNUNET_free(pending);
1217       GNUNET_break(0);
1218     }
1219
1220   hash_from_uid (route_handle->uid, &uid_key);
1221   GNUNET_assert (GNUNET_CONTAINER_multihashmap_remove
1222                  (route_handle->dht_handle->outstanding_requests, &uid_key,
1223                   route_handle) == GNUNET_YES);
1224
1225   GNUNET_free(route_handle->message);
1226   GNUNET_free(route_handle);
1227 }
1228
1229
1230 /**
1231  * Stop async DHT-get.
1232  *
1233  * @param get_handle handle to the GET operation to stop
1234  * @param cont continuation to call once this message is sent to the service or times out
1235  * @param cont_cls closure for the continuation
1236  */
1237 void
1238 GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle,
1239                      GNUNET_SCHEDULER_Task cont, void *cont_cls)
1240 {
1241   if ((get_handle->route_handle->dht_handle->current != NULL) &&
1242       (get_handle->route_handle->dht_handle->retransmit_stage != DHT_RETRANSMITTING))
1243     {
1244       if (cont != NULL)
1245         {
1246           GNUNET_SCHEDULER_add_continuation (get_handle->route_handle->dht_handle->sched, cont, cont_cls,
1247                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1248         }
1249       return;
1250     }
1251
1252 #if DEBUG_DHT_API
1253   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1254               "`%s': Removing pending get request with key %s, uid %llu\n",
1255               "DHT API", GNUNET_h2s (&get_handle->route_handle->key),
1256               get_handle->route_handle->uid);
1257 #endif
1258   GNUNET_DHT_route_stop (get_handle->route_handle, cont, cont_cls);
1259   GNUNET_free (get_handle);
1260 }
1261
1262
1263 /**
1264  * Perform an asynchronous FIND PEER operation on the DHT.
1265  *
1266  * @param handle handle to the DHT service
1267  * @param timeout timeout for this request to be sent to the
1268  *        service
1269  * @param options routing options for this message
1270  * @param key the key to look up
1271  * @param proc function to call on each result
1272  * @param proc_cls closure for proc
1273  * @param cont continuation to call once message sent
1274  * @param cont_cls closure for continuation
1275  *
1276  * @return handle to stop the async get, NULL on error
1277  */
1278 struct GNUNET_DHT_FindPeerHandle *
1279 GNUNET_DHT_find_peer_start (struct GNUNET_DHT_Handle *handle,
1280                             struct GNUNET_TIME_Relative timeout,
1281                             enum GNUNET_DHT_RouteOption options,
1282                             const GNUNET_HashCode * key,
1283                             GNUNET_DHT_FindPeerProcessor proc,
1284                             void *proc_cls,
1285                             GNUNET_SCHEDULER_Task cont,
1286                             void *cont_cls)
1287 {
1288   struct GNUNET_DHT_FindPeerHandle *find_peer_handle;
1289   struct GNUNET_DHT_FindPeerMessage find_peer_msg;
1290
1291   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))  /* Can't send right now, we have a pending message... */
1292     return NULL;
1293
1294   find_peer_handle =
1295     GNUNET_malloc (sizeof (struct GNUNET_DHT_FindPeerHandle));
1296   find_peer_handle->find_peer_context.proc = proc;
1297   find_peer_handle->find_peer_context.proc_cls = proc_cls;
1298
1299 #if DEBUG_DHT_API
1300   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1301               "`%s': Inserting pending `%s' request with key %s\n", "DHT API",
1302               "FIND PEER", GNUNET_h2s (key));
1303 #endif
1304
1305   find_peer_msg.header.size = htons(sizeof(struct GNUNET_DHT_FindPeerMessage));
1306   find_peer_msg.header.type = htons(GNUNET_MESSAGE_TYPE_DHT_FIND_PEER);
1307   find_peer_handle->route_handle =
1308     GNUNET_DHT_route_start (handle, key, 0, options, &find_peer_msg.header,
1309                             timeout, &find_peer_reply_iterator,
1310                             find_peer_handle, cont, cont_cls);
1311   return find_peer_handle;
1312 }
1313
1314 /**
1315  * Stop async find peer.  Frees associated resources.
1316  *
1317  * @param find_peer_handle GET operation to stop.
1318  * @param cont continuation to call once this message is sent to the service or times out
1319  * @param cont_cls closure for the continuation
1320  */
1321 void
1322 GNUNET_DHT_find_peer_stop (struct GNUNET_DHT_FindPeerHandle *find_peer_handle,
1323                            GNUNET_SCHEDULER_Task cont, void *cont_cls)
1324 {
1325   if ((find_peer_handle->route_handle->dht_handle->current != NULL) &&
1326       (find_peer_handle->route_handle->dht_handle->retransmit_stage != DHT_RETRANSMITTING))
1327     {
1328       if (cont != NULL)
1329         {
1330           GNUNET_SCHEDULER_add_continuation (find_peer_handle->route_handle->dht_handle->sched, cont, cont_cls,
1331                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1332         }
1333       return;
1334     }
1335
1336 #if DEBUG_DHT_API
1337   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1338               "`%s': Removing pending `%s' request with key %s, uid %llu\n",
1339               "DHT API", "FIND PEER",
1340               GNUNET_h2s (&find_peer_handle->route_handle->key),
1341               find_peer_handle->route_handle->uid);
1342 #endif
1343   GNUNET_DHT_route_stop (find_peer_handle->route_handle, cont, cont_cls);
1344   GNUNET_free (find_peer_handle);
1345
1346 }
1347
1348
1349 /**
1350  * Perform a PUT operation storing data in the DHT.
1351  *
1352  * @param handle handle to DHT service
1353  * @param key the key to store under
1354  * @param type type of the value
1355  * @param size number of bytes in data; must be less than 64k
1356  * @param data the data to store
1357  * @param exp desired expiration time for the value
1358  * @param timeout how long to wait for transmission of this request
1359  * @param cont continuation to call when done;
1360  *             reason will be TIMEOUT on error,
1361  *             reason will be PREREQ_DONE on success
1362  * @param cont_cls closure for cont
1363  *
1364  * @return GNUNET_YES if put message is queued for transmission
1365  */
1366 void
1367 GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle,
1368                 const GNUNET_HashCode * key,
1369                 enum GNUNET_BLOCK_Type type,
1370                 uint32_t size,
1371                 const char *data,
1372                 struct GNUNET_TIME_Absolute exp,
1373                 struct GNUNET_TIME_Relative timeout,
1374                 GNUNET_SCHEDULER_Task cont, void *cont_cls)
1375 {
1376   struct GNUNET_DHT_PutMessage *put_msg;
1377   struct GNUNET_DHT_RouteHandle *put_route;
1378   size_t msize;
1379
1380   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
1381     {
1382       GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "handle->current is not null!\n");
1383       if (cont != NULL)
1384         {
1385           GNUNET_SCHEDULER_add_continuation (handle->sched, cont, cont_cls,
1386                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1387         }
1388       return;
1389     }
1390
1391 #if DEBUG_DHT_API
1392   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1393               "`%s': Inserting pending put request with key %s\n", "DHT API",
1394               GNUNET_h2s (key));
1395 #endif
1396
1397   msize = sizeof (struct GNUNET_DHT_PutMessage) + size;
1398   put_msg = GNUNET_malloc (msize);
1399   put_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_PUT);
1400   put_msg->header.size = htons (msize);
1401   put_msg->type = htons (type);
1402   put_msg->data_size = htons (size);
1403   put_msg->expiration = GNUNET_TIME_absolute_hton(exp);
1404   memcpy (&put_msg[1], data, size);
1405
1406   put_route = GNUNET_DHT_route_start (handle, key, DEFAULT_PUT_REPLICATION, 0, &put_msg->header, timeout, NULL,
1407                                       NULL, cont, cont_cls);
1408
1409   if (put_route == NULL) /* Route start failed! */
1410     {
1411       GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "route start for PUT failed!\n");
1412       if (cont != NULL)
1413         {
1414           GNUNET_SCHEDULER_add_continuation (handle->sched, cont, cont_cls,
1415                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1416         }
1417     }
1418   else
1419     {
1420       GNUNET_free(put_route);
1421     }
1422
1423   GNUNET_free (put_msg);
1424 }
1425
1426 /* end of dht_api.c */