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   if (buf == NULL)
 396     {
 397 #if DEBUG_DHT_API
 398       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 399                   "`%s': In transmit_pending buf is NULL\n", "DHT API");
 400 #endif
 401       finish (handle, GNUNET_SYSERR);
 402       return 0;
 403     }
 404
 405   handle->th = NULL;
 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       finish (handle, GNUNET_SYSERR);
 441       return;
 442     }
 443
 444   if (NULL ==
 445       (handle->th = GNUNET_CLIENT_notify_transmit_ready (handle->client,
 446                                                          ntohs (handle->
 447                                                                 current->msg->
 448                                                                 size),
 449                                                          handle->current->
 450                                                          timeout, GNUNET_YES,
 451                                                          &transmit_pending,
 452                                                          handle)))
 453     {
 454 #if DEBUG_DHT_API
 455       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 456                   "Failed to transmit request to dht service.\n");
 457 #endif
 458       finish (handle, GNUNET_SYSERR);
 459       return;
 460     }
 461 #if DEBUG_DHT_API
 462   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 463               "`%s': Scheduled sending message of size %d to service\n",
 464               "DHT API", ntohs (handle->current->msg->size));
 465 #endif
 466 }
 467
 468 /**
 469  * Send complete (or failed), call continuation if we have one.
 470  * Forward declaration.
 471  */
 472 static void
 473 finish_retransmission (struct GNUNET_DHT_Handle *handle, int code);
 474
 475 /* Forward declaration */
 476 static size_t
 477 transmit_pending_retransmission (void *cls, size_t size, void *buf);
 478
 479 /**
 480  * Try to send messages from list of messages to send
 481  */
 482 static void
 483 process_pending_retransmissions (struct GNUNET_DHT_Handle *handle)
 484 {
 485
 486   if (handle->current == NULL)
 487     return;                     /* action already pending */
 488   if (GNUNET_YES != try_connect (handle))
 489     {
 490       finish_retransmission (handle, GNUNET_SYSERR);
 491       return;
 492     }
 493
 494   if (NULL ==
 495       (handle->th = GNUNET_CLIENT_notify_transmit_ready (handle->client,
 496                                                          ntohs (handle->
 497                                                                 current->msg->
 498                                                                 size),
 499                                                          handle->current->
 500                                                          timeout, GNUNET_YES,
 501                                                          &transmit_pending_retransmission,
 502                                                          handle)))
 503     {
 504 #if DEBUG_DHT_API
 505       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 506                   "Failed to transmit request to dht service.\n");
 507 #endif
 508       finish_retransmission (handle, GNUNET_SYSERR);
 509       return;
 510     }
 511 #if DEBUG_DHT_API
 512   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 513               "`%s': Scheduled sending message of size %d to service\n",
 514               "DHT API", ntohs (handle->current->msg->size));
 515 #endif
 516 }
 517
 518 /**
 519  * Send complete (or failed), call continuation if we have one.
 520  */
 521 static void
 522 finish_retransmission (struct GNUNET_DHT_Handle *handle, int code)
 523 {
 524   struct PendingMessage *pos = handle->current;
 525   struct PendingMessageList *pending_list;
 526 #if DEBUG_DHT_API
 527   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Finish (retransmission) called!\n", "DHT API");
 528 #endif
 529   GNUNET_assert (pos == handle->retransmissions->message);
 530   pending_list = handle->retransmissions;
 531   handle->retransmissions = handle->retransmissions->next;
 532   GNUNET_free (pending_list);
 533
 534   if (handle->retransmissions == NULL)
 535     {
 536       handle->retransmit_stage = DHT_NOT_RETRANSMITTING;
 537     }
 538
 539   if (handle->retransmissions != NULL)
 540     {
 541       handle->current = handle->retransmissions->message;
 542       process_pending_retransmissions(handle);
 543     }
 544   else if (handle->retransmission_buffer != NULL)
 545     {
 546       handle->current = handle->retransmission_buffer;
 547       process_pending_message(handle);
 548     }
 549 }
 550
 551 /**
 552  * Handler for messages received from the DHT service
 553  * a demultiplexer which handles numerous message types
 554  *
 555  */
 556 void
 557 service_message_handler (void *cls,
 558                          const struct GNUNET_MessageHeader *msg)
 559 {
 560   struct GNUNET_DHT_Handle *handle = cls;
 561   struct GNUNET_DHT_RouteResultMessage *dht_msg;
 562   struct GNUNET_MessageHeader *enc_msg;
 563   struct GNUNET_DHT_RouteHandle *route_handle;
 564   uint64_t uid;
 565   GNUNET_HashCode uid_hash;
 566   size_t enc_size;
 567
 568   if (msg == NULL)
 569     {
 570 #if DEBUG_DHT_API
 571       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 572                   "`%s': Received NULL from server, connection down!\n",
 573                   "DHT API");
 574 #endif
 575       GNUNET_CLIENT_disconnect (handle->client, GNUNET_YES);
 576       handle->client = GNUNET_CLIENT_connect (handle->sched,
 577                                               "dht",
 578                                               handle->cfg);
 579       if (handle->current != NULL)
 580         {
 581           handle->th = NULL;
 582           finish(handle, GNUNET_SYSERR); /* If there was a current message, kill it! */
 583         }
 584 #if RETRANSMIT
 585       if ((handle->retransmit_stage != DHT_RETRANSMITTING) && (GNUNET_CONTAINER_multihashmap_iterate(handle->outstanding_requests, &retransmit_iterator, handle) > 0))
 586         {
 587           handle->retransmit_stage = DHT_RETRANSMITTING;
 588           handle->current = handle->retransmissions->message;
 589           process_pending_retransmissions(handle);
 590         }
 591 #endif
 592       return;
 593     }
 594
 595   switch (ntohs (msg->type))
 596     {
 597     case GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_RESULT:
 598       {
 599         dht_msg = (struct GNUNET_DHT_RouteResultMessage *) msg;
 600         uid = GNUNET_ntohll (dht_msg->unique_id);
 601 #if DEBUG_DHT_API
 602         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 603                     "`%s': Received response to message (uid %llu)\n",
 604                     "DHT API", uid);
 605 #endif
 606
 607         hash_from_uid (uid, &uid_hash);
 608         route_handle =
 609           GNUNET_CONTAINER_multihashmap_get (handle->outstanding_requests,
 610                                              &uid_hash);
 611         if (route_handle == NULL)   /* We have no recollection of this request */
 612           {
 613 #if DEBUG_DHT_API
 614             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 615                         "`%s': Received response to message (uid %llu), but have no recollection of it!\n",
 616                         "DHT API", uid);
 617 #endif
 618           }
 619         else
 620           {
 621             enc_size =
 622               ntohs (dht_msg->header.size) -
 623               sizeof (struct GNUNET_DHT_RouteResultMessage);
 624             GNUNET_assert (enc_size > 0);
 625             enc_msg = (struct GNUNET_MessageHeader *) &dht_msg[1];
 626             route_handle->iter (route_handle->iter_cls, enc_msg);
 627           }
 628
 629         break;
 630       }
 631     default:
 632       {
 633         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 634                     "`%s': Received unknown message type %d\n", "DHT API",
 635                     ntohs (msg->type));
 636       }
 637     }
 638   GNUNET_CLIENT_receive (handle->client,
 639                          &service_message_handler,
 640                          handle, GNUNET_TIME_UNIT_FOREVER_REL);
 641
 642 }
 643
 644
 645 /**
 646  * Initialize the connection with the DHT service.
 647  *
 648  * @param sched scheduler to use
 649  * @param cfg configuration to use
 650  * @param ht_len size of the internal hash table to use for
 651  *               processing multiple GET/FIND requests in parallel
 652  *
 653  * @return handle to the DHT service, or NULL on error
 654  */
 655 struct GNUNET_DHT_Handle *
 656 GNUNET_DHT_connect (struct GNUNET_SCHEDULER_Handle *sched,
 657                     const struct GNUNET_CONFIGURATION_Handle *cfg,
 658                     unsigned int ht_len)
 659 {
 660   struct GNUNET_DHT_Handle *handle;
 661
 662   handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_Handle));
 663   handle->cfg = cfg;
 664   handle->sched = sched;
 665   handle->client = GNUNET_CLIENT_connect (sched, "dht", cfg);
 666   handle->uid_gen = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, -1);
 667   if (handle->client == NULL)
 668     {
 669       GNUNET_free (handle);
 670       return NULL;
 671     }
 672   handle->outstanding_requests =
 673     GNUNET_CONTAINER_multihashmap_create (ht_len);
 674 #if DEBUG_DHT_API
 675   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 676               "`%s': Connection to service in progress\n", "DHT API");
 677 #endif
 678   GNUNET_CLIENT_receive (handle->client,
 679                          &service_message_handler,
 680                          handle, GNUNET_TIME_UNIT_FOREVER_REL);
 681   return handle;
 682 }
 683
 684
 685 /**
 686  * Shutdown connection with the DHT service.
 687  *
 688  * @param handle handle of the DHT connection to stop
 689  */
 690 void
 691 GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle)
 692 {
 693 #if DEBUG_DHT_API
 694   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 695               "`%s': Called GNUNET_DHT_disconnect\n", "DHT API");
 696 #endif
 697   GNUNET_assert (handle != NULL);
 698   if (handle->th != NULL)       /* We have a live transmit request */
 699     {
 700       GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
 701       handle->th = NULL;
 702     }
 703   if (handle->current != NULL)  /* We are trying to send something now, clean it up */
 704     GNUNET_free (handle->current);
 705
 706   if (handle->client != NULL)   /* Finally, disconnect from the service */
 707     {
 708       GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
 709       handle->client = NULL;
 710     }
 711
 712   GNUNET_assert(GNUNET_CONTAINER_multihashmap_size(handle->outstanding_requests) == 0);
 713   GNUNET_CONTAINER_multihashmap_destroy(handle->outstanding_requests);
 714   GNUNET_free (handle);
 715 }
 716
 717
 718 /**
 719  * Transmit the next pending message, called by notify_transmit_ready
 720  */
 721 static size_t
 722 transmit_pending_retransmission (void *cls, size_t size, void *buf)
 723 {
 724   struct GNUNET_DHT_Handle *handle = cls;
 725   size_t tsize;
 726
 727 #if DEBUG_DHT_API
 728   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 729               "`%s': In transmit_pending\n", "DHT API");
 730 #endif
 731   if (buf == NULL)
 732     {
 733 #if DEBUG_DHT_API
 734       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 735                   "`%s': In transmit_pending buf is NULL\n", "DHT API");
 736 #endif
 737       finish_retransmission (handle, GNUNET_SYSERR);
 738       return 0;
 739     }
 740
 741   handle->th = NULL;
 742
 743   if (handle->current != NULL)
 744     {
 745       tsize = ntohs (handle->current->msg->size);
 746       if (size >= tsize)
 747         {
 748 #if DEBUG_DHT_API
 749           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 750                       "`%s': Sending message size %d\n", "DHT API", tsize);
 751 #endif
 752           memcpy (buf, handle->current->msg, tsize);
 753           finish_retransmission (handle, GNUNET_OK);
 754           return tsize;
 755         }
 756       else
 757         {
 758           return 0;
 759         }
 760     }
 761   /* Have no pending request */
 762   return 0;
 763 }
 764
 765
 766 /**
 767  * Iterator called on each result obtained from a generic route
 768  * operation
 769  */
 770 void
 771 get_reply_iterator (void *cls, const struct GNUNET_MessageHeader *reply)
 772 {
 773   struct GNUNET_DHT_GetHandle *get_handle = cls;
 774   struct GNUNET_DHT_GetResultMessage *result;
 775   size_t data_size;
 776   char *result_data;
 777
 778   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_GET_RESULT)
 779     return;
 780
 781   GNUNET_assert (ntohs (reply->size) >=
 782                  sizeof (struct GNUNET_DHT_GetResultMessage));
 783   result = (struct GNUNET_DHT_GetResultMessage *) reply;
 784   data_size = ntohs (reply->size) - sizeof(struct GNUNET_DHT_GetResultMessage);
 785
 786   result_data = (char *) &result[1];    /* Set data pointer to end of message */
 787
 788   get_handle->get_context.iter (get_handle->get_context.iter_cls,
 789                                 GNUNET_TIME_absolute_ntoh (result->expiration), &get_handle->route_handle->key,
 790                                 ntohs (result->type), data_size, result_data);
 791 }
 792
 793
 794 /**
 795  * Iterator called on each result obtained from a generic route
 796  * operation
 797  */
 798 void
 799 find_peer_reply_iterator (void *cls, const struct GNUNET_MessageHeader *reply)
 800 {
 801   struct GNUNET_DHT_FindPeerHandle *find_peer_handle = cls;
 802   struct GNUNET_MessageHeader *hello;
 803
 804   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_FIND_PEER_RESULT)
 805     {
 806       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 807                   "Received wrong type of response to a find peer request...\n");
 808       return;
 809     }
 810
 811
 812   GNUNET_assert (ntohs (reply->size) >=
 813                  sizeof (struct GNUNET_MessageHeader));
 814   hello = (struct GNUNET_MessageHeader *)&reply[1];
 815
 816   if (ntohs(hello->type) != GNUNET_MESSAGE_TYPE_HELLO)
 817     {
 818       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 819                   "Encapsulated message of type %d, is not a `%s' message!\n", ntohs(hello->type), "HELLO");
 820       return;
 821     }
 822   find_peer_handle->find_peer_context.proc (find_peer_handle->
 823                                             find_peer_context.proc_cls,
 824                                             (struct GNUNET_HELLO_Message *)hello);
 825 }
 826
 827 /**
 828  * Send a message to the DHT telling it to start issuing random GET
 829  * requests every 'frequency' milliseconds.
 830  *
 831  * @param handle handle to the DHT service
 832  * @param frequency delay (in milliseconds) between sending malicious messages
 833  * @param cont continuation to call once the message is sent
 834  * @param cont_cls closure for continuation
 835  *
 836  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 837  */
 838 int GNUNET_DHT_set_malicious_getter (struct GNUNET_DHT_Handle *handle, int frequency, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 839 {
 840   struct GNUNET_DHT_ControlMessage *msg;
 841   struct PendingMessage *pending;
 842
 843   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 844     return GNUNET_NO;
 845
 846   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 847   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 848   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 849   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_GET);
 850   msg->variable = htons(frequency);
 851
 852   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 853   pending->msg = &msg->header;
 854   pending->timeout = GNUNET_TIME_relative_get_forever();
 855   pending->free_on_send = GNUNET_YES;
 856   pending->cont = cont;
 857   pending->cont_cls = cont_cls;
 858   pending->unique_id = 0;
 859
 860   if (handle->current == NULL)
 861     {
 862       handle->current = pending;
 863       process_pending_message (handle);
 864     }
 865   else
 866   {
 867     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 868     handle->retransmission_buffer = pending;
 869   }
 870
 871   return GNUNET_YES;
 872 }
 873
 874 /**
 875  * Send a message to the DHT telling it to start issuing random PUT
 876  * requests every 'frequency' milliseconds.
 877  *
 878  * @param handle handle to the DHT service
 879  * @param frequency delay (in milliseconds) between sending malicious messages
 880  * @param cont continuation to call once the message is sent
 881  * @param cont_cls closure for continuation
 882  *
 883  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 884  */
 885 int GNUNET_DHT_set_malicious_putter (struct GNUNET_DHT_Handle *handle, int frequency, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 886 {
 887   struct GNUNET_DHT_ControlMessage *msg;
 888   struct PendingMessage *pending;
 889
 890   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 891     return GNUNET_NO;
 892
 893   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 894   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 895   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 896   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_PUT);
 897   msg->variable = htons(frequency);
 898
 899   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 900   pending->msg = &msg->header;
 901   pending->timeout = GNUNET_TIME_relative_get_forever();
 902   pending->free_on_send = GNUNET_YES;
 903   pending->cont = cont;
 904   pending->cont_cls = cont_cls;
 905   pending->unique_id = 0;
 906
 907   if (handle->current == NULL)
 908     {
 909       handle->current = pending;
 910       process_pending_message (handle);
 911     }
 912   else
 913   {
 914     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 915     handle->retransmission_buffer = pending;
 916   }
 917
 918   return GNUNET_YES;
 919 }
 920
 921 /**
 922  * Send a message to the DHT telling it to start dropping
 923  * all requests received.
 924  *
 925  * @param handle handle to the DHT service
 926  * @param cont continuation to call once the message is sent
 927  * @param cont_cls closure for continuation
 928  *
 929  * @return GNUNET_YES if the control message was sent, GNUNET_NO if not
 930  */
 931 int GNUNET_DHT_set_malicious_dropper (struct GNUNET_DHT_Handle *handle, GNUNET_SCHEDULER_Task cont, void *cont_cls)
 932 {
 933   struct GNUNET_DHT_ControlMessage *msg;
 934   struct PendingMessage *pending;
 935
 936   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
 937     return GNUNET_NO;
 938
 939   msg = GNUNET_malloc(sizeof(struct GNUNET_DHT_ControlMessage));
 940   msg->header.size = htons(sizeof(struct GNUNET_DHT_ControlMessage));
 941   msg->header.type = htons(GNUNET_MESSAGE_TYPE_DHT_CONTROL);
 942   msg->command = htons(GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_DROP);
 943   msg->variable = htons(0);
 944
 945   pending = GNUNET_malloc (sizeof (struct PendingMessage));
 946   pending->msg = &msg->header;
 947   pending->timeout = GNUNET_TIME_relative_get_forever();
 948   pending->free_on_send = GNUNET_YES;
 949   pending->cont = cont;
 950   pending->cont_cls = cont_cls;
 951   pending->unique_id = 0;
 952
 953   if (handle->current == NULL)
 954     {
 955       handle->current = pending;
 956       process_pending_message (handle);
 957     }
 958   else
 959   {
 960     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
 961     handle->retransmission_buffer = pending;
 962   }
 963
 964   return GNUNET_YES;
 965 }
 966
 967
 968 /**
 969  * Initiate a generic DHT route operation.
 970  *
 971  * @param handle handle to the DHT service
 972  * @param key the key to look up
 973  * @param desired_replication_level how many peers should ultimately receive
 974  *                this message (advisory only, target may be too high for the
 975  *                given DHT or not hit exactly).
 976  * @param options options for routing
 977  * @param enc send the encapsulated message to a peer close to the key
 978  * @param iter function to call on each result, NULL if no replies are expected
 979  * @param iter_cls closure for iter
 980  * @param timeout when to abort with an error if we fail to get
 981  *                a confirmation for the request (when necessary) or how long
 982  *                to wait for tramission to the service
 983  * @param cont continuation to call when done;
 984  *             reason will be TIMEOUT on error,
 985  *             reason will be PREREQ_DONE on success
 986  * @param cont_cls closure for cont
 987  *
 988  * @return handle to stop the request, NULL if the request is "fire and forget"
 989  */
 990 struct GNUNET_DHT_RouteHandle *
 991 GNUNET_DHT_route_start (struct GNUNET_DHT_Handle *handle,
 992                         const GNUNET_HashCode * key,
 993                         unsigned int desired_replication_level,
 994                         enum GNUNET_DHT_RouteOption options,
 995                         const struct GNUNET_MessageHeader *enc,
 996                         struct GNUNET_TIME_Relative timeout,
 997                         GNUNET_DHT_ReplyProcessor iter,
 998                         void *iter_cls,
 999                         GNUNET_SCHEDULER_Task cont, void *cont_cls)
