src/dht/gnunet-service-dht_clients.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009, 2010, 2011, 2016, 2017 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  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_datacache.h"
  34 #include "gnunet-service-dht_neighbours.h"
  35 #include "dht.h"
  36
  37
  38 /**
  39  * Should routing details be logged to stderr (for debugging)?
  40  */
  41 #define LOG_TRAFFIC(kind, ...) GNUNET_log_from (kind, "dht-traffic", \
  42                                                 __VA_ARGS__)
  43
  44 #define LOG(kind, ...) GNUNET_log_from (kind, "dht-clients", __VA_ARGS__)
  45
  46
  47 /**
  48  * Struct containing information about a client,
  49  * handle to connect to it, and any pending messages
  50  * that need to be sent to it.
  51  */
  52 struct ClientHandle;
  53
  54
  55 /**
  56  * Entry in the local forwarding map for a client's GET request.
  57  */
  58 struct ClientQueryRecord
  59 {
  60   /**
  61    * The key this request was about
  62    */
  63   struct GNUNET_HashCode key;
  64
  65   /**
  66    * Kept in a DLL with @e client.
  67    */
  68   struct ClientQueryRecord *next;
  69
  70   /**
  71    * Kept in a DLL with @e client.
  72    */
  73   struct ClientQueryRecord *prev;
  74
  75   /**
  76    * Client responsible for the request.
  77    */
  78   struct ClientHandle *ch;
  79
  80   /**
  81    * Extended query (see gnunet_block_lib.h), allocated at the end of this struct.
  82    */
  83   const void *xquery;
  84
  85   /**
  86    * Replies we have already seen for this request.
  87    */
  88   struct GNUNET_HashCode *seen_replies;
  89
  90   /**
  91    * Pointer to this nodes heap location in the retry-heap (for fast removal)
  92    */
  93   struct GNUNET_CONTAINER_HeapNode *hnode;
  94
  95   /**
  96    * What's the delay between re-try operations that we currently use for this
  97    * request?
  98    */
  99   struct GNUNET_TIME_Relative retry_frequency;
 100
 101   /**
 102    * What's the next time we should re-try this request?
 103    */
 104   struct GNUNET_TIME_Absolute retry_time;
 105
 106   /**
 107    * The unique identifier of this request
 108    */
 109   uint64_t unique_id;
 110
 111   /**
 112    * Number of bytes in xquery.
 113    */
 114   size_t xquery_size;
 115
 116   /**
 117    * Number of entries in 'seen_replies'.
 118    */
 119   unsigned int seen_replies_count;
 120
 121   /**
 122    * Desired replication level
 123    */
 124   uint32_t replication;
 125
 126   /**
 127    * Any message options for this request
 128    */
 129   uint32_t msg_options;
 130
 131   /**
 132    * The type for the data for the GET request.
 133    */
 134   enum GNUNET_BLOCK_Type type;
 135 };
 136
 137
 138 /**
 139  * Struct containing paremeters of monitoring requests.
 140  */
 141 struct ClientMonitorRecord
 142 {
 143   /**
 144    * Next element in DLL.
 145    */
 146   struct ClientMonitorRecord *next;
 147
 148   /**
 149    * Previous element in DLL.
 150    */
 151   struct ClientMonitorRecord *prev;
 152
 153   /**
 154    * Type of blocks that are of interest
 155    */
 156   enum GNUNET_BLOCK_Type type;
 157
 158   /**
 159    * Key of data of interest, NULL for all.
 160    */
 161   struct GNUNET_HashCode *key;
 162
 163   /**
 164    * Flag whether to notify about GET messages.
 165    */
 166   int16_t get;
 167
 168   /**
 169    * Flag whether to notify about GET_REPONSE messages.
 170    */
 171   int16_t get_resp;
 172
 173   /**
 174    * Flag whether to notify about PUT messages.
 175    */
 176   uint16_t put;
 177
 178   /**
 179    * Client to notify of these requests.
 180    */
 181   struct ClientHandle *ch;
 182 };
 183
 184
 185 /**
 186  * Struct containing information about a client,
 187  * handle to connect to it, and any pending messages
 188  * that need to be sent to it.
 189  */
 190 struct ClientHandle
 191 {
 192   /**
 193    * Linked list of active queries of this client.
 194    */
 195   struct ClientQueryRecord *cqr_head;
 196
 197   /**
 198    * Linked list of active queries of this client.
 199    */
 200   struct ClientQueryRecord *cqr_tail;
 201
 202   /**
 203    * The handle to this client
 204    */
 205   struct GNUNET_SERVICE_Client *client;
 206
 207   /**
 208    * The message queue to this client
 209    */
 210   struct GNUNET_MQ_Handle *mq;
 211 };
 212
 213 /**
 214  * Our handle to the BLOCK library.
 215  */
 216 struct GNUNET_BLOCK_Context *GDS_block_context;
 217
 218 /**
 219  * Handle for the statistics service.
 220  */
 221 struct GNUNET_STATISTICS_Handle *GDS_stats;
 222
 223 /**
 224  * Handle for the service.
 225  */
 226 struct GNUNET_SERVICE_Handle *GDS_service;
 227
 228 /**
 229  * The configuration the DHT service is running with
 230  */
 231 const struct GNUNET_CONFIGURATION_Handle *GDS_cfg;
 232
 233 /**
 234  * List of active monitoring requests.
 235  */
 236 static struct ClientMonitorRecord *monitor_head;
 237
 238 /**
 239  * List of active monitoring requests.
 240  */
 241 static struct ClientMonitorRecord *monitor_tail;
 242
 243 /**
 244  * Hashmap for fast key based lookup, maps keys to `struct ClientQueryRecord` entries.
 245  */
 246 static struct GNUNET_CONTAINER_MultiHashMap *forward_map;
 247
 248 /**
 249  * Heap with all of our client's request, sorted by retry time (earliest on top).
 250  */
 251 static struct GNUNET_CONTAINER_Heap *retry_heap;
 252
 253 /**
 254  * Task that re-transmits requests (using retry_heap).
 255  */
 256 static struct GNUNET_SCHEDULER_Task *retry_task;
 257
 258
 259 /**
 260  * Free data structures associated with the given query.
 261  *
 262  * @param record record to remove
 263  */
 264 static void
 265 remove_client_record (struct ClientQueryRecord *record)
 266 {
 267   struct ClientHandle *ch = record->ch;
 268
 269   GNUNET_CONTAINER_DLL_remove (ch->cqr_head,
 270                                ch->cqr_tail,
 271                                record);
 272   GNUNET_assert (GNUNET_YES ==
 273                  GNUNET_CONTAINER_multihashmap_remove (forward_map,
 274                                                        &record->key,
 275                                                        record));
 276   if (NULL != record->hnode)
 277     GNUNET_CONTAINER_heap_remove_node (record->hnode);
 278   GNUNET_array_grow (record->seen_replies,
 279                      record->seen_replies_count,
 280                      0);
 281   GNUNET_free (record);
 282 }
 283
 284
 285 /**
 286  * Functions with this signature are called whenever a local client is
 287  * connects to us.
 288  *
 289  * @param cls closure (NULL for dht)
 290  * @param client identification of the client
 291  * @param mq message queue for talking to @a client
 292  * @return our `struct ClientHandle` for @a client
 293  */
 294 static void *
 295 client_connect_cb (void *cls,
 296                    struct GNUNET_SERVICE_Client *client,
 297                    struct GNUNET_MQ_Handle *mq)
 298 {
 299   struct ClientHandle *ch;
 300
 301   ch = GNUNET_new (struct ClientHandle);
 302   ch->client = client;
 303   ch->mq = mq;
 304   return ch;
 305 }
 306
 307
 308 /**
 309  * Functions with this signature are called whenever a client
 310  * is disconnected on the network level.
 311  *
 312  * @param cls closure (NULL for dht)
 313  * @param client identification of the client
 314  * @param app_ctx our `struct ClientHandle` for @a client
 315  */
 316 static void
 317 client_disconnect_cb (void *cls,
 318                       struct GNUNET_SERVICE_Client *client,
 319                       void *app_ctx)
 320 {
 321   struct ClientHandle *ch = app_ctx;
 322   struct ClientQueryRecord *cqr;
 323   struct ClientMonitorRecord *monitor;
 324
 325   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 326               "Local client %p disconnects\n",
 327               ch);
 328   monitor = monitor_head;
 329   while (NULL != monitor)
 330   {
 331     if (monitor->ch == ch)
 332     {
 333       struct ClientMonitorRecord *next;
 334
 335       next = monitor->next;
 336       GNUNET_free_non_null (monitor->key);
 337       GNUNET_CONTAINER_DLL_remove (monitor_head,
 338                                    monitor_tail,
 339                                    monitor);
 340       GNUNET_free (monitor);
 341       monitor = next;
 342     }
 343     else
 344     {
 345       monitor = monitor->next;
 346     }
 347   }
 348   while (NULL != (cqr = ch->cqr_head))
 349     remove_client_record (cqr);
 350   GNUNET_free (ch);
 351 }
 352
 353
 354 /**
 355  * Route the given request via the DHT.  This includes updating
 356  * the bloom filter and retransmission times, building the P2P
 357  * message and initiating the routing operation.
 358  */
 359 static void
 360 transmit_request (struct ClientQueryRecord *cqr)
 361 {
 362   struct GNUNET_BLOCK_Group *bg;
 363   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 364
 365   GNUNET_STATISTICS_update (GDS_stats,
 366                             gettext_noop (
 367                               "# GET requests from clients injected"),
 368                             1,
 369                             GNUNET_NO);
 370   bg = GNUNET_BLOCK_group_create (GDS_block_context,
 371                                   cqr->type,
 372                                   GNUNET_CRYPTO_random_u32 (
 373                                     GNUNET_CRYPTO_QUALITY_WEAK,
 374                                     UINT32_MAX),
 375                                   NULL,
 376                                   0,
 377                                   "seen-set-size",
 378                                   cqr->seen_replies_count,
 379                                   NULL);
 380   GNUNET_BLOCK_group_set_seen (bg,
 381                                cqr->seen_replies,
 382                                cqr->seen_replies_count);
 383   peer_bf
 384     = GNUNET_CONTAINER_bloomfilter_init (NULL,
 385                                          DHT_BLOOM_SIZE,
 386                                          GNUNET_CONSTANTS_BLOOMFILTER_K);
 387   LOG (GNUNET_ERROR_TYPE_DEBUG,
 388        "Initiating GET for %s, replication %u, already have %u replies\n",
 389        GNUNET_h2s (&cqr->key),
 390        cqr->replication,
 391        cqr->seen_replies_count);
 392   GDS_NEIGHBOURS_handle_get (cqr->type,
 393                              cqr->msg_options,
 394                              cqr->replication,
 395                              0 /* hop count */,
 396                              &cqr->key,
 397                              cqr->xquery,
 398                              cqr->xquery_size,
 399                              bg,
 400                              peer_bf);
 401   GNUNET_BLOCK_group_destroy (bg);
 402   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 403
 404   /* exponential back-off for retries.
 405    * max GNUNET_TIME_STD_EXPONENTIAL_BACKOFF_THRESHOLD (15 min) */
 406   cqr->retry_frequency = GNUNET_TIME_STD_BACKOFF (cqr->retry_frequency);
 407   cqr->retry_time = GNUNET_TIME_relative_to_absolute (cqr->retry_frequency);
 408 }
 409
 410
 411 /**
 412  * Task that looks at the #retry_heap and transmits all of the requests
 413  * on the heap that are ready for transmission.  Then re-schedules
 414  * itself (unless the heap is empty).
 415  *
 416  * @param cls unused
 417  */
 418 static void
 419 transmit_next_request_task (void *cls)
 420 {
 421   struct ClientQueryRecord *cqr;
 422   struct GNUNET_TIME_Relative delay;
 423
 424   retry_task = NULL;
 425   while (NULL != (cqr = GNUNET_CONTAINER_heap_remove_root (retry_heap)))
 426   {
 427     cqr->hnode = NULL;
 428     delay = GNUNET_TIME_absolute_get_remaining (cqr->retry_time);
 429     if (delay.rel_value_us > 0)
 430     {
 431       cqr->hnode
 432         = GNUNET_CONTAINER_heap_insert (retry_heap,
 433                                         cqr,
 434                                         cqr->retry_time.abs_value_us);
 435       retry_task
 436         = GNUNET_SCHEDULER_add_at (cqr->retry_time,
 437                                    &transmit_next_request_task,
 438                                    NULL);
 439       return;
 440     }
 441     transmit_request (cqr);
 442     cqr->hnode
 443       = GNUNET_CONTAINER_heap_insert (retry_heap,
 444                                       cqr,
 445                                       cqr->retry_time.abs_value_us);
 446   }
 447 }
 448
 449
 450 /**
 451  * Check DHT PUT messages from the client.
 452  *
 453  * @param cls the client we received this message from
 454  * @param dht_msg the actual message received
 455  * @return #GNUNET_OK (always)
 456  */
 457 static int
 458 check_dht_local_put (void *cls,
 459                      const struct GNUNET_DHT_ClientPutMessage *dht_msg)
 460 {
 461   /* always well-formed */
 462   return GNUNET_OK;
 463 }
 464
 465
 466 /**
 467  * Handler for PUT messages.
 468  *
 469  * @param cls the client we received this message from
 470  * @param dht_msg the actual message received
 471  */
 472 static void
 473 handle_dht_local_put (void *cls,
 474                       const struct GNUNET_DHT_ClientPutMessage *dht_msg)
 475 {
 476   struct ClientHandle *ch = cls;
 477   struct GNUNET_CONTAINER_BloomFilter *peer_bf;
 478   uint16_t size;
 479
 480   size = ntohs (dht_msg->header.size);
 481   GNUNET_STATISTICS_update (GDS_stats,
 482                             gettext_noop (
 483                               "# PUT requests received from clients"),
 484                             1,
 485                             GNUNET_NO);
 486   LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
 487                "CLIENT-PUT %s\n",
 488                GNUNET_h2s_full (&dht_msg->key));
 489   /* give to local clients */
 490   LOG (GNUNET_ERROR_TYPE_DEBUG,
 491        "Handling local PUT of %u-bytes for query %s\n",
 492        size - sizeof(struct GNUNET_DHT_ClientPutMessage),
 493        GNUNET_h2s (&dht_msg->key));
 494   GDS_CLIENTS_handle_reply (GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 495                             &dht_msg->key,
 496                             0,
 497                             NULL,
 498                             0,
 499                             NULL,
 500                             ntohl (dht_msg->type),
 501                             size - sizeof(struct GNUNET_DHT_ClientPutMessage),
 502                             &dht_msg[1]);
 503   /* store locally */
 504   GDS_DATACACHE_handle_put (GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 505                             &dht_msg->key,
 506                             0,
 507                             NULL,
 508                             ntohl (dht_msg->type),
 509                             size - sizeof(struct GNUNET_DHT_ClientPutMessage),
 510                             &dht_msg[1]);
 511   /* route to other peers */
 512   peer_bf
 513     = GNUNET_CONTAINER_bloomfilter_init (NULL,
 514                                          DHT_BLOOM_SIZE,
 515                                          GNUNET_CONSTANTS_BLOOMFILTER_K);
 516   GDS_NEIGHBOURS_handle_put (ntohl (dht_msg->type),
 517                              ntohl (dht_msg->options),
 518                              ntohl (dht_msg->desired_replication_level),
 519                              GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 520                              0 /* hop count */,
 521                              peer_bf,
 522                              &dht_msg->key,
 523                              0,
 524                              NULL,
 525                              &dht_msg[1],
 526                              size - sizeof(struct GNUNET_DHT_ClientPutMessage));
 527   GDS_CLIENTS_process_put (ntohl (dht_msg->options),
 528                            ntohl (dht_msg->type),
 529                            0,
 530                            ntohl (dht_msg->desired_replication_level),
 531                            1,
 532                            GDS_NEIGHBOURS_get_id (),
 533                            GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
 534                            &dht_msg->key,
 535                            &dht_msg[1],
 536                            size - sizeof(struct GNUNET_DHT_ClientPutMessage));
 537   GNUNET_CONTAINER_bloomfilter_free (peer_bf);
 538   GNUNET_SERVICE_client_continue (ch->client);
 539 }
 540
 541
 542 /**
 543  * Check DHT GET messages from the client.
 544  *
 545  * @param cls the client we received this message from
 546  * @param message the actual message received
 547  * @return #GNUNET_OK (always)
 548  */
 549 static int
 550 check_dht_local_get (void *cls,
 551                      const struct GNUNET_DHT_ClientGetMessage *get)
 552 {
 553   /* always well-formed */
 554   return GNUNET_OK;
 555 }
 556
 557
 558 /**
 559  * Handle a result from local datacache for a GET operation.
 560  *
 561  * @param cls the `struct ClientHandle` of the client doing the query
 562  * @param type type of the block
 563  * @param expiration_time when does the content expire
 564  * @param key key for the content
 565  * @param put_path_length number of entries in @a put_path
 566  * @param put_path peers the original PUT traversed (if tracked)
 567  * @param get_path_length number of entries in @a get_path
 568  * @param get_path peers this reply has traversed so far (if tracked)
 569  * @param data payload of the reply
 570  * @param data_size number of bytes in @a data
 571  */
 572 static void
 573 handle_local_result (void *cls,
 574                      enum GNUNET_BLOCK_Type type,
 575                      struct GNUNET_TIME_Absolute expiration_time,
 576                      const struct GNUNET_HashCode *key,
 577                      unsigned int put_path_length,
 578                      const struct GNUNET_PeerIdentity *put_path,
 579                      unsigned int get_path_length,
 580                      const struct GNUNET_PeerIdentity *get_path,
 581                      const void *data,
 582                      size_t data_size)
 583 {
 584   // FIXME: this needs some clean up: inline the function,
 585   // possibly avoid even looking up the client!
 586   GDS_CLIENTS_handle_reply (expiration_time,
 587                             key,
 588                             0, NULL,
 589                             put_path_length, put_path,
 590                             type,
 591                             data_size, data);
 592 }
 593
 594
 595 /**
 596  * Handler for DHT GET messages from the client.
 597  *
 598  * @param cls the client we received this message from
 599  * @param message the actual message received
 600  */
 601 static void
 602 handle_dht_local_get (void *cls,
 603                       const struct GNUNET_DHT_ClientGetMessage *get)
 604 {
 605   struct ClientHandle *ch = cls;
 606   struct ClientQueryRecord *cqr;
 607   size_t xquery_size;
 608   const char *xquery;
 609   uint16_t size;
 610
 611   size = ntohs (get->header.size);
 612   xquery_size = size - sizeof(struct GNUNET_DHT_ClientGetMessage);
 613   xquery = (const char *) &get[1];
 614   GNUNET_STATISTICS_update (GDS_stats,
 615                             gettext_noop
 616                               ("# GET requests received from clients"), 1,
 617                             GNUNET_NO);
 618   LOG (GNUNET_ERROR_TYPE_DEBUG,
 619        "Received GET request for %s from local client %p, xq: %.*s\n",
 620        GNUNET_h2s (&get->key),
 621        ch->client,
 622        xquery_size,
 623        xquery);
 624   LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
 625                "CLIENT-GET %s\n",
 626                GNUNET_h2s_full (&get->key));
 627
 628   cqr = GNUNET_malloc (sizeof(struct ClientQueryRecord) + xquery_size);
 629   cqr->key = get->key;
 630   cqr->ch = ch;
 631   cqr->xquery = (void *) &cqr[1];
 632   GNUNET_memcpy (&cqr[1], xquery, xquery_size);
 633   cqr->hnode = GNUNET_CONTAINER_heap_insert (retry_heap, cqr, 0);
 634   cqr->retry_frequency = GNUNET_TIME_UNIT_SECONDS;
 635   cqr->retry_time = GNUNET_TIME_absolute_get ();
 636   cqr->unique_id = get->unique_id;
 637   cqr->xquery_size = xquery_size;
 638   cqr->replication = ntohl (get->desired_replication_level);
 639   cqr->msg_options = ntohl (get->options);
 640   cqr->type = ntohl (get->type);
 641   GNUNET_CONTAINER_DLL_insert (ch->cqr_head,
 642                                ch->cqr_tail,
 643                                cqr);
 644   GNUNET_CONTAINER_multihashmap_put (forward_map,
 645                                      &cqr->key,
 646                                      cqr,
 647                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 648   GDS_CLIENTS_process_get (ntohl (get->options),
 649                            ntohl (get->type),
 650                            0,
 651                            ntohl (get->desired_replication_level),
 652                            1,
 653                            GDS_NEIGHBOURS_get_id (),
 654                            &get->key);
 655   /* start remote requests */
 656   if (NULL != retry_task)
 657     GNUNET_SCHEDULER_cancel (retry_task);
 658   retry_task = GNUNET_SCHEDULER_add_now (&transmit_next_request_task,
 659                                          NULL);
 660   /* perform local lookup */
 661   GDS_DATACACHE_handle_get (&get->key,
 662                             cqr->type,
 663                             cqr->xquery,
 664                             xquery_size,
 665                             NULL,
 666                             &handle_local_result,
 667                             ch);
 668   GNUNET_SERVICE_client_continue (ch->client);
 669 }
 670
 671
 672 /**
 673  * Closure for #find_by_unique_id().
 674  */
 675 struct FindByUniqueIdContext
 676 {
 677   /**
 678    * Where to store the result, if found.
 679    */
 680   struct ClientQueryRecord *cqr;
 681
 682   uint64_t unique_id;
 683 };
 684
 685
 686 /**
 687  * Function called for each existing DHT record for the given
 688  * query.  Checks if it matches the UID given in the closure
 689  * and if so returns the entry as a result.
 690  *
 691  * @param cls the search context
 692  * @param key query for the lookup (not used)
 693  * @param value the `struct ClientQueryRecord`
 694  * @return #GNUNET_YES to continue iteration (result not yet found)
 695  */
 696 static int
 697 find_by_unique_id (void *cls,
 698                    const struct GNUNET_HashCode *key,
 699                    void *value)
 700 {
 701   struct FindByUniqueIdContext *fui_ctx = cls;
 702   struct ClientQueryRecord *cqr = value;
 703
 704   if (cqr->unique_id != fui_ctx->unique_id)
 705     return GNUNET_YES;
 706   fui_ctx->cqr = cqr;
 707   return GNUNET_NO;
 708 }
 709
 710
 711 /**
 712  * Check "GET result seen" messages from the client.
 713  *
 714  * @param cls the client we received this message from
 715  * @param message the actual message received
 716  * @return #GNUNET_OK if @a seen is well-formed
 717  */
 718 static int
 719 check_dht_local_get_result_seen (void *cls,
 720                                  const struct
 721                                  GNUNET_DHT_ClientGetResultSeenMessage *seen)
 722 {
 723   uint16_t size;
 724   unsigned int hash_count;
 725
 726   size = ntohs (seen->header.size);
 727   hash_count = (size - sizeof(struct GNUNET_DHT_ClientGetResultSeenMessage))
 728                / sizeof(struct GNUNET_HashCode);
 729   if (size != sizeof(struct GNUNET_DHT_ClientGetResultSeenMessage)
 730       + hash_count * sizeof(struct GNUNET_HashCode))
 731   {
 732     GNUNET_break (0);
 733     return GNUNET_SYSERR;
 734   }
 735   return GNUNET_OK;
 736 }
 737
 738
 739 /**
 740  * Handler for "GET result seen" messages from the client.
 741  *
 742  * @param cls the client we received this message from
 743  * @param message the actual message received
 744  */
 745 static void
 746 handle_dht_local_get_result_seen (void *cls,
 747                                   const struct
 748                                   GNUNET_DHT_ClientGetResultSeenMessage *seen)
 749 {
 750   struct ClientHandle *ch = cls;
 751   uint16_t size;
 752   unsigned int hash_count;
 753   unsigned int old_count;
 754   const struct GNUNET_HashCode *hc;
 755   struct FindByUniqueIdContext fui_ctx;
 756   struct ClientQueryRecord *cqr;
 757
 758   size = ntohs (seen->header.size);
 759   hash_count = (size - sizeof(struct GNUNET_DHT_ClientGetResultSeenMessage))
 760                / sizeof(struct GNUNET_HashCode);
 761   hc = (const struct GNUNET_HashCode*) &seen[1];
 762   fui_ctx.unique_id = seen->unique_id;
 763   fui_ctx.cqr = NULL;
 764   GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
 765                                               &seen->key,
 766                                               &find_by_unique_id,
 767                                               &fui_ctx);
 768   if (NULL == (cqr = fui_ctx.cqr))
 769   {
 770     GNUNET_break (0);
 771     GNUNET_SERVICE_client_drop (ch->client);
 772     return;
 773   }
 774   /* finally, update 'seen' list */
 775   old_count = cqr->seen_replies_count;
 776   GNUNET_array_grow (cqr->seen_replies,
 777                      cqr->seen_replies_count,
 778                      cqr->seen_replies_count + hash_count);
 779   GNUNET_memcpy (&cqr->seen_replies[old_count],
 780                  hc,
 781                  sizeof(struct GNUNET_HashCode) * hash_count);
 782 }
 783
 784
 785 /**
 786  * Closure for #remove_by_unique_id().
 787  */
 788 struct RemoveByUniqueIdContext
 789 {
 790   /**
 791    * Client that issued the removal request.
 792    */
 793   struct ClientHandle *ch;
 794
 795   /**
 796    * Unique ID of the request.
 797    */
 798   uint64_t unique_id;
 799 };
 800
 801
 802 /**
 803  * Iterator over hash map entries that frees all entries
 804  * that match the given client and unique ID.
 805  *
 806  * @param cls unique ID and client to search for in source routes
 807  * @param key current key code
 808  * @param value value in the hash map, a ClientQueryRecord
 809  * @return #GNUNET_YES (we should continue to iterate)
 810  */
 811 static int
 812 remove_by_unique_id (void *cls,
 813                      const struct GNUNET_HashCode *key,
 814                      void *value)
 815 {
 816   const struct RemoveByUniqueIdContext *ctx = cls;
 817   struct ClientQueryRecord *cqr = value;
 818
 819   if (cqr->unique_id != ctx->unique_id)
 820     return GNUNET_YES;
 821   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 822               "Removing client %p's record for key %s (by unique id)\n",
 823               ctx->ch->client,
 824               GNUNET_h2s (key));
 825   remove_client_record (cqr);
 826   return GNUNET_YES;
 827 }
 828
 829
 830 /**
 831  * Handler for any generic DHT stop messages, calls the appropriate handler
 832  * depending on message type (if processed locally)
 833  *
 834  * @param cls client we received this message from
 835  * @param message the actual message received
 836  *
 837  */
 838 static void
 839 handle_dht_local_get_stop (void *cls,
 840                            const struct
 841                            GNUNET_DHT_ClientGetStopMessage *dht_stop_msg)
 842 {
 843   struct ClientHandle *ch = cls;
 844   struct RemoveByUniqueIdContext ctx;
 845
 846   GNUNET_STATISTICS_update (GDS_stats,
 847                             gettext_noop
 848                               ("# GET STOP requests received from clients"), 1,
 849                             GNUNET_NO);
 850   LOG (GNUNET_ERROR_TYPE_DEBUG,
 851        "Received GET STOP request for %s from local client %p\n",
 852        GNUNET_h2s (&dht_stop_msg->key),
 853        ch->client);
 854   ctx.ch = ch;
 855   ctx.unique_id = dht_stop_msg->unique_id;
 856   GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
 857                                               &dht_stop_msg->key,
 858                                               &remove_by_unique_id,
 859                                               &ctx);
 860   GNUNET_SERVICE_client_continue (ch->client);
 861 }
 862
 863
 864 /**
 865  * Handler for monitor start messages
 866  *
 867  * @param cls the client we received this message from
 868  * @param msg the actual message received
 869  *
 870  */
 871 static void
 872 handle_dht_local_monitor (void *cls,
 873                           const struct GNUNET_DHT_MonitorStartStopMessage *msg)
 874 {
 875   struct ClientHandle *ch = cls;
 876   struct ClientMonitorRecord *r;
 877
 878   r = GNUNET_new (struct ClientMonitorRecord);
 879   r->ch = ch;
 880   r->type = ntohl (msg->type);
 881   r->get = ntohs (msg->get);
 882   r->get_resp = ntohs (msg->get_resp);
 883   r->put = ntohs (msg->put);
 884   if (0 == ntohs (msg->filter_key))
 885   {
 886     r->key = NULL;
 887   }
 888   else
 889   {
 890     r->key = GNUNET_new (struct GNUNET_HashCode);
 891     GNUNET_memcpy (r->key,
 892                    &msg->key,
 893                    sizeof(struct GNUNET_HashCode));
 894   }
 895   GNUNET_CONTAINER_DLL_insert (monitor_head,
 896                                monitor_tail,
 897                                r);
 898   GNUNET_SERVICE_client_continue (ch->client);
 899 }
 900
 901
 902 /**
 903  * Handler for monitor stop messages
 904  *
 905  * @param cls the client we received this message from
 906  * @param msg the actual message received
 907  */
 908 static void
 909 handle_dht_local_monitor_stop (void *cls,
 910                                const struct
 911                                GNUNET_DHT_MonitorStartStopMessage *msg)
 912 {
 913   struct ClientHandle *ch = cls;
 914   struct ClientMonitorRecord *r;
 915   int keys_match;
 916
 917   GNUNET_SERVICE_client_continue (ch->client);
 918   for (r = monitor_head; NULL != r; r = r->next)
 919   {
 920     if (NULL == r->key)
 921     {
 922       keys_match = (0 == ntohs (msg->filter_key));
 923     }
 924     else
 925     {
 926       keys_match = ((0 != ntohs (msg->filter_key)) &&
 927                     (! memcmp (r->key,
 928                                &msg->key,
 929                                sizeof(struct GNUNET_HashCode))));
 930     }
 931     if ((ch == r->ch) &&
 932         (ntohl (msg->type) == r->type) &&
 933         (r->get == msg->get) &&
 934         (r->get_resp == msg->get_resp) &&
 935         (r->put == msg->put) &&
 936         keys_match)
 937     {
 938       GNUNET_CONTAINER_DLL_remove (monitor_head,
 939                                    monitor_tail,
 940                                    r);
 941       GNUNET_free_non_null (r->key);
 942       GNUNET_free (r);
 943       return;     /* Delete only ONE entry */
 944     }
 945   }
 946 }
 947
 948
 949 /**
 950  * Closure for #forward_reply()
 951  */
 952 struct ForwardReplyContext
 953 {
 954   /**
 955    * Expiration time of the reply.
 956    */
 957   struct GNUNET_TIME_Absolute expiration;
 958
 959   /**
 960    * GET path taken.
 961    */
 962   const struct GNUNET_PeerIdentity *get_path;
 963
 964   /**
 965    * PUT path taken.
 966    */
 967   const struct GNUNET_PeerIdentity *put_path;
 968
 969   /**
 970    * Embedded payload.
 971    */
 972   const void *data;
 973
 974   /**
 975    * Number of bytes in data.
 976    */
 977   size_t data_size;
 978
 979   /**
 980    * Number of entries in @e get_path.
 981    */
 982   unsigned int get_path_length;
 983
 984   /**
 985    * Number of entries in @e put_path.
 986    */
 987   unsigned int put_path_length;
 988
 989   /**
 990    * Type of the data.
 991    */
 992   enum GNUNET_BLOCK_Type type;
 993 };
 994
 995
 996 /**
 997  * Iterator over hash map entries that send a given reply to
 998  * each of the matching clients.  With some tricky recycling
 999  * of the buffer.
1000  *
1001  * @param cls the 'struct ForwardReplyContext'
1002  * @param key current key
1003  * @param value value in the hash map, a ClientQueryRecord
1004  * @return #GNUNET_YES (we should continue to iterate),
1005  *         if the result is mal-formed, #GNUNET_NO
1006  */
1007 static int
1008 forward_reply (void *cls,
1009                const struct GNUNET_HashCode *key,
1010                void *value)
1011 {
1012   struct ForwardReplyContext *frc = cls;
1013   struct ClientQueryRecord *record = value;
1014   struct GNUNET_MQ_Envelope *env;
1015   struct GNUNET_DHT_ClientResultMessage *reply;
1016   enum GNUNET_BLOCK_EvaluationResult eval;
1017   int do_free;
1018   struct GNUNET_HashCode ch;
1019   struct GNUNET_PeerIdentity *paths;
1020
1021   LOG_TRAFFIC (GNUNET_ERROR_TYPE_DEBUG,
1022                "CLIENT-RESULT %s\n",
1023                GNUNET_h2s_full (key));
1024   if ((record->type != GNUNET_BLOCK_TYPE_ANY) &&
1025       (record->type != frc->type))
1026   {
1027     LOG (GNUNET_ERROR_TYPE_DEBUG,
1028          "Record type mismatch, not passing request for key %s to local client\n",
1029          GNUNET_h2s (key));
1030     GNUNET_STATISTICS_update (GDS_stats,
1031                               gettext_noop
1032                               (
1033                                 "# Key match, type mismatches in REPLY to CLIENT"),
1034                               1, GNUNET_NO);
1035     return GNUNET_YES;          /* type mismatch */
1036   }
1037   GNUNET_CRYPTO_hash (frc->data, frc->data_size, &ch);
1038   for (unsigned int i = 0; i < record->seen_replies_count; i++)
1039     if (0 == memcmp (&record->seen_replies[i],
1040                      &ch,
1041                      sizeof(struct GNUNET_HashCode)))
1042     {
1043       LOG (GNUNET_ERROR_TYPE_DEBUG,
1044            "Duplicate reply, not passing request for key %s to local client\n",
1045            GNUNET_h2s (key));
1046       GNUNET_STATISTICS_update (GDS_stats,
1047                                 gettext_noop
1048                                 (
1049                                   "# Duplicate REPLIES to CLIENT request dropped"),
1050                                 1, GNUNET_NO);
1051       return GNUNET_YES;        /* duplicate */
1052     }
1053   eval
1054     = GNUNET_BLOCK_evaluate (GDS_block_context,
1055                              record->type,
1056                              NULL,
1057                              GNUNET_BLOCK_EO_NONE,
1058                              key,
1059                              record->xquery,
1060                              record->xquery_size,
1061                              frc->data,
1062                              frc->data_size);
1063   LOG (GNUNET_ERROR_TYPE_DEBUG,
1064        "Evaluation result is %d for key %s for local client's query\n",
1065        (int) eval,
1066        GNUNET_h2s (key));
1067   switch (eval)
1068   {
1069   case GNUNET_BLOCK_EVALUATION_OK_LAST:
1070     do_free = GNUNET_YES;
1071     break;
1072
1073   case GNUNET_BLOCK_EVALUATION_OK_MORE:
1074     GNUNET_array_append (record->seen_replies,
1075                          record->seen_replies_count,
1076                          ch);
1077     do_free = GNUNET_NO;
1078     break;
1079
1080   case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
1081     /* should be impossible to encounter here */
1082     GNUNET_break (0);
1083     return GNUNET_YES;
1084
1085   case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
1086     GNUNET_break_op (0);
1087     return GNUNET_NO;
1088
1089   case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
1090     GNUNET_break (0);
1091     return GNUNET_NO;
1092
1093   case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
1094     GNUNET_break (0);
1095     return GNUNET_NO;
1096
1097   case GNUNET_BLOCK_EVALUATION_RESULT_IRRELEVANT:
1098     return GNUNET_YES;
1099
1100   case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
1101     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1102                 _ ("Unsupported block type (%u) in request!\n"), record->type);
1103     return GNUNET_NO;
1104
1105   default:
1106     GNUNET_break (0);
1107     return GNUNET_NO;
1108   }
1109   GNUNET_STATISTICS_update (GDS_stats,
1110                             gettext_noop ("# RESULTS queued for clients"),
1111                             1,
1112                             GNUNET_NO);
1113   env = GNUNET_MQ_msg_extra (reply,
1114                              frc->data_size
1115                              + (frc->get_path_length + frc->put_path_length)
1116                              * sizeof(struct GNUNET_PeerIdentity),
1117                              GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT);
1118   reply->type = htonl (frc->type);
1119   reply->get_path_length = htonl (frc->get_path_length);
1120   reply->put_path_length = htonl (frc->put_path_length);
1121   reply->unique_id = record->unique_id;
1122   reply->expiration = GNUNET_TIME_absolute_hton (frc->expiration);
1123   reply->key = *key;
1124   paths = (struct GNUNET_PeerIdentity *) &reply[1];
1125   GNUNET_memcpy (paths,
1126                  frc->put_path,
1127                  sizeof(struct GNUNET_PeerIdentity) * frc->put_path_length);
1128   GNUNET_memcpy (&paths[frc->put_path_length],
1129                  frc->get_path,
1130                  sizeof(struct GNUNET_PeerIdentity) * frc->get_path_length);
1131   GNUNET_memcpy (&paths[frc->get_path_length + frc->put_path_length],
1132                  frc->data,
1133                  frc->data_size);
1134   LOG (GNUNET_ERROR_TYPE_DEBUG,
1135        "Sending reply to query %s for client %p\n",
1136        GNUNET_h2s (key),
1137        record->ch->client);
1138   GNUNET_MQ_send (record->ch->mq,
1139                   env);
1140   if (GNUNET_YES == do_free)
1141     remove_client_record (record);
1142   return GNUNET_YES;
1143 }
1144
1145
1146 /**
1147  * Handle a reply we've received from another peer.  If the reply
1148  * matches any of our pending queries, forward it to the respective
1149  * client(s).
1150  *
1151  * @param expiration when will the reply expire
1152  * @param key the query this reply is for
1153  * @param get_path_length number of peers in @a get_path
1154  * @param get_path path the reply took on get
1155  * @param put_path_length number of peers in @a put_path
1156  * @param put_path path the reply took on put
1157  * @param type type of the reply
1158  * @param data_size number of bytes in @a data
1159  * @param data application payload data
1160  */
1161 void
1162 GDS_CLIENTS_handle_reply (struct GNUNET_TIME_Absolute expiration,
1163                           const struct GNUNET_HashCode *key,
1164                           unsigned int get_path_length,
1165                           const struct GNUNET_PeerIdentity *get_path,
1166                           unsigned int put_path_length,
1167                           const struct GNUNET_PeerIdentity *put_path,
1168                           enum GNUNET_BLOCK_Type type,
1169                           size_t data_size,
1170                           const void *data)
1171 {
1172   struct ForwardReplyContext frc;
1173   size_t msize;
1174
1175   msize = sizeof(struct GNUNET_DHT_ClientResultMessage) + data_size
1176           + (get_path_length + put_path_length) * sizeof(struct
1177                                                          GNUNET_PeerIdentity);
1178   if (msize >= GNUNET_MAX_MESSAGE_SIZE)
1179   {
1180     GNUNET_break (0);
1181     return;
1182   }
1183   if (NULL == GNUNET_CONTAINER_multihashmap_get (forward_map,
1184                                                  key))
1185   {
1186     LOG (GNUNET_ERROR_TYPE_DEBUG,
1187          "No matching client for reply for key %s\n",
1188          GNUNET_h2s (key));
1189     GNUNET_STATISTICS_update (GDS_stats,
1190                               gettext_noop (
1191                                 "# REPLIES ignored for CLIENTS (no match)"),
1192                               1,
1193                               GNUNET_NO);
1194     return;                     /* no matching request, fast exit! */
1195   }
1196   frc.expiration = expiration;
1197   frc.get_path = get_path;
1198   frc.put_path = put_path;
1199   frc.data = data;
1200   frc.data_size = data_size;
1201   frc.get_path_length = get_path_length;
1202   frc.put_path_length = put_path_length;
1203   frc.type = type;
1204   LOG (GNUNET_ERROR_TYPE_DEBUG,
1205        "Forwarding reply for key %s to client\n",
1206        GNUNET_h2s (key));
1207   GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
1208                                               key,
1209                                               &forward_reply,
1210                                               &frc);
1211 }
1212
1213
1214 /**
1215  * Check if some client is monitoring GET messages and notify
1216  * them in that case.
1217  *
1218  * @param options Options, for instance RecordRoute, DemultiplexEverywhere.
1219  * @param type The type of data in the request.
1220  * @param hop_count Hop count so far.
1221  * @param path_length number of entries in path (or 0 if not recorded).
1222  * @param path peers on the GET path (or NULL if not recorded).
1223  * @param desired_replication_level Desired replication level.
1224  * @param key Key of the requested data.
1225  */
1226 void
1227 GDS_CLIENTS_process_get (uint32_t options,
1228                          enum GNUNET_BLOCK_Type type,
1229                          uint32_t hop_count,
1230                          uint32_t desired_replication_level,
1231                          unsigned int path_length,
1232                          const struct GNUNET_PeerIdentity *path,
1233                          const struct GNUNET_HashCode *key)
1234 {
1235   struct ClientMonitorRecord *m;
1236   struct ClientHandle **cl;
1237   unsigned int cl_size;
1238
1239   cl = NULL;
1240   cl_size = 0;
1241   for (m = monitor_head; NULL != m; m = m->next)
1242   {
1243     if (((GNUNET_BLOCK_TYPE_ANY == m->type) ||
1244          (m->type == type)) &&
1245         ((NULL == m->key) ||
1246          (0 == memcmp (key,
1247                        m->key,
1248                        sizeof(struct GNUNET_HashCode)))))
1249     {
1250       struct GNUNET_MQ_Envelope *env;
1251       struct GNUNET_DHT_MonitorGetMessage *mmsg;
1252       struct GNUNET_PeerIdentity *msg_path;
1253       size_t msize;
1254       unsigned int i;
1255
1256       /* Don't send duplicates */
1257       for (i = 0; i < cl_size; i++)
1258         if (cl[i] == m->ch)
1259           break;
1260       if (i < cl_size)
1261         continue;
1262       GNUNET_array_append (cl,
1263                            cl_size,
1264                            m->ch);
1265
1266       msize = path_length * sizeof(struct GNUNET_PeerIdentity);
1267       env = GNUNET_MQ_msg_extra (mmsg,
1268                                  msize,
1269                                  GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET);
1270       mmsg->options = htonl (options);
1271       mmsg->type = htonl (type);
1272       mmsg->hop_count = htonl (hop_count);
1273       mmsg->desired_replication_level = htonl (desired_replication_level);
1274       mmsg->get_path_length = htonl (path_length);
1275       mmsg->key = *key;
1276       msg_path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1277       GNUNET_memcpy (msg_path,
1278                      path,
1279                      path_length * sizeof(struct GNUNET_PeerIdentity));
1280       GNUNET_MQ_send (m->ch->mq,
1281                       env);
1282     }
1283   }
1284   GNUNET_free_non_null (cl);
1285 }
1286
1287
1288 /**
1289  * Check if some client is monitoring GET RESP messages and notify
1290  * them in that case.
1291  *
1292  * @param type The type of data in the result.
1293  * @param get_path Peers on GET path (or NULL if not recorded).
1294  * @param get_path_length number of entries in get_path.
1295  * @param put_path peers on the PUT path (or NULL if not recorded).
1296  * @param put_path_length number of entries in get_path.
1297  * @param exp Expiration time of the data.
1298  * @param key Key of the data.
1299  * @param data Pointer to the result data.
1300  * @param size Number of bytes in @a data.
1301  */
1302 void
1303 GDS_CLIENTS_process_get_resp (enum GNUNET_BLOCK_Type type,
1304                               const struct GNUNET_PeerIdentity *get_path,
1305                               unsigned int get_path_length,
1306                               const struct GNUNET_PeerIdentity *put_path,
1307                               unsigned int put_path_length,
1308                               struct GNUNET_TIME_Absolute exp,
1309                               const struct GNUNET_HashCode *key,
1310                               const void *data,
1311                               size_t size)
1312 {
1313   struct ClientMonitorRecord *m;
1314   struct ClientHandle **cl;
1315   unsigned int cl_size;
1316
1317   cl = NULL;
1318   cl_size = 0;
1319   for (m = monitor_head; NULL != m; m = m->next)
1320   {
1321     if (((GNUNET_BLOCK_TYPE_ANY == m->type) || (m->type == type) ) &&
1322         ((NULL == m->key) ||
1323          (memcmp (key, m->key, sizeof(struct GNUNET_HashCode)) == 0) ))
1324     {
1325       struct GNUNET_MQ_Envelope *env;
1326       struct GNUNET_DHT_MonitorGetRespMessage *mmsg;
1327       struct GNUNET_PeerIdentity *path;
1328       size_t msize;
1329       unsigned int i;
1330
1331       /* Don't send duplicates */
1332       for (i = 0; i < cl_size; i++)
1333         if (cl[i] == m->ch)
1334           break;
1335       if (i < cl_size)
1336         continue;
1337       GNUNET_array_append (cl,
1338                            cl_size,
1339                            m->ch);
1340
1341       msize = size;
1342       msize += (get_path_length + put_path_length)
1343                * sizeof(struct GNUNET_PeerIdentity);
1344       env = GNUNET_MQ_msg_extra (mmsg,
1345                                  msize,
1346                                  GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET_RESP);
1347       mmsg->type = htonl (type);
1348       mmsg->put_path_length = htonl (put_path_length);
1349       mmsg->get_path_length = htonl (get_path_length);
1350       mmsg->expiration_time = GNUNET_TIME_absolute_hton (exp);
1351       mmsg->key = *key;
1352       path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1353       GNUNET_memcpy (path,
1354                      put_path,
1355                      put_path_length * sizeof(struct GNUNET_PeerIdentity));
1356       GNUNET_memcpy (path,
1357                      get_path,
1358                      get_path_length * sizeof(struct GNUNET_PeerIdentity));
1359       GNUNET_memcpy (&path[get_path_length],
1360                      data,
1361                      size);
1362       GNUNET_MQ_send (m->ch->mq,
1363                       env);
1364     }
1365   }
1366   GNUNET_free_non_null (cl);
1367 }
1368
1369
1370 /**
1371  * Check if some client is monitoring PUT messages and notify
1372  * them in that case.
1373  *
1374  * @param options Options, for instance RecordRoute, DemultiplexEverywhere.
1375  * @param type The type of data in the request.
1376  * @param hop_count Hop count so far.
1377  * @param path_length number of entries in path (or 0 if not recorded).
1378  * @param path peers on the PUT path (or NULL if not recorded).
1379  * @param desired_replication_level Desired replication level.
1380  * @param exp Expiration time of the data.
1381  * @param key Key under which data is to be stored.
1382  * @param data Pointer to the data carried.
1383  * @param size Number of bytes in data.
1384  */
1385 void
1386 GDS_CLIENTS_process_put (uint32_t options,
1387                          enum GNUNET_BLOCK_Type type,
1388                          uint32_t hop_count,
1389                          uint32_t desired_replication_level,
1390                          unsigned int path_length,
1391                          const struct GNUNET_PeerIdentity *path,
1392                          struct GNUNET_TIME_Absolute exp,
1393                          const struct GNUNET_HashCode *key,
1394                          const void *data,
1395                          size_t size)
1396 {
1397   struct ClientMonitorRecord *m;
1398   struct ClientHandle **cl;
1399   unsigned int cl_size;
1400
1401   cl = NULL;
1402   cl_size = 0;
1403   for (m = monitor_head; NULL != m; m = m->next)
1404   {
1405     if (((GNUNET_BLOCK_TYPE_ANY == m->type) || (m->type == type) ) &&
1406         ((NULL == m->key) ||
1407          (memcmp (key, m->key, sizeof(struct GNUNET_HashCode)) == 0) ))
1408     {
1409       struct GNUNET_MQ_Envelope *env;
1410       struct GNUNET_DHT_MonitorPutMessage *mmsg;
1411       struct GNUNET_PeerIdentity *msg_path;
1412       size_t msize;
1413       unsigned int i;
1414
1415       /* Don't send duplicates */
1416       for (i = 0; i < cl_size; i++)
1417         if (cl[i] == m->ch)
1418           break;
1419       if (i < cl_size)
1420         continue;
1421       GNUNET_array_append (cl,
1422                            cl_size,
1423                            m->ch);
1424
1425       msize = size;
1426       msize += path_length * sizeof(struct GNUNET_PeerIdentity);
1427       env = GNUNET_MQ_msg_extra (mmsg,
1428                                  msize,
1429                                  GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT);
1430       mmsg->options = htonl (options);
1431       mmsg->type = htonl (type);
1432       mmsg->hop_count = htonl (hop_count);
1433       mmsg->desired_replication_level = htonl (desired_replication_level);
1434       mmsg->put_path_length = htonl (path_length);
1435       mmsg->key = *key;
1436       mmsg->expiration_time = GNUNET_TIME_absolute_hton (exp);
1437       msg_path = (struct GNUNET_PeerIdentity *) &mmsg[1];
1438       GNUNET_memcpy (msg_path,
1439                      path,
1440                      path_length * sizeof(struct GNUNET_PeerIdentity));
1441       GNUNET_memcpy (&msg_path[path_length],
1442                      data,
1443                      size);
1444       GNUNET_MQ_send (m->ch->mq,
1445                       env);
1446     }
1447   }
1448   GNUNET_free_non_null (cl);
1449 }
1450
1451
1452 /**
1453  * Initialize client subsystem.
1454  *
1455  * @param server the initialized server
1456  */
1457 static void
1458 GDS_CLIENTS_init ()
1459 {
1460   forward_map
1461     = GNUNET_CONTAINER_multihashmap_create (1024,
1462                                             GNUNET_YES);
1463   retry_heap
1464     = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1465 }
1466
1467
1468 /**
1469  * Shutdown client subsystem.
1470  */
1471 static void
1472 GDS_CLIENTS_stop ()
1473 {
1474   if (NULL != retry_task)
1475   {
1476     GNUNET_SCHEDULER_cancel (retry_task);
1477     retry_task = NULL;
1478   }
1479 }
1480
1481
1482 /**
1483  * Define "main" method using service macro.
1484  *
1485  * @param name name of the service, i.e. "dht" or "xdht"
1486  * @param run name of the initializaton method for the service
1487  */
1488 #define GDS_DHT_SERVICE_INIT(name, run)          \
1489   GNUNET_SERVICE_MAIN \
1490     (name, \
1491     GNUNET_SERVICE_OPTION_NONE, \
1492     run, \
1493     &client_connect_cb, \
1494     &client_disconnect_cb, \
1495     NULL, \
1496     GNUNET_MQ_hd_var_size (dht_local_put, \
1497                            GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT, \
1498                            struct GNUNET_DHT_ClientPutMessage, \
1499                            NULL), \
1500     GNUNET_MQ_hd_var_size (dht_local_get, \
1501                            GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET, \
1502                            struct GNUNET_DHT_ClientGetMessage, \
1503                            NULL), \
1504     GNUNET_MQ_hd_fixed_size (dht_local_get_stop, \
1505                              GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP, \
1506                              struct GNUNET_DHT_ClientGetStopMessage, \
1507                              NULL), \
1508     GNUNET_MQ_hd_fixed_size (dht_local_monitor, \
1509                              GNUNET_MESSAGE_TYPE_DHT_MONITOR_START, \
1510                              struct GNUNET_DHT_MonitorStartStopMessage, \
1511                              NULL), \
1512     GNUNET_MQ_hd_fixed_size (dht_local_monitor_stop, \
1513                              GNUNET_MESSAGE_TYPE_DHT_MONITOR_STOP, \
1514                              struct GNUNET_DHT_MonitorStartStopMessage, \
1515                              NULL), \
1516     GNUNET_MQ_hd_var_size (dht_local_get_result_seen, \
1517                            GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_RESULTS_KNOWN, \
1518                            struct GNUNET_DHT_ClientGetResultSeenMessage, \
1519                            NULL), \
1520     GNUNET_MQ_handler_end ())
1521
1522
1523 /**
1524  * MINIMIZE heap size (way below 128k) since this process doesn't need much.
1525  */
1526 void __attribute__ ((destructor))
1527 GDS_CLIENTS_done ()
1528 {
1529   if (NULL != retry_heap)
1530   {
1531     GNUNET_assert (0 == GNUNET_CONTAINER_heap_get_size (retry_heap));
1532     GNUNET_CONTAINER_heap_destroy (retry_heap);
1533     retry_heap = NULL;
1534   }
1535   if (NULL != forward_map)
1536   {
1537     GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (forward_map));
1538     GNUNET_CONTAINER_multihashmap_destroy (forward_map);
1539     forward_map = NULL;
1540   }
1541 }
1542
1543
1544 /* end of gnunet-service-dht_clients.c */