src/dht/gnunet-service-dht_clients.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 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/gnunet-service-dht_clients.c
  23  * @brief GNUnet DHT service's client management code
  24  * @author Christian Grothoff
  25  * @author Nathan Evans
  26  */
  27
  28 #include "platform.h"
  29 #include "gnunet_constants.h"
  30 #include "gnunet_protocols.h"
  31 #include "gnunet_statistics_service.h"
  32 #include "gnunet-service-dht.h"
  33 #include "gnunet-service-dht_clients.h"
  34 #include "gnunet-service-dht_datacache.h"
  35 #include "gnunet-service-dht_neighbours.h"
  36 #include "dht.h"
  37
  38
  39 /**
  40  * Linked list of messages to send to clients.
  41  */
  42 struct PendingMessage
  43 {
  44   /**
  45    * Pointer to next item in the list
  46    */
  47   struct PendingMessage *next;
  48
  49   /**
  50    * Pointer to previous item in the list
  51    */
  52   struct PendingMessage *prev;
  53
  54   /**
  55    * Actual message to be sent, allocated at the end of the struct:
  56    * // msg = (cast) &pm[1];
  57    * // memcpy (&pm[1], data, len);
  58    */
  59   const struct GNUNET_MessageHeader *msg;
  60
  61 };
  62
  63
  64 /**
  65  * Struct containing information about a client,
  66  * handle to connect to it, and any pending messages
  67  * that need to be sent to it.
  68  */
  69 struct ClientList
  70 {
  71   /**
  72    * Linked list of active clients
  73    */
  74   struct ClientList *next;
  75
  76   /**
  77    * Linked list of active clients
  78    */
  79   struct ClientList *prev;
  80
  81   /**
  82    * The handle to this client
  83    */
  84   struct GNUNET_SERVER_Client *client_handle;
  85
  86   /**
  87    * Handle to the current transmission request, NULL
  88    * if none pending.
  89    */
  90   struct GNUNET_SERVER_TransmitHandle *transmit_handle;
  91
  92   /**
  93    * Linked list of pending messages for this client
  94    */
  95   struct PendingMessage *pending_head;
  96
  97   /**
  98    * Tail of linked list of pending messages for this client
  99    */
 100   struct PendingMessage *pending_tail;
 101
 102 };
 103
 104
 105 /**
 106  * Entry in the DHT routing table for a client's GET request.
 107  */
 108 struct ClientQueryRecord
 109 {
 110
 111   /**
 112    * The key this request was about
 113    */
 114   GNUNET_HashCode key;
 115
 116   /**
 117    * Client responsible for the request.
 118    */
 119   struct ClientList *client;
 120
 121   /**
 122    * Extended query (see gnunet_block_lib.h), allocated at the end of this struct.
 123    */
 124   const void *xquery;
 125
 126   /**
 127    * Replies we have already seen for this request.
 128    */
 129   GNUNET_HashCode *seen_replies;
 130
 131   /**
 132    * Pointer to this nodes heap location in the retry-heap (for fast removal)
 133    */
 134   struct GNUNET_CONTAINER_HeapNode *hnode;
 135
 136   /**
 137    * What's the delay between re-try operations that we currently use for this
 138    * request?
 139    */
 140   struct GNUNET_TIME_Relative retry_frequency;
 141
 142   /**
 143    * What's the next time we should re-try this request?
 144    */
 145   struct GNUNET_TIME_Absolute retry_time;
 146
 147   /**
 148    * The unique identifier of this request
 149    */
 150   uint64_t unique_id;
 151
 152   /**
 153    * Number of bytes in xquery.
 154    */
 155   size_t xquery_size;
 156
 157   /**
 158    * Number of entries in 'seen_replies'.
 159    */
 160   unsigned int seen_replies_count;
 161
 162   /**
 163    * Desired replication level
 164    */
 165   uint32_t replication;
 166
 167   /**
 168    * Any message options for this request
 169    */
 170   uint32_t msg_options;
 171
 172   /**
 173    * The type for the data for the GET request.
 174    */
 175   enum GNUNET_BLOCK_Type type;
 176
 177 };
 178
 179
 180 /**
 181  * Struct containing paremeters of monitoring requests.
 182  */
 183 struct ClientMonitorRecord
 184 {
 185
 186   /**
 187    * Next element in DLL.
 188    */
 189   struct ClientMonitorRecord    *next;
 190
 191   /**
 192    * Previous element in DLL.
 193    */
 194   struct ClientMonitorRecord    *prev;
 195
 196   /**
 197    * Type of blocks that are of interest
 198    */
 199   enum GNUNET_BLOCK_Type        type;
 200
 201   /**
 202    * Key of data of interest, NULL for all.
 203    */
 204   GNUNET_HashCode         *key;
 205
 206   /**
 207    * Flag whether to notify about GET messages.
 208    */
 209   int16_t get;
 210
 211   /**
 212    * Flag whether to notify about GET_REPONSE messages.
 213    */
 214   int16_t get_resp;
 215
 216   /**
 217    * Flag whether to notify about PUT messages.
 218    */
 219   uint16_t put;
 220
 221   /**
 222    * Client to notify of these requests.
 223    */
 224   struct ClientList             *client;
 225 };
 226
 227
 228 /**
 229  * List of active clients.
 230  */
 231 static struct ClientList *client_head;
 232
 233 /**
 234  * List of active clients.
 235  */
 236 static struct ClientList *client_tail;
 237
 238 /**
 239  * List of active monitoring requests.
 240  */
 241 static struct ClientMonitorRecord *monitor_head;
 242
 243 /**
 244  * List of active monitoring requests.
 245  */
 246 static struct ClientMonitorRecord *monitor_tail;
 247
 248 /**
 249  * Hashmap for fast key based lookup, maps keys to 'struct ClientQueryRecord' entries.
 250  */
 251 static struct GNUNET_CONTAINER_MultiHashMap *forward_map;
 252
 253 /**
 254  * Heap with all of our client's request, sorted by retry time (earliest on top).
 255  */
 256 static struct GNUNET_CONTAINER_Heap *retry_heap;
 257
 258 /**
 259  * Task that re-transmits requests (using retry_heap).
 260  */
 261 static GNUNET_SCHEDULER_TaskIdentifier retry_task;
 262
 263
 264 /**
 265  * Find a client if it exists, add it otherwise.
 266  *
 267  * @param client the server handle to the client
 268  *
 269  * @return the client if found, a new client otherwise
 270  */
 271 static struct ClientList *
 272 find_active_client (struct GNUNET_SERVER_Client *client)
 273 {
 274   struct ClientList *pos = client_head;
 275   struct ClientList *ret;
 276
 277   while (pos != NULL)
 278   {
 279     if (pos->client_handle == client)
 280       return pos;
 281     pos = pos->next;
 282   }
 283   ret = GNUNET_malloc (sizeof (struct ClientList));
 284   ret->client_handle = client;
 285   GNUNET_CONTAINER_DLL_insert (client_head, client_tail, ret);
 286   return ret;
 287 }
 288
 289
 290 /**
 291  * Iterator over hash map entries that frees all entries
 292  * associated with the given client.
 293  *
 294  * @param cls client to search for in source routes
 295  * @param key current key code (ignored)
 296  * @param value value in the hash map, a ClientQueryRecord
 297  * @return GNUNET_YES (we should continue to iterate)
 298  */
 299 static int
 300 remove_client_records (void *cls, const GNUNET_HashCode * key, void *value)
 301 {
 302   struct ClientList *client = cls;
 303   struct ClientQueryRecord *record = value;
 304
 305   if (record->client != client)
 306     return GNUNET_YES;
 307 #if DEBUG_DHT
 308   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 309               "Removing client %p's record for key %s\n", client,
 310               GNUNET_h2s (key));
 311 #endif
 312   GNUNET_assert (GNUNET_YES ==
 313                  GNUNET_CONTAINER_multihashmap_remove (forward_map, key,
 314                                                        record));
 315   if (NULL != record->hnode)
 316     GNUNET_CONTAINER_heap_remove_node (record->hnode);
 317   GNUNET_array_grow (record->seen_replies, record->seen_replies_count, 0);
 318   GNUNET_free (record);
 319   return GNUNET_YES;
 320 }
 321
 322
 323 /**
 324  * Functions with this signature are called whenever a client
 325  * is disconnected on the network level.
 326  *
 327  * @param cls closure (NULL for dht)
 328  * @param client identification of the client; NULL
 329  *        for the last call when the server is destroyed
 330  */
 331 static void
 332 handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 333 {
 334   struct ClientList *pos;
 335   struct PendingMessage *reply;
 336   struct ClientMonitorRecord *monitor;
 337
 338 #if DEBUG_DHT
 339   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Local client %p disconnects\n", client);
 340 #endif
 341   pos = find_active_client (client);
 342   GNUNET_CONTAINER_DLL_remove (client_head, client_tail, pos);
 343   if (pos->transmit_handle != NULL)
 344     GNUNET_SERVER_notify_transmit_ready_cancel (pos->transmit_handle);
 345   while (NULL != (reply = pos->pending_head))
 346   {
 347     GNUNET_CONTAINER_DLL_remove (pos->pending_head, pos->pending_tail, reply);
 348     GNUNET_free (reply);
 349   }
 350   monitor = monitor_head;
 351   while (NULL != monitor)
 352   {
 353     if (monitor->client == pos)
 354     {
 355       struct ClientMonitorRecord *next;
 356
 357       GNUNET_free_non_null (monitor->key);
 358       next = monitor->next;
 359       GNUNET_CONTAINER_DLL_remove (monitor_head, monitor_tail, monitor);
 360       GNUNET_free (monitor);
 361       monitor = next;
 362     }
 363     else
 364       monitor = monitor->next;
 365   }
 366   GNUNET_CONTAINER_multihashmap_iterate (forward_map, &remove_client_records,
 367                                          pos);
 368   GNUNET_free (pos);
 369 }
 370
 371
 372 /**
 373  * Route the given request via the DHT.  This includes updating
 374  * the bloom filter and retransmission times, building the P2P
 375  * message and initiating the routing operation.
 376  */
 377 static void
 378 transmit_request (struct ClientQueryRecord *cqr)
 379 {
 380   int32_t reply_bf_mutator;
 381   struct GNUNET_CONTAINER_BloomFilter *reply_bf;
 382   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 383
 384   GNUNET_STATISTICS_update (GDS_stats,
 385                             gettext_noop
 386                             ("# GET requests from clients injected"), 1,
 387                             GNUNET_NO);
 388   reply_bf_mutator =
 389       (int32_t) GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 390                                           UINT32_MAX);
 391   reply_bf =
 392       GNUNET_BLOCK_construct_bloomfilter (reply_bf_mutator, cqr->seen_replies,
 393                                           cqr->seen_replies_count);
 394   peer_bf =
 395       GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE,
 396                                          GNUNET_CONSTANTS_BLOOMFILTER_K);
 397   GDS_NEIGHBOURS_handle_get (cqr->type, cqr->msg_options, cqr->replication,
 398                              0 /* hop count */ ,
 399                              &cqr->key, cqr->xquery, cqr->xquery_size, reply_bf,
 400                              reply_bf_mutator, peer_bf);
 401   GNUNET_CONTAINER_bloomfilter_free (reply_bf);
 402   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 403
 404   /* exponential back-off for retries, max 1h */
 405   cqr->retry_frequency =
 406       GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS,
 407                                 GNUNET_TIME_relative_multiply
 408                                 (cqr->retry_frequency, 2));
 409   cqr->retry_time = GNUNET_TIME_relative_to_absolute (cqr->retry_frequency);
 410 }
 411
 412
 413 /**
 414  * Task that looks at the 'retry_heap' and transmits all of the requests
 415  * on the heap that are ready for transmission.  Then re-schedules
 416  * itself (unless the heap is empty).
 417  *
 418  * @param cls unused
 419  * @param tc scheduler context
 420  */
 421 static void
 422 transmit_next_request_task (void *cls,
 423                             const struct GNUNET_SCHEDULER_TaskContext *tc)
 424 {
 425   struct ClientQueryRecord *cqr;
 426   struct GNUNET_TIME_Relative delay;
 427
 428   retry_task = GNUNET_SCHEDULER_NO_TASK;
 429   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 430     return;
 431   while (NULL != (cqr = GNUNET_CONTAINER_heap_remove_root (retry_heap)))
 432   {
 433     cqr->hnode = NULL;
 434     delay = GNUNET_TIME_absolute_get_remaining (cqr->retry_time);
 435     if (delay.rel_value > 0)
 436     {
 437       cqr->hnode =
 438           GNUNET_CONTAINER_heap_insert (retry_heap, cqr,
 439                                         cqr->retry_time.abs_value);
 440       retry_task =
 441           GNUNET_SCHEDULER_add_delayed (delay, &transmit_next_request_task,
 442                                         NULL);
 443       return;
 444     }
 445     transmit_request (cqr);
 446     cqr->hnode =
 447         GNUNET_CONTAINER_heap_insert (retry_heap, cqr,
 448                                       cqr->retry_time.abs_value);
 449   }
 450 }
 451
 452
 453 /**
 454  * Handler for PUT messages.
 455  *
 456  * @param cls closure for the service
 457  * @param client the client we received this message from
 458  * @param message the actual message received
 459  */
 460 static void
 461 handle_dht_local_put (void *cls, struct GNUNET_SERVER_Client *client,
 462                       const struct GNUNET_MessageHeader *message)
 463 {
 464   const struct GNUNET_DHT_ClientPutMessage *dht_msg;
 465   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 466   uint16_t size;
 467
 468   size = ntohs (message->size);
 469   if (size < sizeof (struct GNUNET_DHT_ClientPutMessage))
 470   {
 471     GNUNET_break (0);
 472     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 473     return;
 474   }
 475   GNUNET_STATISTICS_update (GDS_stats,
 476                             gettext_noop
 477                             ("# PUT requests received from clients"), 1,
 478                             GNUNET_NO);
 479   dht_msg = (const struct GNUNET_DHT_ClientPutMessage *) message;
 480   /* give to local clients */
 481 #if DEBUG_DHT
 482   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 483               "Handling local PUT of %u-bytes for query %s\n",
 484               size - sizeof (struct GNUNET_DHT_ClientPutMessage),
 485               GNUNET_h2s (&dht_msg->key));
 486 #endif
 487   GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 488                             &dht_msg->key, 0, NULL, 0, NULL,
 489                             ntohl (dht_msg->type),
 490                             size - sizeof (struct GNUNET_DHT_ClientPutMessage),
 491                             &dht_msg[1]);
 492   /* store locally */
 493   GDS_DATACACHE_handle_put (GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 494                             &dht_msg->key, 0, NULL, ntohl (dht_msg->type),
 495                             size - sizeof (struct GNUNET_DHT_ClientPutMessage),
 496                             &dht_msg[1]);
 497   /* route to other peers */
 498   peer_bf =
 499       GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE,
 500                                          GNUNET_CONSTANTS_BLOOMFILTER_K);
 501   GDS_NEIGHBOURS_handle_put (ntohl (dht_msg->type), ntohl (dht_msg->options),
 502                              ntohl (dht_msg->desired_replication_level),
 503                              GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 504                              0 /* hop count */ ,
 505                              peer_bf, &dht_msg->key, 0, NULL, &dht_msg[1],
 506                              size -
 507                              sizeof (struct GNUNET_DHT_ClientPutMessage));
 508   GDS_CLIENTS_process_put (ntohl (dht_msg->options),
 509                            ntohl (dht_msg->type),
 510                            0,
 511                            ntohl (dht_msg->desired_replication_level),
 512                            1,
 513                            GDS_NEIGHBOURS_get_id(),
 514                            GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 515                            &dht_msg->key,
 516                            &dht_msg[1],
 517                            size - sizeof (struct GNUNET_DHT_ClientPutMessage));
 518   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 519   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 520 }
 521
 522
 523 /**
 524  * Handler for any generic DHT messages, calls the appropriate handler
 525  * depending on message type, sends confirmation if responses aren't otherwise
 526  * expected.
 527  *
 528  * @param cls closure for the service
 529  * @param client the client we received this message from
 530  * @param message the actual message received
 531  */
 532 static void
 533 handle_dht_local_get (void *cls, struct GNUNET_SERVER_Client *client,
 534                       const struct GNUNET_MessageHeader *message)
 535 {
 536   const struct GNUNET_DHT_ClientGetMessage *get;
 537   struct ClientQueryRecord *cqr;
 538   size_t xquery_size;
 539   const char *xquery;
 540   uint16_t size;
 541
 542   size = ntohs (message->size);
 543   if (size < sizeof (struct GNUNET_DHT_ClientGetMessage))
 544   {
 545     GNUNET_break (0);
 546     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 547     return;
 548   }
 549   xquery_size = size - sizeof (struct GNUNET_DHT_ClientGetMessage);
 550   get = (const struct GNUNET_DHT_ClientGetMessage *) message;
 551   xquery = (const char *) &get[1];
 552   GNUNET_STATISTICS_update (GDS_stats,
 553                             gettext_noop
 554                             ("# GET requests received from clients"), 1,
 555                             GNUNET_NO);
 556 #if DEBUG_DHT
 557   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 558               "Received request for %s from local client %p\n",
 559               GNUNET_h2s (&get->key), client);
 560 #endif
 561   cqr = GNUNET_malloc (sizeof (struct ClientQueryRecord) + xquery_size);
 562   cqr->key = get->key;
 563   cqr->client = find_active_client (client);
 564   cqr->xquery = (void *) &cqr[1];
 565   memcpy (&cqr[1], xquery, xquery_size);
 566   cqr->hnode = GNUNET_CONTAINER_heap_insert (retry_heap, cqr, 0);
 567   cqr->retry_frequency = GNUNET_TIME_UNIT_MILLISECONDS;
 568   cqr->retry_time = GNUNET_TIME_absolute_get ();
 569   cqr->unique_id = get->unique_id;
 570   cqr->xquery_size = xquery_size;
 571   cqr->replication = ntohl (get->desired_replication_level);
 572   cqr->msg_options = ntohl (get->options);
 573   cqr->type = ntohl (get->type);
 574   GNUNET_CONTAINER_multihashmap_put (forward_map, &get->key, cqr,
 575                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 576   GDS_CLIENTS_process_get (ntohl (get->options),
 577                            ntohl (get->type),
 578                            0,
 579                            ntohl (get->desired_replication_level),
 580                            1,
 581                            GDS_NEIGHBOURS_get_id(),
 582                            &get->key);
 583   /* start remote requests */
 584   if (GNUNET_SCHEDULER_NO_TASK != retry_task)
 585     GNUNET_SCHEDULER_cancel (retry_task);
 586   retry_task = GNUNET_SCHEDULER_add_now (&transmit_next_request_task, NULL);
 587   /* perform local lookup */
 588   GDS_DATACACHE_handle_get (&get->key, cqr->type, cqr->xquery, xquery_size,
 589                             NULL, 0);
 590   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 591 }
 592
 593
 594 /**
 595  * Closure for 'remove_by_unique_id'.
 596  */
 597 struct RemoveByUniqueIdContext
 598 {
 599   /**
 600    * Client that issued the removal request.
 601    */
 602   struct ClientList *client;
 603
 604   /**
 605    * Unique ID of the request.
 606    */
 607   uint64_t unique_id;
 608 };
 609
 610
 611 /**
 612  * Iterator over hash map entries that frees all entries
 613  * that match the given client and unique ID.
 614  *
 615  * @param cls unique ID and client to search for in source routes
 616  * @param key current key code
 617  * @param value value in the hash map, a ClientQueryRecord
 618  * @return GNUNET_YES (we should continue to iterate)
 619  */
 620 static int
 621 remove_by_unique_id (void *cls, const GNUNET_HashCode * key, void *value)
 622 {
 623   const struct RemoveByUniqueIdContext *ctx = cls;
 624   struct ClientQueryRecord *record = value;
 625
 626   if (record->unique_id != ctx->unique_id)
 627     return GNUNET_YES;
 628 #if DEBUG_DHT
 629   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 630               "Removing client %p's record for key %s (by unique id)\n",
 631               ctx->client->client_handle, GNUNET_h2s (key));
 632 #endif
 633   return remove_client_records (ctx->client, key, record);
 634 }
 635
 636
 637 /**
 638  * Handler for any generic DHT stop messages, calls the appropriate handler
 639  * depending on message type (if processed locally)
 640  *
 641  * @param cls closure for the service
 642  * @param client the client we received this message from
 643  * @param message the actual message received
 644  *
 645  */
 646 static void
 647 handle_dht_local_get_stop (void *cls, struct GNUNET_SERVER_Client *client,
 648                            const struct GNUNET_MessageHeader *message)
 649 {
 650   const struct GNUNET_DHT_ClientGetStopMessage *dht_stop_msg =
 651       (const struct GNUNET_DHT_ClientGetStopMessage *) message;
 652   struct RemoveByUniqueIdContext ctx;
 653
 654   GNUNET_STATISTICS_update (GDS_stats,
 655                             gettext_noop
 656                             ("# GET STOP requests received from clients"), 1,
 657                             GNUNET_NO);
 658 #if DEBUG_DHT
 659   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p stopped request for key %s\n",
 660               client, GNUNET_h2s (&dht_stop_msg->key));
 661 #endif
 662   ctx.client = find_active_client (client);
 663   ctx.unique_id = dht_stop_msg->unique_id;
 664   GNUNET_CONTAINER_multihashmap_get_multiple (forward_map, &dht_stop_msg->key,
 665                                               &remove_by_unique_id, &ctx);
 666   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 667 }
 668
 669
 670 /**
 671  * Handler for monitor messages
 672  *
 673  * @param cls closure for the service
 674  * @param client the client we received this message from
 675  * @param message the actual message received
 676  *
 677  */
 678 static void
 679 handle_dht_local_monitor (void *cls, struct GNUNET_SERVER_Client *client,
 680                           const struct GNUNET_MessageHeader *message)
 681 {
 682   struct ClientMonitorRecord *r;
 683   const struct GNUNET_DHT_MonitorStartMessage *msg;
 684   unsigned int i;
 685   char *c;
 686
 687   msg = (struct GNUNET_DHT_MonitorStartMessage *) message;
 688   r = GNUNET_malloc (sizeof(struct ClientMonitorRecord));
 689
 690   r->client = find_active_client(client);
 691   r->type = ntohl(msg->type);
 692   r->get = msg->get;
 693   r->get_resp = msg->get_resp;
 694   r->put = msg->put;
 695   c = (char *) &msg->key;
 696   for (i = 0; i < sizeof (GNUNET_HashCode) && c[i] == 0; i++);
 697   if (sizeof (GNUNET_HashCode) == i)
 698     r->key = NULL;
 699   else
 700   {
 701     r->key = GNUNET_malloc (sizeof (GNUNET_HashCode));
 702     memcpy (r->key, &msg->key, sizeof (GNUNET_HashCode));
 703   }
 704   GNUNET_CONTAINER_DLL_insert (monitor_head, monitor_tail, r);
 705   // FIXME add remove somewhere
 706   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 707 }
 708
 709
 710 /**
 711  * Task run to check for messages that need to be sent to a client.
 712  *
 713  * @param client a ClientList, containing the client and any messages to be sent to it
 714  */
 715 static void
 716 process_pending_messages (struct ClientList *client);
 717
 718
 719 /**
 720  * Callback called as a result of issuing a GNUNET_SERVER_notify_transmit_ready
 721  * request.  A ClientList is passed as closure, take the head of the list
 722  * and copy it into buf, which has the result of sending the message to the
 723  * client.
 724  *
 725  * @param cls closure to this call
 726  * @param size maximum number of bytes available to send
 727  * @param buf where to copy the actual message to
 728  *
 729  * @return the number of bytes actually copied, 0 indicates failure
 730  */
 731 static size_t
 732 send_reply_to_client (void *cls, size_t size, void *buf)
 733 {
 734   struct ClientList *client = cls;
 735   char *cbuf = buf;
 736   struct PendingMessage *reply;
 737   size_t off;
 738   size_t msize;
 739
 740   client->transmit_handle = NULL;
 741   if (buf == NULL)
 742   {
 743     /* client disconnected */
 744 #if DEBUG_DHT
 745     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 746                 "Client %p disconnected, pending messages will be discarded\n",
 747                 client->client_handle);
 748 #endif
 749     return 0;
 750   }
 751   off = 0;
 752   while ((NULL != (reply = client->pending_head)) &&
 753          (size >= off + (msize = ntohs (reply->msg->size))))
 754   {
 755     GNUNET_CONTAINER_DLL_remove (client->pending_head, client->pending_tail,
 756                                  reply);
 757     memcpy (&cbuf[off], reply->msg, msize);
 758     GNUNET_free (reply);
 759 #if DEBUG_DHT
 760     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting %u bytes to client %p\n",
 761                 msize, client->client_handle);
 762 #endif
 763     off += msize;
 764   }
 765   process_pending_messages (client);
 766 #if DEBUG_DHT
 767   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitted %u/%u bytes to client %p\n",
 768               (unsigned int) off, (unsigned int) size, client->client_handle);
 769 #endif
 770   return off;
 771 }
 772
 773
 774 /**
 775  * Task run to check for messages that need to be sent to a client.
 776  *
 777  * @param client a ClientList, containing the client and any messages to be sent to it
 778  */
 779 static void
 780 process_pending_messages (struct ClientList *client)
 781 {
 782   if ((client->pending_head == NULL) || (client->transmit_handle != NULL))
 783   {
 784 #if DEBUG_DHT
 785     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 786                 "Not asking for transmission to %p now: %s\n",
 787                 client->client_handle,
 788                 client->pending_head ==
 789                 NULL ? "no more messages" : "request already pending");
 790 #endif
 791     return;
 792   }
 793 #if DEBUG_DHT
 794   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 795               "Asking for transmission of %u bytes to client %p\n",
 796               ntohs (client->pending_head->msg->size), client->client_handle);
 797 #endif
 798   client->transmit_handle =
 799       GNUNET_SERVER_notify_transmit_ready (client->client_handle,
 800                                            ntohs (client->pending_head->
 801                                                   msg->size),
 802                                            GNUNET_TIME_UNIT_FOREVER_REL,
 803                                            &send_reply_to_client, client);
 804 }
 805
 806
 807 /**
 808  * Add a PendingMessage to the clients list of messages to be sent
 809  *
 810  * @param client the active client to send the message to
 811  * @param pending_message the actual message to send
 812  */
 813 static void
 814 add_pending_message (struct ClientList *client,
 815                      struct PendingMessage *pending_message)
 816 {
 817   GNUNET_CONTAINER_DLL_insert_tail (client->pending_head, client->pending_tail,
 818                                     pending_message);
 819   process_pending_messages (client);
 820 }
 821
 822
 823 /**
 824  * Closure for 'forward_reply'
 825  */
 826 struct ForwardReplyContext
 827 {
 828
 829   /**
 830    * Actual message to send to matching clients.
 831    */
 832   struct PendingMessage *pm;
 833
 834   /**
 835    * Embedded payload.
 836    */
 837   const void *data;
 838
 839   /**
 840    * Type of the data.
 841    */
 842   enum GNUNET_BLOCK_Type type;
 843
 844   /**
 845    * Number of bytes in data.
 846    */
 847   size_t data_size;
 848
 849   /**
 850    * Do we need to copy 'pm' because it was already used?
 851    */
 852   int do_copy;
 853
 854 };
 855
 856
 857 /**
 858  * Iterator over hash map entries that send a given reply to
 859  * each of the matching clients.  With some tricky recycling
 860  * of the buffer.
 861  *
 862  * @param cls the 'struct ForwardReplyContext'
 863  * @param key current key
 864  * @param value value in the hash map, a ClientQueryRecord
 865  * @return GNUNET_YES (we should continue to iterate),
 866  *         if the result is mal-formed, GNUNET_NO
 867  */
 868 static int
 869 forward_reply (void *cls, const GNUNET_HashCode * key, void *value)
 870 {
 871   struct ForwardReplyContext *frc = cls;
 872   struct ClientQueryRecord *record = value;
 873   struct PendingMessage *pm;
 874   struct GNUNET_DHT_ClientResultMessage *reply;
 875   enum GNUNET_BLOCK_EvaluationResult eval;
 876   int do_free;
 877   GNUNET_HashCode ch;
 878   unsigned int i;
 879
 880   if ((record->type != GNUNET_BLOCK_TYPE_ANY) && (record->type != frc->type))
 881   {
 882 #if DEBUG_DHT
 883     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 884                 "Record type missmatch, not passing request for key %s to local client\n",
 885                 GNUNET_h2s (key));
 886 #endif
 887     GNUNET_STATISTICS_update (GDS_stats,
 888                               gettext_noop
 889                               ("# Key match, type mismatches in REPLY to CLIENT"),
 890                               1, GNUNET_NO);
 891     return GNUNET_YES;          /* type mismatch */
 892   }
 893   GNUNET_CRYPTO_hash (frc->data, frc->data_size, &ch);
 894   for (i = 0; i < record->seen_replies_count; i++)
 895     if (0 == memcmp (&record->seen_replies[i], &ch, sizeof (GNUNET_HashCode)))
 896     {
 897 #if DEBUG_DHT
 898       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 899                   "Duplicate reply, not passing request for key %s to local client\n",
 900                   GNUNET_h2s (key));
 901 #endif
 902       GNUNET_STATISTICS_update (GDS_stats,
 903                                 gettext_noop
 904                                 ("# Duplicate REPLIES to CLIENT request dropped"),
 905                                 1, GNUNET_NO);
 906       return GNUNET_YES;        /* duplicate */
 907     }
 908   eval =
 909       GNUNET_BLOCK_evaluate (GDS_block_context, record->type, key, NULL, 0,
 910                              record->xquery, record->xquery_size, frc->data,
 911                              frc->data_size);
 912 #if DEBUG_DHT
 913   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 914               "Evaluation result is %d for key %s for local client's query\n",
 915               (int) eval, GNUNET_h2s (key));
 916 #endif
 917   switch (eval)
 918   {
 919   case GNUNET_BLOCK_EVALUATION_OK_LAST:
 920     do_free = GNUNET_YES;
 921     break;
 922   case GNUNET_BLOCK_EVALUATION_OK_MORE:
 923     GNUNET_array_append (record->seen_replies, record->seen_replies_count, ch);
 924     do_free = GNUNET_NO;
 925     break;
 926   case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
 927     /* should be impossible to encounter here */
 928     GNUNET_break (0);
 929     return GNUNET_YES;
 930   case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
 931     GNUNET_break_op (0);
 932     return GNUNET_NO;
 933   case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
 934     GNUNET_break (0);
 935     return GNUNET_NO;
 936   case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
 937     GNUNET_break (0);
 938     return GNUNET_NO;
 939   case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
 940     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 941                 _("Unsupported block type (%u) in request!\n"), record->type);
 942     return GNUNET_NO;
 943   default:
 944     GNUNET_break (0);
 945     return GNUNET_NO;
 946   }
 947   if (GNUNET_NO == frc->do_copy)
 948   {
 949     /* first time, we can use the original data */
 950     pm = frc->pm;
 951     frc->do_copy = GNUNET_YES;
 952   }
 953   else
 954   {
 955     /* two clients waiting for same reply, must copy for queueing */
 956     pm = GNUNET_malloc (sizeof (struct PendingMessage) +
 957                         ntohs (frc->pm->msg->size));
 958     memcpy (pm, frc->pm,
 959             sizeof (struct PendingMessage) + ntohs (frc->pm->msg->size));
 960     pm->next = pm->prev = NULL;
 961   }
 962   GNUNET_STATISTICS_update (GDS_stats,
 963                             gettext_noop ("# RESULTS queued for clients"), 1,
 964                             GNUNET_NO);
 965   reply = (struct GNUNET_DHT_ClientResultMessage *) &pm[1];
 966   reply->unique_id = record->unique_id;
 967 #if DEBUG_DHT
 968   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 969               "Queueing reply to query %s for client %p\n", GNUNET_h2s (key),
 970               record->client->client_handle);
 971 #endif
 972   add_pending_message (record->client, pm);
 973   if (GNUNET_YES == do_free)
 974     remove_client_records (record->client, key, record);
 975   return GNUNET_YES;
 976 }
 977
 978
 979 /**
 980  * Handle a reply we've received from another peer.  If the reply
 981  * matches any of our pending queries, forward it to the respective
 982  * client(s).
 983  *
 984  * @param expiration when will the reply expire
 985  * @param key the query this reply is for
 986  * @param get_path_length number of peers in 'get_path'
 987  * @param get_path path the reply took on get
 988  * @param put_path_length number of peers in 'put_path'
 989  * @param put_path path the reply took on put
 990  * @param type type of the reply
 991  * @param data_size number of bytes in 'data'
 992  * @param data application payload data
 993  */
 994 void
 995 GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
 996                           const GNUNET_HashCode * key,
 997                           unsigned int get_path_length,
 998                           const struct GNUNET_PeerIdentity *get_path,
 999                           unsigned int put_path_length,