1000 {
1001   struct GNUNET_DHT_RouteHandle *route_handle;
1002   struct PendingMessage *pending;
1003   struct GNUNET_DHT_RouteMessage *message;
1004   uint16_t msize;
1005   GNUNET_HashCode uid_key;
1006
1007   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
1008     return NULL;
1009
1010   if (sizeof (struct GNUNET_DHT_RouteMessage) + ntohs (enc->size) >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1011     {
1012       GNUNET_break (0);
1013       return NULL;
1014     }
1015
1016   route_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_RouteHandle));
1017   memcpy (&route_handle->key, key, sizeof (GNUNET_HashCode));
1018   route_handle->iter = iter;
1019   route_handle->iter_cls = iter_cls;
1020   route_handle->dht_handle = handle;
1021   route_handle->uid = handle->uid_gen++;
1022   if (iter != NULL)
1023     {
1024       hash_from_uid (route_handle->uid, &uid_key);
1025       GNUNET_CONTAINER_multihashmap_put (handle->outstanding_requests,
1026                                          &uid_key, route_handle,
1027                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1028     }
1029
1030 #if DEBUG_DHT_API
1031       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1032                   "`%s': Unique ID is %llu\n", "DHT API", route_handle->uid);
1033 #endif
1034
1035   msize = sizeof (struct GNUNET_DHT_RouteMessage) + ntohs (enc->size);
1036   message = GNUNET_malloc (msize);
1037   message->header.size = htons (msize);
1038   message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE);
1039   memcpy (&message->key, key, sizeof (GNUNET_HashCode));
1040   message->options = htonl (options);
1041   message->desired_replication_level = htonl (options);
1042   message->unique_id = GNUNET_htonll (route_handle->uid);
1043   memcpy (&message[1], enc, ntohs (enc->size));
1044   pending = GNUNET_malloc (sizeof (struct PendingMessage));
1045   pending->msg = &message->header;
1046   pending->timeout = timeout;
1047   if (iter == NULL)
1048     pending->free_on_send = GNUNET_YES;
1049   pending->cont = cont;
1050   pending->cont_cls = cont_cls;
1051   pending->unique_id = route_handle->uid;
1052   if (handle->current == NULL)
1053     {
1054       handle->current = pending;
1055       process_pending_message (handle);
1056     }
1057   else
1058   {
1059     handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
1060     handle->retransmission_buffer = pending;
1061   }
1062
1063   route_handle->message = message;
1064   return route_handle;
1065 }
1066
1067
1068 /**
1069  * Perform an asynchronous GET operation on the DHT identified.
1070  *
1071  * @param handle handle to the DHT service
1072  * @param timeout how long to wait for transmission of this request to the service
1073  * @param type expected type of the response object
1074  * @param key the key to look up
1075  * @param iter function to call on each result
1076  * @param iter_cls closure for iter
1077  * @param cont continuation to call once message sent
1078  * @param cont_cls closure for continuation
1079  *
1080  * @return handle to stop the async get
1081  */
1082 struct GNUNET_DHT_GetHandle *
1083 GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
1084                       struct GNUNET_TIME_Relative timeout,
1085                       uint32_t type,
1086                       const GNUNET_HashCode * key,
1087                       GNUNET_DHT_GetIterator iter,
1088                       void *iter_cls,
1089                       GNUNET_SCHEDULER_Task cont, void *cont_cls)
1090 {
1091   struct GNUNET_DHT_GetHandle *get_handle;
1092   struct GNUNET_DHT_GetMessage get_msg;
1093
1094   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING)) /* Can't send right now, we have a pending message... */
1095     return NULL;
1096
1097   get_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle));
1098   get_handle->get_context.iter = iter;
1099   get_handle->get_context.iter_cls = iter_cls;
1100
1101 #if DEBUG_DHT_API
1102   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1103               "`%s': Inserting pending get request with key %s\n", "DHT API",
1104               GNUNET_h2s (key));
1105 #endif
1106
1107   get_msg.header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET);
1108   get_msg.header.size = htons (sizeof (struct GNUNET_DHT_GetMessage));
1109   get_msg.type = htons (type);
1110
1111   get_handle->route_handle =
1112     GNUNET_DHT_route_start (handle, key, 0, 0, &get_msg.header, timeout,
1113                             &get_reply_iterator, get_handle, cont, cont_cls);
1114
1115   return get_handle;
1116 }
1117
1118
1119 /**
1120  * Stop a previously issued routing request
1121  *
1122  * @param route_handle handle to the request to stop
1123  * @param cont continuation to call once this message is sent to the service or times out
1124  * @param cont_cls closure for the continuation
1125  */
1126 void
1127 GNUNET_DHT_route_stop (struct GNUNET_DHT_RouteHandle *route_handle,
1128                        GNUNET_SCHEDULER_Task cont, void *cont_cls)
1129 {
1130   struct PendingMessage *pending;
1131   struct GNUNET_DHT_StopMessage *message;
1132   size_t msize;
1133   GNUNET_HashCode uid_key;
1134
1135   msize = sizeof (struct GNUNET_DHT_StopMessage);
1136   message = GNUNET_malloc (msize);
1137   message->header.size = htons (msize);
1138   message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_STOP);
1139 #if DEBUG_DHT_API
1140   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1141               "`%s': Remove outstanding request for uid %llu\n", "DHT API",
1142               route_handle->uid);
1143 #endif
1144   message->unique_id = GNUNET_htonll (route_handle->uid);
1145   memcpy(&message->key, &route_handle->key, sizeof(GNUNET_HashCode));
1146   pending = GNUNET_malloc (sizeof (struct PendingMessage));
1147   pending->msg = (struct GNUNET_MessageHeader *) message;
1148   pending->timeout = GNUNET_TIME_relative_get_forever();
1149   pending->cont = cont;
1150   pending->cont_cls = cont_cls;
1151   pending->free_on_send = GNUNET_YES;
1152   pending->unique_id = 0; /* When finished is called, free pending->msg */
1153
1154   if (route_handle->dht_handle->current == NULL)
1155     {
1156       route_handle->dht_handle->current = pending;
1157       process_pending_message (route_handle->dht_handle);
1158     }
1159   else if (route_handle->dht_handle->retransmit_stage == DHT_RETRANSMITTING)
1160     {
1161       route_handle->dht_handle->retransmit_stage = DHT_RETRANSMITTING_MESSAGE_QUEUED;
1162       route_handle->dht_handle->retransmission_buffer = pending;
1163     }
1164   else
1165     {
1166       GNUNET_free(pending);
1167       GNUNET_break(0);
1168     }
1169
1170   hash_from_uid (route_handle->uid, &uid_key);
1171   GNUNET_assert (GNUNET_CONTAINER_multihashmap_remove
1172                  (route_handle->dht_handle->outstanding_requests, &uid_key,
1173                   route_handle) == GNUNET_YES);
1174
1175   GNUNET_free(route_handle->message);
1176   GNUNET_free(route_handle);
1177 }
1178
1179
1180 /**
1181  * Stop async DHT-get.
1182  *
1183  * @param get_handle handle to the GET operation to stop
1184  * @param cont continuation to call once this message is sent to the service or times out
1185  * @param cont_cls closure for the continuation
1186  */
1187 void
1188 GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle,
1189                      GNUNET_SCHEDULER_Task cont, void *cont_cls)
1190 {
1191   if ((get_handle->route_handle->dht_handle->current != NULL) &&
1192       (get_handle->route_handle->dht_handle->retransmit_stage != DHT_RETRANSMITTING))
1193     {
1194       if (cont != NULL)
1195         {
1196           GNUNET_SCHEDULER_add_continuation (get_handle->route_handle->dht_handle->sched, cont, cont_cls,
1197                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1198         }
1199       return;
1200     }
1201
1202 #if DEBUG_DHT_API
1203   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1204               "`%s': Removing pending get request with key %s, uid %llu\n",
1205               "DHT API", GNUNET_h2s (&get_handle->route_handle->key),
1206               get_handle->route_handle->uid);
1207 #endif
1208   GNUNET_DHT_route_stop (get_handle->route_handle, cont, cont_cls);
1209   GNUNET_free (get_handle);
1210 }
1211
1212
1213 /**
1214  * Perform an asynchronous FIND PEER operation on the DHT.
1215  *
1216  * @param handle handle to the DHT service
1217  * @param timeout timeout for this request to be sent to the
1218  *        service
1219  * @param options routing options for this message
1220  * @param key the key to look up
1221  * @param proc function to call on each result
1222  * @param proc_cls closure for proc
1223  * @param cont continuation to call once message sent
1224  * @param cont_cls closure for continuation
1225  *
1226  * @return handle to stop the async get, NULL on error
1227  */
1228 struct GNUNET_DHT_FindPeerHandle *
1229 GNUNET_DHT_find_peer_start (struct GNUNET_DHT_Handle *handle,
1230                             struct GNUNET_TIME_Relative timeout,
1231                             enum GNUNET_DHT_RouteOption options,
1232                             const GNUNET_HashCode * key,
1233                             GNUNET_DHT_FindPeerProcessor proc,
1234                             void *proc_cls,
1235                             GNUNET_SCHEDULER_Task cont,
1236                             void *cont_cls)
1237 {
1238   struct GNUNET_DHT_FindPeerHandle *find_peer_handle;
1239   struct GNUNET_DHT_FindPeerMessage find_peer_msg;
1240
1241   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))  /* Can't send right now, we have a pending message... */
1242     return NULL;
1243
1244   find_peer_handle =
1245     GNUNET_malloc (sizeof (struct GNUNET_DHT_FindPeerHandle));
1246   find_peer_handle->find_peer_context.proc = proc;
1247   find_peer_handle->find_peer_context.proc_cls = proc_cls;
1248
1249 #if DEBUG_DHT_API
1250   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1251               "`%s': Inserting pending `%s' request with key %s\n", "DHT API",
1252               "FIND PEER", GNUNET_h2s (key));
1253 #endif
1254
1255   find_peer_msg.header.size = htons(sizeof(struct GNUNET_DHT_FindPeerMessage));
1256   find_peer_msg.header.type = htons(GNUNET_MESSAGE_TYPE_DHT_FIND_PEER);
1257   find_peer_handle->route_handle =
1258     GNUNET_DHT_route_start (handle, key, 0, options, &find_peer_msg.header,
1259                             timeout, &find_peer_reply_iterator,
1260                             find_peer_handle, cont, cont_cls);
1261   return find_peer_handle;
1262 }
1263
1264 /**
1265  * Stop async find peer.  Frees associated resources.
1266  *
1267  * @param find_peer_handle GET operation to stop.
1268  * @param cont continuation to call once this message is sent to the service or times out
1269  * @param cont_cls closure for the continuation
1270  */
1271 void
1272 GNUNET_DHT_find_peer_stop (struct GNUNET_DHT_FindPeerHandle *find_peer_handle,
1273                            GNUNET_SCHEDULER_Task cont, void *cont_cls)
1274 {
1275   if ((find_peer_handle->route_handle->dht_handle->current != NULL) &&
1276       (find_peer_handle->route_handle->dht_handle->retransmit_stage != DHT_RETRANSMITTING))
1277     {
1278       if (cont != NULL)
1279         {
1280           GNUNET_SCHEDULER_add_continuation (find_peer_handle->route_handle->dht_handle->sched, cont, cont_cls,
1281                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1282         }
1283       return;
1284     }
1285
1286 #if DEBUG_DHT_API
1287   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1288               "`%s': Removing pending `%s' request with key %s, uid %llu\n",
1289               "DHT API", "FIND PEER",
1290               GNUNET_h2s (&find_peer_handle->route_handle->key),
1291               find_peer_handle->route_handle->uid);
1292 #endif
1293   GNUNET_DHT_route_stop (find_peer_handle->route_handle, cont, cont_cls);
1294   GNUNET_free (find_peer_handle);
1295
1296 }
1297
1298
1299 /**
1300  * Perform a PUT operation storing data in the DHT.
1301  *
1302  * @param handle handle to DHT service
1303  * @param key the key to store under
1304  * @param type type of the value
1305  * @param size number of bytes in data; must be less than 64k
1306  * @param data the data to store
1307  * @param exp desired expiration time for the value
1308  * @param timeout how long to wait for transmission of this request
1309  * @param cont continuation to call when done;
1310  *             reason will be TIMEOUT on error,
1311  *             reason will be PREREQ_DONE on success
1312  * @param cont_cls closure for cont
1313  *
1314  * @return GNUNET_YES if put message is queued for transmission
1315  */
1316 void
1317 GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle,
1318                 const GNUNET_HashCode * key,
1319                 uint32_t type,
1320                 uint32_t size,
1321                 const char *data,
1322                 struct GNUNET_TIME_Absolute exp,
1323                 struct GNUNET_TIME_Relative timeout,
1324                 GNUNET_SCHEDULER_Task cont, void *cont_cls)
1325 {
1326   struct GNUNET_DHT_PutMessage *put_msg;
1327   struct GNUNET_DHT_RouteHandle *put_route;
1328   size_t msize;
1329
1330   if ((handle->current != NULL) && (handle->retransmit_stage != DHT_RETRANSMITTING))
1331     {
1332       GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "handle->current is not null!\n");
1333       if (cont != NULL)
1334         {
1335           GNUNET_SCHEDULER_add_continuation (handle->sched, cont, cont_cls,
1336                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1337         }
1338       return;
1339     }
1340
1341 #if DEBUG_DHT_API
1342   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1343               "`%s': Inserting pending put request with key %s\n", "DHT API",
1344               GNUNET_h2s (key));
1345 #endif
1346
1347   msize = sizeof (struct GNUNET_DHT_PutMessage) + size;
1348   put_msg = GNUNET_malloc (msize);
1349   put_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_PUT);
1350   put_msg->header.size = htons (msize);
1351   put_msg->type = htons (type);
1352   put_msg->data_size = htons (size);
1353   put_msg->expiration = GNUNET_TIME_absolute_hton(exp);
1354   memcpy (&put_msg[1], data, size);
1355
1356   put_route = GNUNET_DHT_route_start (handle, key, 0, 0, &put_msg->header, timeout, NULL,
1357                                       NULL, cont, cont_cls);
1358
1359   if (put_route == NULL) /* Route start failed! */
1360     {
1361       GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "route start for PUT failed!\n");
1362       if (cont != NULL)
1363         {
1364           GNUNET_SCHEDULER_add_continuation (handle->sched, cont, cont_cls,
1365                                              GNUNET_SCHEDULER_REASON_TIMEOUT);
1366         }
1367     }
1368   else
1369     {
1370       GNUNET_free(put_route);
1371     }
1372
1373   GNUNET_free (put_msg);
1374 }