1000                           const struct GNUNET_PeerIdentity *put_path,
1001                           enum GNUNET_BLOCK_Type type, size_t data_size,
1002                           const void *data)
1003 {
1004   struct ForwardReplyContext frc;
1005   struct PendingMessage *pm;
1006   struct GNUNET_DHT_ClientResultMessage *reply;
1007   struct GNUNET_PeerIdentity *paths;
1008   size_t msize;
1009
1010   if (NULL == GNUNET_CONTAINER_multihashmap_get (forward_map, key))
1011   {
1012     GNUNET_STATISTICS_update (GDS_stats,
1013                               gettext_noop
1014                               ("# REPLIES ignored for CLIENTS (no match)"), 1,
1015                               GNUNET_NO);
1016     return;                     /* no matching request, fast exit! */
1017   }
1018   msize =
1019       sizeof (struct GNUNET_DHT_ClientResultMessage) + data_size +
1020       (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity);
1021   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1022   {
1023     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1024                 _("Could not pass reply to client, message too big!\n"));
1025     return;
1026   }
1027   pm = (struct PendingMessage *) GNUNET_malloc (msize +
1028                                                 sizeof (struct PendingMessage));
1029   reply = (struct GNUNET_DHT_ClientResultMessage *) &pm[1];
1030   pm->msg = &reply->header;
1031   reply->header.size = htons ((uint16_t) msize);
1032   reply->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT);
1033   reply->type = htonl (type);
1034   reply->get_path_length = htonl (get_path_length);
1035   reply->put_path_length = htonl (put_path_length);
1036   reply->unique_id = 0;         /* filled in later */
1037   reply->expiration = GNUNET_TIME_absolute_hton (expiration);
1038   reply->key = *key;
1039   paths = (struct GNUNET_PeerIdentity *) &reply[1];
1040   memcpy (paths, put_path,
1041           sizeof (struct GNUNET_PeerIdentity) * put_path_length);
1042   memcpy (&paths[put_path_length], get_path,
1043           sizeof (struct GNUNET_PeerIdentity) * get_path_length);
1044   memcpy (&paths[get_path_length + put_path_length], data, data_size);
1045   frc.do_copy = GNUNET_NO;
1046   frc.pm = pm;
1047   frc.data = data;
1048   frc.data_size = data_size;
1049   frc.type = type;
1050   GNUNET_CONTAINER_multihashmap_get_multiple (forward_map, key, &forward_reply,
1051                                               &frc);
1052   if (GNUNET_NO == frc.do_copy)
1053   {
1054     /* did not match any of the requests, free! */
1055     GNUNET_STATISTICS_update (GDS_stats,
1056                               gettext_noop
1057                               ("# REPLIES ignored for CLIENTS (no match)"), 1,
1058                               GNUNET_NO);
1059     GNUNET_free (pm);
1060   }
1061 }
1062
1063
1064 /**
1065  * Check if some client is monitoring GET messages and notify
1066  * them in that case.
1067  *
1068  * @param options Options, for instance RecordRoute, DemultiplexEverywhere.
1069  * @param type The type of data in the request.
1070  * @param hop_count Hop count so far.
1071  * @param path_length number of entries in path (or 0 if not recorded).
1072  * @param path peers on the GET path (or NULL if not recorded).
1073  * @param desired_replication_level Desired replication level.
1074  * @param key Key of the requested data.
1075  */
1076 void
1077 GDS_CLIENTS_process_get (uint32_t options,
1078                          enum GNUNET_BLOCK_Type type,
1079                          uint32_t hop_count,
1080                          uint32_t desired_replication_level,
1081                          unsigned int path_length,
1082                          const struct GNUNET_PeerIdentity *path,
1083                          const GNUNET_HashCode * key)
1084 {
1085   struct ClientMonitorRecord *m;
1086   struct ClientList **cl;
1087   unsigned int cl_size;
1088
1089   cl = NULL;
1090   cl_size = 0;
1091   for (m = monitor_head; NULL != m; m = m->next)
1092   {
1093     if ((GNUNET_BLOCK_TYPE_ANY == m->type || m->type == type) &&
1094         (NULL == m->key ||
1095          memcmp (key, m->key, sizeof(GNUNET_HashCode)) == 0))
1096     {
1097       struct PendingMessage *pm;
1098       struct GNUNET_DHT_MonitorGetMessage *mmsg;
1099       struct GNUNET_PeerIdentity *msg_path;
1100       size_t msize;
1101       unsigned int i;
1102
1103       /* Don't send duplicates */
1104       for (i = 0; i < cl_size; i++)
1105         if (cl[i] == m->client)
1106           break;
1107       if (i < cl_size)
1108         continue;
1109       GNUNET_array_append (cl, cl_size, m->client);
1110
1111       msize = path_length * sizeof (struct GNUNET_PeerIdentity);
1112       msize += sizeof (struct GNUNET_DHT_MonitorGetMessage);
1113       msize += sizeof (struct PendingMessage);
1114       pm = (struct PendingMessage *) GNUNET_malloc (msize);
1115       mmsg = (struct GNUNET_DHT_MonitorGetMessage *) &pm[1];
1116       pm->msg = (struct GNUNET_MessageHeader *) mmsg;
1117       mmsg->header.size = htons (msize - sizeof (struct PendingMessage));
1118       mmsg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET);
1119       mmsg->options = htonl(options);
1120       mmsg->type = htonl(type);
1121       mmsg->hop_count = htonl(hop_count);
1122       mmsg->desired_replication_level = htonl(desired_replication_level);
1123       mmsg->get_path_length = htonl(path_length);
1124       memcpy (&mmsg->key, key, sizeof (GNUNET_HashCode));
1125       msg_path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1126       if (path_length > 0)
1127         memcpy (msg_path, path,
1128                 path_length * sizeof (struct GNUNET_PeerIdentity));
1129       add_pending_message (m->client, pm);
1130     }
1131   }
1132   GNUNET_free_non_null (cl);
1133 }
1134
1135
1136 /**
1137  * Check if some client is monitoring GET RESP messages and notify
1138  * them in that case.
1139  *
1140  * @param type The type of data in the result.
1141  * @param get_path Peers on GET path (or NULL if not recorded).
1142  * @param get_path_length number of entries in get_path.
1143  * @param put_path peers on the PUT path (or NULL if not recorded).
1144  * @param put_path_length number of entries in get_path.
1145  * @param exp Expiration time of the data.
1146  * @param key Key of the data.
1147  * @param data Pointer to the result data.
1148  * @param size Number of bytes in data.
1149  */
1150 void
1151 GDS_CLIENTS_process_get_resp (enum GNUNET_BLOCK_Type type,
1152                               const struct GNUNET_PeerIdentity *get_path,
1153                               unsigned int get_path_length,
1154                               const struct GNUNET_PeerIdentity *put_path,
1155                               unsigned int put_path_length,
1156                               struct GNUNET_TIME_Absolute exp,
1157                               const GNUNET_HashCode * key,
1158                               const void *data,
1159                               size_t size)
1160 {
1161   struct ClientMonitorRecord *m;
1162   struct ClientList **cl;
1163   unsigned int cl_size;
1164
1165   cl = NULL;
1166   cl_size = 0;
1167   for (m = monitor_head; NULL != m; m = m->next)
1168   {
1169     if ((GNUNET_BLOCK_TYPE_ANY == m->type || m->type == type) &&
1170         (NULL == m->key ||
1171          memcmp (key, m->key, sizeof(GNUNET_HashCode)) == 0))
1172     {
1173       struct PendingMessage *pm;
1174       struct GNUNET_DHT_MonitorGetRespMessage *mmsg;
1175       struct GNUNET_PeerIdentity *path;
1176       size_t msize;
1177       unsigned int i;
1178
1179       /* Don't send duplicates */
1180       for (i = 0; i < cl_size; i++)
1181         if (cl[i] == m->client)
1182           break;
1183       if (i < cl_size)
1184         continue;
1185       GNUNET_array_append (cl, cl_size, m->client);
1186
1187       msize = size;
1188       msize += (get_path_length + put_path_length)
1189                * sizeof (struct GNUNET_PeerIdentity);
1190       msize += sizeof (struct GNUNET_DHT_MonitorGetRespMessage);
1191       msize += sizeof (struct PendingMessage);
1192       pm = (struct PendingMessage *) GNUNET_malloc (msize);
1193       mmsg = (struct GNUNET_DHT_MonitorGetRespMessage *) &pm[1];
1194       pm->msg = (struct GNUNET_MessageHeader *) mmsg;
1195       mmsg->header.size = htons (msize - sizeof (struct PendingMessage));
1196       mmsg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET_RESP);
1197       mmsg->type = htonl(type);
1198       mmsg->put_path_length = htonl(put_path_length);
1199       mmsg->get_path_length = htonl(get_path_length);
1200       path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1201       if (put_path_length > 0)
1202       {
1203         memcpy (path, put_path,
1204                 put_path_length * sizeof (struct GNUNET_PeerIdentity));
1205         path = &path[put_path_length];
1206       }
1207       if (get_path_length > 0)
1208         memcpy (path, get_path,
1209                 get_path_length * sizeof (struct GNUNET_PeerIdentity));
1210       mmsg->expiration_time = GNUNET_TIME_absolute_hton(exp);
1211       memcpy (&mmsg->key, key, sizeof (GNUNET_HashCode));
1212       if (size > 0)
1213         memcpy (&path[get_path_length], data, size);
1214       add_pending_message (m->client, pm);
1215     }
1216   }
1217   GNUNET_free_non_null (cl);
1218 }
1219
1220
1221 /**
1222  * Check if some client is monitoring PUT messages and notify
1223  * them in that case.
1224  *
1225  * @param options Options, for instance RecordRoute, DemultiplexEverywhere.
1226  * @param type The type of data in the request.
1227  * @param hop_count Hop count so far.
1228  * @param path_length number of entries in path (or 0 if not recorded).
1229  * @param path peers on the PUT path (or NULL if not recorded).
1230  * @param desired_replication_level Desired replication level.
1231  * @param exp Expiration time of the data.
1232  * @param key Key under which data is to be stored.
1233  * @param data Pointer to the data carried.
1234  * @param size Number of bytes in data.
1235  */
1236 void
1237 GDS_CLIENTS_process_put (uint32_t options,
1238                          enum GNUNET_BLOCK_Type type,
1239                          uint32_t hop_count,
1240                          uint32_t desired_replication_level,
1241                          unsigned int path_length,
1242                          const struct GNUNET_PeerIdentity *path,
1243                          struct GNUNET_TIME_Absolute exp,
1244                          const GNUNET_HashCode * key,
1245                          const void *data,
1246                          size_t size)
1247 {
1248   struct ClientMonitorRecord *m;
1249   struct ClientList **cl;
1250   unsigned int cl_size;
1251
1252   cl = NULL;
1253   cl_size = 0;
1254   for (m = monitor_head; NULL != m; m = m->next)
1255   {
1256     if ((GNUNET_BLOCK_TYPE_ANY == m->type || m->type == type) &&
1257         (NULL == m->key ||
1258          memcmp (key, m->key, sizeof(GNUNET_HashCode)) == 0))
1259     {
1260       struct PendingMessage *pm;
1261       struct GNUNET_DHT_MonitorPutMessage *mmsg;
1262       struct GNUNET_PeerIdentity *msg_path;
1263       size_t msize;
1264       unsigned int i;
1265
1266       /* Don't send duplicates */
1267       for (i = 0; i < cl_size; i++)
1268         if (cl[i] == m->client)
1269           break;
1270       if (i < cl_size)
1271         continue;
1272       GNUNET_array_append (cl, cl_size, m->client);
1273
1274       msize = size;
1275       msize += path_length * sizeof (struct GNUNET_PeerIdentity);
1276       msize += sizeof (struct GNUNET_DHT_MonitorPutMessage);
1277       msize += sizeof (struct PendingMessage);
1278       pm = (struct PendingMessage *) GNUNET_malloc (msize);
1279       mmsg = (struct GNUNET_DHT_MonitorPutMessage *) &pm[1];
1280       pm->msg = (struct GNUNET_MessageHeader *) mmsg;
1281       mmsg->header.size = htons (msize - sizeof (struct PendingMessage));
1282       mmsg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT);
1283       mmsg->options = htonl(options);
1284       mmsg->type = htonl(type);
1285       mmsg->hop_count = htonl(hop_count);
1286       mmsg->desired_replication_level = htonl(desired_replication_level);
1287       mmsg->put_path_length = htonl(path_length);
1288       msg_path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1289       if (path_length > 0)
1290       {
1291         memcpy (msg_path, path,
1292                 path_length * sizeof (struct GNUNET_PeerIdentity));
1293       }
1294       mmsg->expiration_time = GNUNET_TIME_absolute_hton(exp);
1295       memcpy (&mmsg->key, key, sizeof (GNUNET_HashCode));
1296       if (size > 0)
1297         memcpy (&msg_path[path_length], data, size);
1298       add_pending_message (m->client, pm);
1299     }
1300   }
1301   GNUNET_free_non_null (cl);
1302 }
1303
1304
1305 /**
1306  * Initialize client subsystem.
1307  *
1308  * @param server the initialized server
1309  */
1310 void
1311 GDS_CLIENTS_init (struct GNUNET_SERVER_Handle *server)
1312 {
1313   static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = {
1314     {&handle_dht_local_put, NULL,
1315      GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT, 0},
1316     {&handle_dht_local_get, NULL,
1317      GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET, 0},
1318     {&handle_dht_local_get_stop, NULL,
1319      GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP,
1320      sizeof (struct GNUNET_DHT_ClientGetStopMessage)},
1321     {&handle_dht_local_monitor, NULL,
1322      GNUNET_MESSAGE_TYPE_DHT_MONITOR_START,
1323      sizeof (struct GNUNET_DHT_MonitorStartMessage)},
1324     {NULL, NULL, 0, 0}
1325   };
1326   forward_map = GNUNET_CONTAINER_multihashmap_create (1024);
1327   retry_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1328   GNUNET_SERVER_add_handlers (server, plugin_handlers);
1329   GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
1330 }
1331
1332
1333 /**
1334  * Shutdown client subsystem.
1335  */
1336 void
1337 GDS_CLIENTS_done ()
1338 {
1339   GNUNET_assert (client_head == NULL);
1340   GNUNET_assert (client_tail == NULL);
1341   if (GNUNET_SCHEDULER_NO_TASK != retry_task)
1342   {
1343     GNUNET_SCHEDULER_cancel (retry_task);
1344     retry_task = GNUNET_SCHEDULER_NO_TASK;
1345   }
1346   if (NULL != retry_heap)
1347   {
1348     GNUNET_assert (0 == GNUNET_CONTAINER_heap_get_size (retry_heap));
1349     GNUNET_CONTAINER_heap_destroy (retry_heap);
1350     retry_heap = NULL;
1351   }
1352   if (NULL != forward_map)
1353   {
1354     GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (forward_map));
1355     GNUNET_CONTAINER_multihashmap_destroy (forward_map);
1356     forward_map = NULL;
1357   }
1358 }
1359
1360 /* end of gnunet-service-dht_clients.c */