src/fs/gnunet-service-fs_pr.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2013 GNUnet e.V.
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file fs/gnunet-service-fs_pr.c
  23  * @brief API to handle pending requests
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_load_lib.h"
  29 #include "gnunet-service-fs.h"
  30 #include "gnunet-service-fs_cp.h"
  31 #include "gnunet-service-fs_indexing.h"
  32 #include "gnunet-service-fs_pe.h"
  33 #include "gnunet-service-fs_pr.h"
  34 #include "gnunet-service-fs_cadet.h"
  35
  36
  37 /**
  38  * Desired replication level for GETs.
  39  */
  40 #define DHT_GET_REPLICATION 5
  41
  42 /**
  43  * Maximum size of the datastore queue for P2P operations.  Needs to
  44  * be large enough to queue #MAX_QUEUE_PER_PEER operations for roughly
  45  * the number of active (connected) peers.
  46  */
  47 #define MAX_DATASTORE_QUEUE (16 * MAX_QUEUE_PER_PEER)
  48
  49 /**
  50  * Bandwidth value of a 0-priority content (must be fairly high
  51  * compared to query since content is typically significantly larger
  52  * -- and more valueable since it can take many queries to get one
  53  * piece of content).
  54  */
  55 #define CONTENT_BANDWIDTH_VALUE 800
  56
  57 /**
  58  * Hard limit on the number of results we may get from the datastore per query.
  59  */
  60 #define MAX_RESULTS (100 * 1024)
  61
  62 /**
  63  * Collect an instane number of statistics?  May cause excessive IPC.
  64  */
  65 #define INSANE_STATISTICS GNUNET_NO
  66
  67 /**
  68  * If obtaining a block via cadet fails, how often do we retry it before
  69  * giving up for good (and sticking to non-anonymous transfer)?
  70  */
  71 #define CADET_RETRY_MAX 3
  72
  73
  74 /**
  75  * An active request.
  76  */
  77 struct GSF_PendingRequest
  78 {
  79   /**
  80    * Public data for the request.
  81    */
  82   struct GSF_PendingRequestData public_data;
  83
  84   /**
  85    * Function to call if we encounter a reply.
  86    */
  87   GSF_PendingRequestReplyHandler rh;
  88
  89   /**
  90    * Closure for @e rh
  91    */
  92   void *rh_cls;
  93
  94   /**
  95    * Array of hash codes of replies we've already seen.
  96    */
  97   struct GNUNET_HashCode *replies_seen;
  98
  99   /**
 100    * Block group for filtering replies we've already seen.
 101    */
 102   struct GNUNET_BLOCK_Group *bg;
 103
 104   /**
 105    * Entry for this pending request in the expiration heap, or NULL.
 106    */
 107   struct GNUNET_CONTAINER_HeapNode *hnode;
 108
 109   /**
 110    * Datastore queue entry for this request (or NULL for none).
 111    */
 112   struct GNUNET_DATASTORE_QueueEntry *qe;
 113
 114   /**
 115    * DHT request handle for this request (or NULL for none).
 116    */
 117   struct GNUNET_DHT_GetHandle *gh;
 118
 119   /**
 120    * Cadet request handle for this request (or NULL for none).
 121    */
 122   struct GSF_CadetRequest *cadet_request;
 123
 124   /**
 125    * Function to call upon completion of the local get
 126    * request, or NULL for none.
 127    */
 128   GSF_LocalLookupContinuation llc_cont;
 129
 130   /**
 131    * Closure for @e llc_cont.
 132    */
 133   void *llc_cont_cls;
 134
 135   /**
 136    * Last result from the local datastore lookup evaluation.
 137    */
 138   enum GNUNET_BLOCK_EvaluationResult local_result;
 139
 140   /**
 141    * Identity of the peer that we should use for the 'sender'
 142    * (recipient of the response) when forwarding (0 for none).
 143    */
 144   GNUNET_PEER_Id sender_pid;
 145
 146   /**
 147    * Identity of the peer that we should never forward this query
 148    * to since it originated this query (0 for none).
 149    */
 150   GNUNET_PEER_Id origin_pid;
 151
 152   /**
 153    * Time we started the last datastore lookup.
 154    */
 155   struct GNUNET_TIME_Absolute qe_start;
 156
 157   /**
 158    * Task that warns us if the local datastore lookup takes too long.
 159    */
 160   struct GNUNET_SCHEDULER_Task * warn_task;
 161
 162   /**
 163    * Current offset for querying our local datastore for results.
 164    * Starts at a random value, incremented until we get the same
 165    * UID again (detected using 'first_uid'), which is then used
 166    * to termiante the iteration.
 167    */
 168   uint64_t local_result_offset;
 169
 170   /**
 171    * Unique ID of the first result from the local datastore;
 172    * used to detect wrap-around of the offset.
 173    */
 174   uint64_t first_uid;
 175
 176   /**
 177    * How often have we retried this request via 'cadet'?
 178    * (used to bound overall retries).
 179    */
 180   unsigned int cadet_retry_count;
 181
 182   /**
 183    * Number of valid entries in the 'replies_seen' array.
 184    */
 185   unsigned int replies_seen_count;
 186
 187   /**
 188    * Length of the 'replies_seen' array.
 189    */
 190   unsigned int replies_seen_size;
 191
 192   /**
 193    * Do we have a first UID yet?
 194    */
 195   unsigned int have_first_uid;
 196
 197 };
 198
 199
 200 /**
 201  * All pending requests, ordered by the query.  Entries
 202  * are of type 'struct GSF_PendingRequest*'.
 203  */
 204 static struct GNUNET_CONTAINER_MultiHashMap *pr_map;
 205
 206
 207 /**
 208  * Datastore 'PUT' load tracking.
 209  */
 210 static struct GNUNET_LOAD_Value *datastore_put_load;
 211
 212
 213 /**
 214  * Are we allowed to migrate content to this peer.
 215  */
 216 static int active_to_migration;
 217
 218
 219 /**
 220  * Heap with the request that will expire next at the top.  Contains
 221  * pointers of type "struct PendingRequest*"; these will *also* be
 222  * aliased from the "requests_by_peer" data structures and the
 223  * "requests_by_query" table.  Note that requests from our clients
 224  * don't expire and are thus NOT in the "requests_by_expiration"
 225  * (or the "requests_by_peer" tables).
 226  */
 227 static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
 228
 229
 230 /**
 231  * Maximum number of requests (from other peers, overall) that we're
 232  * willing to have pending at any given point in time.  Can be changed
 233  * via the configuration file (32k is just the default).
 234  */
 235 static unsigned long long max_pending_requests = (32 * 1024);
 236
 237
 238
 239 /**
 240  * Recalculate our bloom filter for filtering replies.  This function
 241  * will create a new bloom filter from scratch, so it should only be
 242  * called if we have no bloomfilter at all (and hence can create a
 243  * fresh one of minimal size without problems) OR if our peer is the
 244  * initiator (in which case we may resize to larger than mimimum size).
 245  *
 246  * @param type type of the request
 247  * @param pr request for which the BF is to be recomputed
 248  */
 249 static void
 250 refresh_bloomfilter (enum GNUNET_BLOCK_Type type,
 251                      struct GSF_PendingRequest *pr)
 252 {
 253   if (NULL != pr->bg)
 254   {
 255     GNUNET_BLOCK_group_destroy (pr->bg);
 256     pr->bg = NULL;
 257   }
 258   if (GNUNET_BLOCK_TYPE_FS_UBLOCK != type)
 259     return; /* no need */
 260   pr->bg
 261     = GNUNET_BLOCK_group_create (GSF_block_ctx,
 262                                  type,
 263                                  GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 264                                                            UINT32_MAX),
 265                                  NULL,
 266                                  0,
 267                                  "fs-seen-set-size",
 268                                  pr->replies_seen_count);
 269   if (NULL == pr->bg)
 270     return;
 271   GNUNET_break (GNUNET_OK ==
 272                 GNUNET_BLOCK_group_set_seen (pr->bg,
 273                                              pr->replies_seen,
 274                                              pr->replies_seen_count));
 275 }
 276
 277
 278 /**
 279  * Create a new pending request.
 280  *
 281  * @param options request options
 282  * @param type type of the block that is being requested
 283  * @param query key for the lookup
 284  * @param target preferred target for the request, NULL for none
 285  * @param bf_data raw data for bloom filter for known replies, can be NULL
 286  * @param bf_size number of bytes in @a bf_data
 287  * @param mingle mingle value for bf
 288  * @param anonymity_level desired anonymity level
 289  * @param priority maximum outgoing cummulative request priority to use
 290  * @param ttl current time-to-live for the request
 291  * @param sender_pid peer ID to use for the sender when forwarding, 0 for none
 292  * @param origin_pid peer ID of origin of query (do not loop back)
 293  * @param replies_seen hash codes of known local replies
 294  * @param replies_seen_count size of the @a replies_seen array
 295  * @param rh handle to call when we get a reply
 296  * @param rh_cls closure for @a rh
 297  * @return handle for the new pending request
 298  */
 299 struct GSF_PendingRequest *
 300 GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
 301                              enum GNUNET_BLOCK_Type type,
 302                              const struct GNUNET_HashCode *query,
 303                              const struct GNUNET_PeerIdentity *target,
 304                              const char *bf_data,
 305                              size_t bf_size,
 306                              uint32_t mingle,
 307                              uint32_t anonymity_level,
 308                              uint32_t priority,
 309                              int32_t ttl,
 310                              GNUNET_PEER_Id sender_pid,
 311                              GNUNET_PEER_Id origin_pid,
 312                              const struct GNUNET_HashCode *replies_seen,
 313                              unsigned int replies_seen_count,
 314                              GSF_PendingRequestReplyHandler rh,
 315                              void *rh_cls)
 316 {
 317   struct GSF_PendingRequest *pr;
 318   struct GSF_PendingRequest *dpr;
 319   size_t extra;
 320   struct GNUNET_HashCode *eptr;
 321
 322   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 323               "Creating request handle for `%s' of type %d\n",
 324               GNUNET_h2s (query), type);
 325 #if INSANE_STATISTICS
 326   GNUNET_STATISTICS_update (GSF_stats,
 327                             gettext_noop ("# Pending requests created"), 1,
 328                             GNUNET_NO);
 329 #endif
 330   extra = 0;
 331   if (NULL != target)
 332     extra += sizeof (struct GNUNET_PeerIdentity);
 333   pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest) + extra);
 334   pr->local_result_offset =
 335       GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
 336   pr->public_data.query = *query;
 337   eptr = (struct GNUNET_HashCode *) &pr[1];
 338   if (NULL != target)
 339   {
 340     pr->public_data.target = (struct GNUNET_PeerIdentity *) eptr;
 341     GNUNET_memcpy (eptr,
 342             target,
 343             sizeof (struct GNUNET_PeerIdentity));
 344   }
 345   pr->public_data.anonymity_level = anonymity_level;
 346   pr->public_data.priority = priority;
 347   pr->public_data.original_priority = priority;
 348   pr->public_data.options = options;
 349   pr->public_data.type = type;
 350   pr->public_data.start_time = GNUNET_TIME_absolute_get ();
 351   pr->sender_pid = sender_pid;
 352   pr->origin_pid = origin_pid;
 353   pr->rh = rh;
 354   pr->rh_cls = rh_cls;
 355   GNUNET_assert ((sender_pid != 0) || (0 == (options & GSF_PRO_FORWARD_ONLY)));
 356   if (ttl >= 0)
 357     pr->public_data.ttl =
 358         GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_multiply
 359                                           (GNUNET_TIME_UNIT_SECONDS,
 360                                            (uint32_t) ttl));
 361   else
 362     pr->public_data.ttl =
 363         GNUNET_TIME_absolute_subtract (pr->public_data.start_time,
 364                                        GNUNET_TIME_relative_multiply
 365                                        (GNUNET_TIME_UNIT_SECONDS,
 366                                         (uint32_t) (-ttl)));
 367   if (replies_seen_count > 0)
 368   {
 369     pr->replies_seen_size = replies_seen_count;
 370     pr->replies_seen = GNUNET_new_array (pr->replies_seen_size,
 371                                          struct GNUNET_HashCode);
 372     GNUNET_memcpy (pr->replies_seen,
 373                    replies_seen,
 374                    replies_seen_count * sizeof (struct GNUNET_HashCode));
 375     pr->replies_seen_count = replies_seen_count;
 376   }
 377   if ( (NULL != bf_data) &&
 378        (GNUNET_BLOCK_TYPE_FS_UBLOCK == pr->public_data.type) )
 379   {
 380     pr->bg
 381       =  GNUNET_BLOCK_group_create (GSF_block_ctx,
 382                                     pr->public_data.type,
 383                                     mingle,
 384                                     bf_data,
 385                                     bf_size,
 386                                     "fs-seen-set-size",
 387                                     0);
 388   }
 389   else if ((replies_seen_count > 0) &&
 390            (0 != (options & GSF_PRO_BLOOMFILTER_FULL_REFRESH)))
 391   {
 392     refresh_bloomfilter (pr->public_data.type,
 393                          pr);
 394   }
 395   GNUNET_CONTAINER_multihashmap_put (pr_map,
 396                                      &pr->public_data.query,
 397                                      pr,
 398                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 399   if (0 == (options & GSF_PRO_REQUEST_NEVER_EXPIRES))
 400   {
 401     pr->hnode =
 402         GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
 403                                       pr,
 404                                       pr->public_data.ttl.abs_value_us);
 405     /* make sure we don't track too many requests */
 406     while (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) >
 407            max_pending_requests)
 408     {
 409       dpr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
 410       GNUNET_assert (NULL != dpr);
 411       if (pr == dpr)
 412         break;                  /* let the request live briefly... */
 413       if (NULL != dpr->rh)
 414         dpr->rh (dpr->rh_cls,
 415                  GNUNET_BLOCK_EVALUATION_REQUEST_VALID,
 416                  dpr,
 417                  UINT32_MAX,
 418                  GNUNET_TIME_UNIT_FOREVER_ABS,
 419                  GNUNET_TIME_UNIT_FOREVER_ABS,
 420                  GNUNET_BLOCK_TYPE_ANY,
 421                  NULL,
 422                  0);
 423       GSF_pending_request_cancel_ (dpr,
 424                                    GNUNET_YES);
 425     }
 426   }
 427   GNUNET_STATISTICS_update (GSF_stats,
 428                             gettext_noop ("# Pending requests active"), 1,
 429                             GNUNET_NO);
 430   return pr;
 431 }
 432
 433 /**
 434  * Obtain the public data associated with a pending request
 435  *
 436  * @param pr pending request
 437  * @return associated public data
 438  */
 439 struct GSF_PendingRequestData *
 440 GSF_pending_request_get_data_ (struct GSF_PendingRequest *pr)
 441 {
 442   return &pr->public_data;
 443 }
 444
 445
 446 /**
 447  * Test if two pending requests are compatible (would generate
 448  * the same query modulo filters and should thus be processed
 449  * jointly).
 450  *
 451  * @param pra a pending request
 452  * @param prb another pending request
 453  * @return #GNUNET_OK if the requests are compatible
 454  */
 455 int
 456 GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra,
 457                                     struct GSF_PendingRequest *prb)
 458 {
 459   if ( (pra->public_data.type != prb->public_data.type) ||
 460        (0 != memcmp (&pra->public_data.query,
 461                      &prb->public_data.query,
 462                      sizeof (struct GNUNET_HashCode))))
 463     return GNUNET_NO;
 464   return GNUNET_OK;
 465 }
 466
 467
 468 /**
 469  * Update a given pending request with additional replies
 470  * that have been seen.
 471  *
 472  * @param pr request to update
 473  * @param replies_seen hash codes of replies that we've seen
 474  * @param replies_seen_count size of the replies_seen array
 475  */
 476 void
 477 GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
 478                              const struct GNUNET_HashCode * replies_seen,
 479                              unsigned int replies_seen_count)
 480 {
 481   if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
 482     return;                     /* integer overflow */
 483   if (0 != (pr->public_data.options & GSF_PRO_BLOOMFILTER_FULL_REFRESH))
 484   {
 485     /* we're responsible for the BF, full refresh */
 486     if (replies_seen_count + pr->replies_seen_count > pr->replies_seen_size)
 487       GNUNET_array_grow (pr->replies_seen,
 488                          pr->replies_seen_size,
 489                          replies_seen_count + pr->replies_seen_count);
 490     GNUNET_memcpy (&pr->replies_seen[pr->replies_seen_count],
 491                    replies_seen,
 492                    sizeof (struct GNUNET_HashCode) * replies_seen_count);
 493     pr->replies_seen_count += replies_seen_count;
 494     refresh_bloomfilter (pr->public_data.type,
 495                          pr);
 496   }
 497   else
 498   {
 499     if (NULL == pr->bg)
 500     {
 501       /* we're not the initiator, but the initiator did not give us
 502        * any bloom-filter, so we need to create one on-the-fly */
 503       refresh_bloomfilter (pr->public_data.type,
 504                            pr);
 505     }
 506     else
 507     {
 508       GNUNET_break (GNUNET_OK ==
 509                     GNUNET_BLOCK_group_set_seen (pr->bg,
 510                                                  replies_seen,
 511                                                  pr->replies_seen_count));
 512     }
 513   }
 514   if (NULL != pr->gh)
 515     GNUNET_DHT_get_filter_known_results (pr->gh,
 516                                          replies_seen_count,
 517                                          replies_seen);
 518 }
 519
 520
 521 /**
 522  * Generate the message corresponding to the given pending request for
 523  * transmission to other peers.
 524  *
 525  * @param pr request to generate the message for
 526  * @return envelope with the request message
 527  */
 528 struct GNUNET_MQ_Envelope *
 529 GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr)
 530 {
 531   struct GNUNET_MQ_Envelope *env;
 532   struct GetMessage *gm;
 533   struct GNUNET_PeerIdentity *ext;
 534   unsigned int k;
 535   uint32_t bm;
 536   uint32_t prio;
 537   size_t bf_size;
 538   struct GNUNET_TIME_Absolute now;
 539   int64_t ttl;
 540   int do_route;
 541   void *bf_data;
 542   uint32_t bf_nonce;
 543
 544   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 545               "Building request message for `%s' of type %d\n",
 546               GNUNET_h2s (&pr->public_data.query),
 547               pr->public_data.type);
 548   k = 0;
 549   bm = 0;
 550   do_route = (0 == (pr->public_data.options & GSF_PRO_FORWARD_ONLY));
 551   if ((!do_route) && (pr->sender_pid == 0))
 552   {
 553     GNUNET_break (0);
 554     do_route = GNUNET_YES;
 555   }
 556   if (!do_route)
 557   {
 558     bm |= GET_MESSAGE_BIT_RETURN_TO;
 559     k++;
 560   }
 561   if (NULL != pr->public_data.target)
 562   {
 563     bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
 564     k++;
 565   }
 566   if (GNUNET_OK !=
 567       GNUNET_BLOCK_group_serialize (pr->bg,
 568                                     &bf_nonce,
 569                                     &bf_data,
 570                                     &bf_size))
 571   {
 572     bf_size = 0;
 573     bf_data = NULL;
 574   }
 575   env = GNUNET_MQ_msg_extra (gm,
 576                              bf_size + k * sizeof (struct GNUNET_PeerIdentity),
 577                              GNUNET_MESSAGE_TYPE_FS_GET);
 578   gm->type = htonl (pr->public_data.type);
 579   if (do_route)
 580     prio =
 581         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 582                                   pr->public_data.priority + 1);
 583   else
 584     prio = 0;
 585   pr->public_data.priority -= prio;
 586   pr->public_data.num_transmissions++;
 587   pr->public_data.respect_offered += prio;
 588   gm->priority = htonl (prio);
 589   now = GNUNET_TIME_absolute_get ();
 590   ttl = (int64_t) (pr->public_data.ttl.abs_value_us - now.abs_value_us);
 591   gm->ttl = htonl (ttl / 1000LL / 1000LL);
 592   gm->filter_mutator = htonl (bf_nonce);
 593   gm->hash_bitmap = htonl (bm);
 594   gm->query = pr->public_data.query;
 595   ext = (struct GNUNET_PeerIdentity *) &gm[1];
 596   k = 0;
 597   if (! do_route)
 598     GNUNET_PEER_resolve (pr->sender_pid,
 599                          &ext[k++]);
 600   if (NULL != pr->public_data.target)
 601     ext[k++] = *pr->public_data.target;
 602   GNUNET_memcpy (&ext[k],
 603                  bf_data,
 604                  bf_size);
 605   GNUNET_free_non_null (bf_data);
 606   return env;
 607 }
 608
 609
 610 /**
 611  * Iterator to free pending requests.
 612  *
 613  * @param cls closure, unused
 614  * @param key current key code
 615  * @param value value in the hash map (pending request)
 616  * @return #GNUNET_YES (we should continue to iterate)
 617  */
 618 static int
 619 clean_request (void *cls,
 620                const struct GNUNET_HashCode *key,
 621                void *value)
 622 {
 623   struct GSF_PendingRequest *pr = value;
 624   GSF_LocalLookupContinuation cont;
 625
 626   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 627               "Cleaning up pending request for `%s'.\n",
 628               GNUNET_h2s (key));
 629   if (NULL != pr->cadet_request)
 630   {
 631     pr->cadet_retry_count = CADET_RETRY_MAX;
 632     GSF_cadet_query_cancel (pr->cadet_request);
 633     pr->cadet_request = NULL;
 634   }
 635   if (NULL != (cont = pr->llc_cont))
 636   {
 637     pr->llc_cont = NULL;
 638     cont (pr->llc_cont_cls,
 639           pr,
 640           pr->local_result);
 641   }
 642   GSF_plan_notify_request_done_ (pr);
 643   GNUNET_free_non_null (pr->replies_seen);
 644   GNUNET_BLOCK_group_destroy (pr->bg);
 645   pr->bg = NULL;
 646   GNUNET_PEER_change_rc (pr->sender_pid, -1);
 647   pr->sender_pid = 0;
 648   GNUNET_PEER_change_rc (pr->origin_pid, -1);
 649   pr->origin_pid = 0;
 650   if (NULL != pr->hnode)
 651   {
 652     GNUNET_CONTAINER_heap_remove_node (pr->hnode);
 653     pr->hnode = NULL;
 654   }
 655   if (NULL != pr->qe)
 656   {
 657     GNUNET_DATASTORE_cancel (pr->qe);
 658     pr->qe = NULL;
 659   }
 660   if (NULL != pr->gh)
 661   {
 662     GNUNET_DHT_get_stop (pr->gh);
 663     pr->gh = NULL;
 664   }
 665   if (NULL != pr->warn_task)
 666   {
 667     GNUNET_SCHEDULER_cancel (pr->warn_task);
 668     pr->warn_task = NULL;
 669   }
 670   GNUNET_assert (GNUNET_OK ==
 671                  GNUNET_CONTAINER_multihashmap_remove (pr_map,
 672                                                        &pr->public_data.query,
 673                                                        pr));
 674   GNUNET_STATISTICS_update (GSF_stats,
 675                             gettext_noop ("# Pending requests active"),
 676                             -1,
 677                             GNUNET_NO);
 678   GNUNET_free (pr);
 679   return GNUNET_YES;
 680 }
 681
 682
 683 /**
 684  * Explicitly cancel a pending request.
 685  *
 686  * @param pr request to cancel
 687  * @param full_cleanup fully purge the request
 688  */
 689 void
 690 GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr,
 691                              int full_cleanup)
 692 {
 693   GSF_LocalLookupContinuation cont;
 694
 695   if (NULL == pr_map)
 696     return;                     /* already cleaned up! */
 697   if (GNUNET_NO == full_cleanup)
 698   {
 699     /* make request inactive (we're no longer interested in more results),
 700      * but do NOT remove from our data-structures, we still need it there
 701      * to prevent the request from looping */
 702     pr->rh = NULL;
 703     if (NULL != pr->cadet_request)
 704     {
 705       pr->cadet_retry_count = CADET_RETRY_MAX;
 706       GSF_cadet_query_cancel (pr->cadet_request);
 707       pr->cadet_request = NULL;
 708     }
 709     if (NULL != (cont = pr->llc_cont))
 710     {
 711       pr->llc_cont = NULL;
 712       cont (pr->llc_cont_cls,
 713             pr,
 714             pr->local_result);
 715     }
 716     GSF_plan_notify_request_done_ (pr);
 717     if (NULL != pr->qe)
 718     {
 719       GNUNET_DATASTORE_cancel (pr->qe);
 720       pr->qe = NULL;
 721     }
 722     if (NULL != pr->gh)
 723     {
 724       GNUNET_DHT_get_stop (pr->gh);
 725       pr->gh = NULL;
 726     }
 727     if (NULL != pr->warn_task)
 728     {
 729       GNUNET_SCHEDULER_cancel (pr->warn_task);
 730       pr->warn_task = NULL;
 731     }
 732     return;
 733   }
 734   GNUNET_assert (GNUNET_YES ==
 735                  clean_request (NULL,
 736                                 &pr->public_data.query,
 737                                 pr));
 738 }
 739
 740
 741 /**
 742  * Iterate over all pending requests.
 743  *
 744  * @param it function to call for each request
 745  * @param cls closure for @a it
 746  */
 747 void
 748 GSF_iterate_pending_requests_ (GSF_PendingRequestIterator it, void *cls)
 749 {
 750   GNUNET_CONTAINER_multihashmap_iterate (pr_map,
 751                                          (GNUNET_CONTAINER_HashMapIterator) it,
 752                                          cls);
 753 }
 754
 755
 756 /**
 757  * Closure for process_reply() function.
 758  */
 759 struct ProcessReplyClosure
 760 {
 761   /**
 762    * The data for the reply.
 763    */
 764   const void *data;
 765
 766   /**
 767    * Who gave us this reply? NULL for local host (or DHT)
 768    */
 769   struct GSF_ConnectedPeer *sender;
 770
 771   /**
 772    * When the reply expires.
 773    */
 774   struct GNUNET_TIME_Absolute expiration;
 775
 776   /**
 777    * Size of data.
 778    */
 779   size_t size;
 780
 781   /**
 782    * Type of the block.
 783    */
 784   enum GNUNET_BLOCK_Type type;
 785
 786   /**
 787    * Control flags for evaluation.
 788    */
 789   enum GNUNET_BLOCK_EvaluationOptions eo;
 790
 791   /**
 792    * How much was this reply worth to us?
 793    */
 794   uint32_t priority;
 795
 796   /**
 797    * Anonymity requirements for this reply.
 798    */
 799   uint32_t anonymity_level;
 800
 801   /**
 802    * Evaluation result (returned).
 803    */
 804   enum GNUNET_BLOCK_EvaluationResult eval;
 805
 806   /**
 807    * Did we find a matching request?
 808    */
 809   int request_found;
 810 };
 811
 812
 813 /**
 814  * Update the performance data for the sender (if any) since
 815  * the sender successfully answered one of our queries.
 816  *
 817  * @param prq information about the sender
 818  * @param pr request that was satisfied
 819  */
 820 static void
 821 update_request_performance_data (struct ProcessReplyClosure *prq,
 822                                  struct GSF_PendingRequest *pr)
 823 {
 824   if (prq->sender == NULL)
 825     return;
 826   GSF_peer_update_performance_ (prq->sender, pr->public_data.start_time,
 827                                 prq->priority);
 828 }
 829
 830
 831 /**
 832  * We have received a reply; handle it!
 833  *
 834  * @param cls response (a `struct ProcessReplyClosure`)
 835  * @param key our query
 836  * @param value value in the hash map (info about the query)
 837  * @return #GNUNET_YES (we should continue to iterate)
 838  */
 839 static int
 840 process_reply (void *cls,
 841                const struct GNUNET_HashCode *key,
 842                void *value)
 843 {
 844   struct ProcessReplyClosure *prq = cls;
 845   struct GSF_PendingRequest *pr = value;
 846   struct GNUNET_HashCode chash;
 847   struct GNUNET_TIME_Absolute last_transmission;
 848
 849   if (NULL == pr->rh)
 850     return GNUNET_YES;
 851   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 852               "Matched result (type %u) for query `%s' with pending request\n",
 853               (unsigned int) prq->type,
 854               GNUNET_h2s (key));
 855   GNUNET_STATISTICS_update (GSF_stats,
 856                             gettext_noop ("# replies received and matched"), 1,
 857                             GNUNET_NO);
 858   prq->eval =
 859       GNUNET_BLOCK_evaluate (GSF_block_ctx,
 860                              prq->type,
 861                              pr->bg,
 862                              prq->eo,
 863                              key,
 864                              NULL,
 865                              0,
 866                              prq->data,
 867                              prq->size);
 868   switch (prq->eval)
 869   {
 870   case GNUNET_BLOCK_EVALUATION_OK_MORE:
 871     update_request_performance_data (prq, pr);
 872     break;
 873   case GNUNET_BLOCK_EVALUATION_OK_LAST:
 874     /* short cut: stop processing early, no BF-update, etc. */
 875     update_request_performance_data (prq, pr);
 876     GNUNET_LOAD_update (GSF_rt_entry_lifetime,
 877                         GNUNET_TIME_absolute_get_duration (pr->
 878                                                            public_data.start_time).rel_value_us);
 879     if (GNUNET_YES !=
 880         GSF_request_plan_reference_get_last_transmission_ (pr->public_data.pr_head,
 881                                                            prq->sender,
 882                                                            &last_transmission))
 883       last_transmission.abs_value_us = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us;
 884     /* pass on to other peers / local clients */
 885     pr->rh (pr->rh_cls, prq->eval, pr, prq->anonymity_level, prq->expiration,
 886             last_transmission, prq->type, prq->data, prq->size);
 887     return GNUNET_YES;
 888   case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
 889 #if INSANE_STATISTICS
 890     GNUNET_STATISTICS_update (GSF_stats,
 891                               gettext_noop
 892                               ("# duplicate replies discarded (bloomfilter)"),
 893                               1, GNUNET_NO);
 894 #endif
 895     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 896                 "Duplicate response, discarding.\n");
 897     return GNUNET_YES;          /* duplicate */
 898   case GNUNET_BLOCK_EVALUATION_RESULT_IRRELEVANT:
 899     GNUNET_STATISTICS_update (GSF_stats,
 900                               gettext_noop
 901                               ("# irrelevant replies discarded"),
 902                               1, GNUNET_NO);
 903     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 904                 "Irrelevant response, ignoring.\n");
 905     return GNUNET_YES;
 906   case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
 907     return GNUNET_YES;          /* wrong namespace */
 908   case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
 909     GNUNET_break (0);
 910     return GNUNET_YES;
 911   case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
 912     GNUNET_break (0);
 913     return GNUNET_YES;
 914   case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
 915     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 916                 _("Unsupported block type %u\n"),
 917                 prq->type);
 918     return GNUNET_NO;
 919   }
 920   /* update bloomfilter */
 921   GNUNET_CRYPTO_hash (prq->data,
 922                       prq->size,
 923                       &chash);
 924   GSF_pending_request_update_ (pr,
 925                                &chash,
 926                                1);
 927   if (NULL == prq->sender)
 928   {
 929     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 930                 "Found result for query `%s' in local datastore\n",
 931                 GNUNET_h2s (key));
 932     GNUNET_STATISTICS_update (GSF_stats,
 933                               gettext_noop ("# results found locally"),
 934                               1,
 935                               GNUNET_NO);
 936   }
 937   else
 938   {
 939     GSF_dht_lookup_ (pr);
 940   }
 941   prq->priority += pr->public_data.original_priority;
 942   pr->public_data.priority = 0;
 943   pr->public_data.original_priority = 0;
 944   pr->public_data.results_found++;
 945   prq->request_found = GNUNET_YES;
 946   /* finally, pass on to other peer / local client */
 947   if (! GSF_request_plan_reference_get_last_transmission_ (pr->public_data.pr_head,
 948                                                            prq->sender,
 949                                                            &last_transmission))
 950     last_transmission.abs_value_us = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us;
 951   pr->rh (pr->rh_cls,
 952           prq->eval,
 953           pr,
 954           prq->anonymity_level,
 955           prq->expiration,
 956           last_transmission,
 957           prq->type,
 958           prq->data,
 959           prq->size);
 960   return GNUNET_YES;
 961 }
 962
 963
 964 /**
 965  * Context for put_migration_continuation().
 966  */
 967 struct PutMigrationContext
 968 {
 969
 970   /**
 971    * Start time for the operation.
 972    */
 973   struct GNUNET_TIME_Absolute start;
 974
 975   /**
 976    * Request origin.
 977    */
 978   struct GNUNET_PeerIdentity origin;
 979
 980   /**
 981    * #GNUNET_YES if we had a matching request for this block,
 982    * #GNUNET_NO if not.
 983    */
 984   int requested;
 985 };
 986
 987
 988 /**
 989  * Continuation called to notify client about result of the
 990  * operation.
 991  *
 992  * @param cls closure
 993  * @param success #GNUNET_SYSERR on failure
 994  * @param min_expiration minimum expiration time required for content to be stored
 995  * @param msg NULL on success, otherwise an error message
 996  */
 997 static void
 998 put_migration_continuation (void *cls, int success,
 999                             struct GNUNET_TIME_Absolute min_expiration,
1000                             const char *msg)
1001 {
1002   struct PutMigrationContext *pmc = cls;
1003   struct GSF_ConnectedPeer *cp;
1004   struct GNUNET_TIME_Relative mig_pause;
1005   struct GSF_PeerPerformanceData *ppd;
1006
1007   if (NULL != datastore_put_load)
1008   {
1009     if (GNUNET_SYSERR != success)
1010     {
1011       GNUNET_LOAD_update (datastore_put_load,
1012                           GNUNET_TIME_absolute_get_duration (pmc->start).rel_value_us);
1013     }
1014     else
1015     {
1016       /* on queue failure / timeout, increase the put load dramatically */
1017       GNUNET_LOAD_update (datastore_put_load,
1018                           GNUNET_TIME_UNIT_MINUTES.rel_value_us);
1019     }
1020   }
1021   cp = GSF_peer_get_ (&pmc->origin);
1022   if (GNUNET_OK == success)
1023   {
1024     if (NULL != cp)
1025     {
1026       ppd = GSF_get_peer_performance_data_ (cp);
1027       ppd->migration_delay.rel_value_us /= 2;
1028     }
1029     GNUNET_free (pmc);
1030     return;
1031   }
1032   if ( (GNUNET_NO == success) &&
1033        (GNUNET_NO == pmc->requested) &&
1034        (NULL != cp) )
1035   {
1036     ppd = GSF_get_peer_performance_data_ (cp);
1037     if (min_expiration.abs_value_us > 0)
1038     {
1039       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1040                   "Asking to stop migration for %s because datastore is full\n",
1041                   GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_expiration), GNUNET_YES));
1042       GSF_block_peer_migration_ (cp, min_expiration);
1043     }
1044     else
1045     {
1046       ppd->migration_delay = GNUNET_TIME_relative_max (GNUNET_TIME_UNIT_SECONDS,
1047                                                        ppd->migration_delay);
1048       ppd->migration_delay = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS,
1049                                                        ppd->migration_delay);
1050       mig_pause.rel_value_us = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1051                                                          ppd->migration_delay.rel_value_us);
1052       ppd->migration_delay = GNUNET_TIME_relative_saturating_multiply (ppd->migration_delay, 2);
1053       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1054                   "Replicated content already exists locally, asking to stop migration for %s\n",
1055                   GNUNET_STRINGS_relative_time_to_string (mig_pause,
1056                                                           GNUNET_YES));
1057       GSF_block_peer_migration_ (cp,
1058                                  GNUNET_TIME_relative_to_absolute (mig_pause));
1059     }
1060   }
1061   GNUNET_free (pmc);
1062   GNUNET_STATISTICS_update (GSF_stats,
1063                             gettext_noop ("# Datastore `PUT' failures"), 1,
1064                             GNUNET_NO);
1065 }
1066
1067
1068 /**
1069  * Test if the DATABASE (PUT) load on this peer is too high
1070  * to even consider processing the query at
1071  * all.
1072  *
1073  * @param priority the priority of the item
1074  * @return #GNUNET_YES if the load is too high to do anything (load high)
1075  *         #GNUNET_NO to process normally (load normal or low)
1076  */
1077 static int
1078 test_put_load_too_high (uint32_t priority)
1079 {
1080   double ld;
1081
1082   if (NULL == datastore_put_load)
1083     return GNUNET_NO;
1084   if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
1085     return GNUNET_NO;           /* very fast */
1086   ld = GNUNET_LOAD_get_load (datastore_put_load);
1087   if (ld < 2.0 * (1 + priority))
1088     return GNUNET_NO;
1089   GNUNET_STATISTICS_update (GSF_stats,
1090                             gettext_noop
1091                             ("# storage requests dropped due to high load"), 1,
1092                             GNUNET_NO);
1093   return GNUNET_YES;
1094 }
1095
1096
1097 /**
1098  * Iterator called on each result obtained for a DHT
1099  * operation that expects a reply
1100  *
1101  * @param cls closure
1102  * @param exp when will this value expire
1103  * @param key key of the result
1104  * @param get_path peers on reply path (or NULL if not recorded)
1105  * @param get_path_length number of entries in @a get_path
1106  * @param put_path peers on the PUT path (or NULL if not recorded)
1107  * @param put_path_length number of entries in @a get_path
1108  * @param type type of the result
1109  * @param size number of bytes in @a data
1110  * @param data pointer to the result data
1111  */
1112 static void
1113 handle_dht_reply (void *cls,
1114                   struct GNUNET_TIME_Absolute exp,
1115                   const struct GNUNET_HashCode *key,
1116                   const struct GNUNET_PeerIdentity *get_path,
1117                   unsigned int get_path_length,
1118                   const struct GNUNET_PeerIdentity *put_path,
1119                   unsigned int put_path_length,
1120                   enum GNUNET_BLOCK_Type type,
1121                   size_t size,
1122                   const void *data)
1123 {
1124   struct GSF_PendingRequest *pr = cls;
1125   struct ProcessReplyClosure prq;
1126   struct PutMigrationContext *pmc;
1127
1128   GNUNET_STATISTICS_update (GSF_stats,
1129                             gettext_noop ("# Replies received from DHT"), 1,
1130                             GNUNET_NO);
1131   memset (&prq, 0, sizeof (prq));
1132   prq.data = data;
1133   prq.expiration = exp;
1134   /* do not allow migrated content to live longer than 1 year */
1135   prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_YEARS),
1136                                              prq.expiration);
1137   prq.size = size;
1138   prq.type = type;
1139   prq.eo = GNUNET_BLOCK_EO_NONE;
1140   process_reply (&prq, key, pr);
1141   if ((GNUNET_YES == active_to_migration) &&
1142       (GNUNET_NO == test_put_load_too_high (prq.priority)))
1143   {
1144     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1145                 "Replicating result for query `%s' with priority %u\n",
1146                 GNUNET_h2s (key), prq.priority);
1147     pmc = GNUNET_new (struct PutMigrationContext);
1148     pmc->start = GNUNET_TIME_absolute_get ();
1149     pmc->requested = GNUNET_YES;
1150     if (NULL ==
1151         GNUNET_DATASTORE_put (GSF_dsh, 0, key, size, data, type, prq.priority,
1152                               1 /* anonymity */ ,
1153                               0 /* replication */ ,
1154                               exp, 1 + prq.priority, MAX_DATASTORE_QUEUE,
1155                               &put_migration_continuation, pmc))
1156     {
1157       put_migration_continuation (pmc,
1158                                   GNUNET_SYSERR,
1159                                   GNUNET_TIME_UNIT_ZERO_ABS,
1160                                   NULL);
1161     }
1162   }
1163 }
1164
1165
1166 /**
1167  * Consider looking up the data in the DHT (anonymity-level permitting).
1168  *
1169  * @param pr the pending request to process
1170  */
1171 void
1172 GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
1173 {
1174   const void *xquery;
1175   size_t xquery_size;
1176   struct GNUNET_PeerIdentity pi;
1177   char buf[sizeof (struct GNUNET_HashCode) * 2] GNUNET_ALIGN;
1178
1179   if (0 != pr->public_data.anonymity_level)
1180     return;
1181   if (NULL != pr->gh)
1182   {
1183     GNUNET_DHT_get_stop (pr->gh);
1184     pr->gh = NULL;
1185   }
1186   xquery = NULL;
1187   xquery_size = 0;
1188   if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
1189   {
1190     GNUNET_assert (0 != pr->sender_pid);
1191     GNUNET_PEER_resolve (pr->sender_pid, &pi);
1192     GNUNET_memcpy (&buf[xquery_size], &pi, sizeof (struct GNUNET_PeerIdentity));
1193     xquery_size += sizeof (struct GNUNET_PeerIdentity);
1194   }
1195   pr->gh =
1196       GNUNET_DHT_get_start (GSF_dht,
1197                             pr->public_data.type, &pr->public_data.query,
1198                             DHT_GET_REPLICATION,
1199                             GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
1200                             xquery, xquery_size, &handle_dht_reply, pr);
1201   if ( (NULL != pr->gh) &&
1202        (0 != pr->replies_seen_count) )
1203     GNUNET_DHT_get_filter_known_results (pr->gh,
1204                                          pr->replies_seen_count,
1205                                          pr->replies_seen);
1206 }
1207
1208
1209 /**
1210  * Function called with a reply from the cadet.
1211  *
1212  * @param cls the pending request struct
1213  * @param type type of the block, ANY on error
1214  * @param expiration expiration time for the block
1215  * @param data_size number of bytes in @a data, 0 on error
1216  * @param data reply block data, NULL on error
1217  */
1218 static void
1219 cadet_reply_proc (void *cls,
1220                  enum GNUNET_BLOCK_Type type,
1221                  struct GNUNET_TIME_Absolute expiration,
1222                  size_t data_size,
1223                  const void *data)
1224 {
1225   struct GSF_PendingRequest *pr = cls;
1226   struct ProcessReplyClosure prq;
1227   struct GNUNET_HashCode query;
1228
1229   pr->cadet_request = NULL;
1230   if (GNUNET_BLOCK_TYPE_ANY == type)
1231   {
1232     GNUNET_break (NULL == data);
1233     GNUNET_break (0 == data_size);
1234     pr->cadet_retry_count++;
1235     if (pr->cadet_retry_count >= CADET_RETRY_MAX)
1236       return; /* give up on cadet */
1237     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1238                 "Error retrieiving block via cadet\n");
1239     /* retry -- without delay, as this is non-anonymous
1240        and cadet/cadet connect will take some time anyway */
1241     pr->cadet_request = GSF_cadet_query (pr->public_data.target,
1242                                          &pr->public_data.query,
1243                                          pr->public_data.type,
1244                                          &cadet_reply_proc,
1245                                          pr);
1246     return;
1247   }
1248   if (GNUNET_YES !=
1249       GNUNET_BLOCK_get_key (GSF_block_ctx,
1250                             type,
1251                             data, data_size, &query))
1252   {
1253     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1254                 "Failed to derive key for block of type %d\n",
1255                 (int) type);
1256     GNUNET_break_op (0);
1257     return;
1258   }
1259   GNUNET_STATISTICS_update (GSF_stats,
1260                             gettext_noop ("# Replies received from CADET"), 1,
1261                             GNUNET_NO);
1262   memset (&prq, 0, sizeof (prq));
1263   prq.data = data;
1264   prq.expiration = expiration;
1265   /* do not allow migrated content to live longer than 1 year */
1266   prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_YEARS),
1267                                              prq.expiration);
1268   prq.size = data_size;
1269   prq.type = type;
1270   prq.eo = GNUNET_BLOCK_EO_NONE;
1271   process_reply (&prq, &query, pr);
1272 }
1273
1274
1275 /**
1276  * Consider downloading via cadet (if possible)
1277  *
1278  * @param pr the pending request to process
1279  */
1280 void
1281 GSF_cadet_lookup_ (struct GSF_PendingRequest *pr)
1282 {
1283   if (0 != pr->public_data.anonymity_level)
1284     return;
1285   if (0 == pr->public_data.target)
1286   {
1287     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1288                 "Cannot do cadet-based download, target peer not known\n");
1289     return;
1290   }
1291   if (NULL != pr->cadet_request)
1292     return;
1293   pr->cadet_request = GSF_cadet_query (pr->public_data.target,
1294                                        &pr->public_data.query,
1295                                        pr->public_data.type,
1296                                        &cadet_reply_proc,
1297                                        pr);
1298 }
1299
1300
1301 /**
1302  * Task that issues a warning if the datastore lookup takes too long.
1303  *
1304  * @param cls the `struct GSF_PendingRequest`
1305  */
1306 static void
1307 warn_delay_task (void *cls)
1308 {
1309   struct GSF_PendingRequest *pr = cls;
1310
1311   GNUNET_log (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
1312               _("Datastore lookup already took %s!\n"),
1313               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_duration (pr->qe_start),
1314                                                       GNUNET_YES));
1315   pr->warn_task =
1316       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1317                                     &warn_delay_task,
1318                                     pr);
1319 }
1320
1321
1322 /**
1323  * Task that issues a warning if the datastore lookup takes too long.
1324  *
1325  * @param cls the `struct GSF_PendingRequest`
1326  */
1327 static void
1328 odc_warn_delay_task (void *cls)
1329 {
1330   struct GSF_PendingRequest *pr = cls;
1331
1332   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1333               _("On-demand lookup already took %s!\n"),
1334               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_duration (pr->qe_start), GNUNET_YES));
1335   pr->warn_task =
1336       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1337                                     &odc_warn_delay_task, pr);
1338 }
1339
1340
1341 /**
1342  * We're processing (local) results for a search request
1343  * from another peer.  Pass applicable results to the
1344  * peer and if we are done either clean up (operation
1345  * complete) or forward to other peers (more results possible).
1346  *
1347  * @param cls our closure (`struct GSF_PendingRequest *`)
1348  * @param key key for the content
1349  * @param size number of bytes in @a data
1350  * @param data content stored
1351  * @param type type of the content
1352  * @param priority priority of the content
1353  * @param anonymity anonymity-level for the content
1354  * @param expiration expiration time for the content
1355  * @param uid unique identifier for the datum;
1356  *        maybe 0 if no unique identifier is available
1357  */
1358 static void
1359 process_local_reply (void *cls,
1360                      const struct GNUNET_HashCode *key,
1361                      size_t size,
1362                      const void *data,
1363                      enum GNUNET_BLOCK_Type type,
1364                      uint32_t priority,
1365                      uint32_t anonymity,
1366                      struct GNUNET_TIME_Absolute expiration,
1367                      uint64_t uid)
1368 {
1369   struct GSF_PendingRequest *pr = cls;
1370   GSF_LocalLookupContinuation cont;
1371   struct ProcessReplyClosure prq;
1372   struct GNUNET_HashCode query;
1373   unsigned int old_rf;
1374
1375   GNUNET_SCHEDULER_cancel (pr->warn_task);
1376   pr->warn_task = NULL;
1377   if (NULL != pr->qe)
1378   {
1379     pr->qe = NULL;
1380     if (NULL == key)
1381     {
1382 #if INSANE_STATISTICS
1383       GNUNET_STATISTICS_update (GSF_stats,
1384                                 gettext_noop
1385                                 ("# Datastore lookups concluded (no results)"),
1386                                 1, GNUNET_NO);
1387 #endif
1388     }
1389     if (GNUNET_NO == pr->have_first_uid)
1390     {
1391       pr->first_uid = uid;
1392       pr->have_first_uid = 1;
1393     }
1394     else
1395     {
1396       if ((uid == pr->first_uid) && (key != NULL))
1397       {
1398         GNUNET_STATISTICS_update (GSF_stats,
1399                                   gettext_noop
1400                                   ("# Datastore lookups concluded (seen all)"),
1401                                   1, GNUNET_NO);
1402         key = NULL;             /* all replies seen! */
1403       }
1404       pr->have_first_uid++;
1405       if ((pr->have_first_uid > MAX_RESULTS) && (key != NULL))
1406       {
1407         GNUNET_STATISTICS_update (GSF_stats,
1408                                   gettext_noop
1409                                   ("# Datastore lookups aborted (more than MAX_RESULTS)"),
1410                                   1, GNUNET_NO);
1411         key = NULL;             /* all replies seen! */
1412       }
1413     }
1414   }
1415   if (NULL == key)
1416   {
1417     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1418                 "No further local responses available.\n");
1419 #if INSANE_STATISTICS
1420     if ((pr->public_data.type == GNUNET_BLOCK_TYPE_FS_DBLOCK) ||
1421         (pr->public_data.type == GNUNET_BLOCK_TYPE_FS_IBLOCK))
1422       GNUNET_STATISTICS_update (GSF_stats,
1423                                 gettext_noop
1424                                 ("# requested DBLOCK or IBLOCK not found"), 1,
1425                                 GNUNET_NO);
1426 #endif
1427     goto check_error_and_continue;
1428   }
1429   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1430               "Received reply for `%s' of type %d with UID %llu from datastore.\n",
1431               GNUNET_h2s (key), type, (unsigned long long) uid);
1432   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1433   {
1434     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1435                 "Found ONDEMAND block, performing on-demand encoding\n");
1436     GNUNET_STATISTICS_update (GSF_stats,
1437                               gettext_noop
1438                               ("# on-demand blocks matched requests"), 1,
1439                               GNUNET_NO);
1440     pr->qe_start = GNUNET_TIME_absolute_get ();
1441     pr->warn_task =
1442         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1443                                       &odc_warn_delay_task, pr);
1444     if (GNUNET_OK ==
1445         GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
1446                                           anonymity, expiration, uid,
1447                                           &process_local_reply, pr))
1448     {
1449       GNUNET_STATISTICS_update (GSF_stats,
1450                                 gettext_noop
1451                                 ("# on-demand lookups performed successfully"),
1452                                 1, GNUNET_NO);
1453       return;                   /* we're done */
1454     }
1455     GNUNET_STATISTICS_update (GSF_stats,
1456                               gettext_noop ("# on-demand lookups failed"), 1,
1457                               GNUNET_NO);
1458     GNUNET_SCHEDULER_cancel (pr->warn_task);
1459     pr->warn_task =
1460         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1461                                       &warn_delay_task, pr);
1462     pr->qe =
1463         GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset - 1,
1464                                   &pr->public_data.query,
1465                                   pr->public_data.type ==
1466                                   GNUNET_BLOCK_TYPE_FS_DBLOCK ?
1467                                   GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
1468                                   (0 !=
1469                                    (GSF_PRO_PRIORITY_UNLIMITED &
1470                                     pr->public_data.options)) ? UINT_MAX : 1
1471                                   /* queue priority */ ,
1472                                   (0 !=
1473                                    (GSF_PRO_PRIORITY_UNLIMITED &
1474                                     pr->public_data.options)) ? UINT_MAX :
1475                                   GSF_datastore_queue_size
1476                                   /* max queue size */ ,
1477                                   &process_local_reply, pr);
1478     if (NULL != pr->qe)
1479       return;                   /* we're done */
1480     GNUNET_STATISTICS_update (GSF_stats,
1481                               gettext_noop
1482                               ("# Datastore lookups concluded (error queueing)"),
1483                               1, GNUNET_NO);
1484     goto check_error_and_continue;
1485   }
1486   old_rf = pr->public_data.results_found;
1487   memset (&prq, 0, sizeof (prq));
1488   prq.data = data;
1489   prq.expiration = expiration;
1490   prq.size = size;
1491   if (GNUNET_OK !=
1492       GNUNET_BLOCK_get_key (GSF_block_ctx, type, data, size, &query))
1493   {
1494     GNUNET_break (0);
1495     GNUNET_DATASTORE_remove (GSF_dsh, key, size, data, -1, -1,
1496                              NULL, NULL);
1497     pr->qe_start = GNUNET_TIME_absolute_get ();
1498     pr->warn_task =
1499         GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1500                                       &warn_delay_task, pr);
1501     pr->qe =
1502         GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset - 1,
1503                                   &pr->public_data.query,
1504                                   pr->public_data.type ==
1505                                   GNUNET_BLOCK_TYPE_FS_DBLOCK ?
1506                                   GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
1507                                   (0 !=
1508                                    (GSF_PRO_PRIORITY_UNLIMITED &
1509                                     pr->public_data.options)) ? UINT_MAX : 1
1510                                   /* queue priority */ ,
1511                                   (0 !=
1512                                    (GSF_PRO_PRIORITY_UNLIMITED &
1513                                     pr->public_data.options)) ? UINT_MAX :
1514                                   GSF_datastore_queue_size
1515                                   /* max queue size */ ,
1516                                   &process_local_reply, pr);
1517     if (NULL == pr->qe)
1518     {
1519       GNUNET_STATISTICS_update (GSF_stats,
1520                                 gettext_noop
1521                                 ("# Datastore lookups concluded (error queueing)"),
1522                                 1, GNUNET_NO);
1523       goto check_error_and_continue;
1524     }
1525     return;
1526   }
1527   prq.type = type;
1528   prq.priority = priority;
1529   prq.request_found = GNUNET_NO;
1530   prq.anonymity_level = anonymity;
1531   if ((0 == old_rf) && (0 == pr->public_data.results_found))
1532     GSF_update_datastore_delay_ (pr->public_data.start_time);
1533   prq.eo = GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO;
1534   process_reply (&prq, key, pr);
1535   pr->local_result = prq.eval;
1536   if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
1537   {
1538     GNUNET_STATISTICS_update (GSF_stats,
1539                               gettext_noop
1540                               ("# Datastore lookups concluded (found last result)"),
1541                               1,
1542                               GNUNET_NO);
1543     goto check_error_and_continue;
1544   }
1545   if ((0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
1546       ((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
1547        (pr->public_data.results_found > 5 + 2 * pr->public_data.priority)))
1548   {
1549     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1550                 "Load too high, done with request\n");
1551     GNUNET_STATISTICS_update (GSF_stats,
1552                               gettext_noop ("# Datastore lookups concluded (load too high)"),
1553                               1,
1554                               GNUNET_NO);
1555     goto check_error_and_continue;
1556   }
1557   pr->qe_start = GNUNET_TIME_absolute_get ();
1558   pr->warn_task =
1559       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1560                                     &warn_delay_task,
1561                                     pr);
1562   pr->qe =
1563       GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
1564                                 &pr->public_data.query,
1565                                 pr->public_data.type ==
1566                                 GNUNET_BLOCK_TYPE_FS_DBLOCK ?
1567                                 GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
1568                                 (0 !=
1569                                  (GSF_PRO_PRIORITY_UNLIMITED & pr->
1570                                   public_data.options)) ? UINT_MAX : 1
1571                                 /* queue priority */ ,
1572                                 (0 !=
1573                                  (GSF_PRO_PRIORITY_UNLIMITED & pr->
1574                                   public_data.options)) ? UINT_MAX :
1575                                 GSF_datastore_queue_size
1576                                 /* max queue size */ ,
1577                                 &process_local_reply, pr);
1578   /* check if we successfully queued another datastore request;
1579    * if so, return, otherwise call our continuation (if we have
1580    * any) */
1581 check_error_and_continue:
1582   if (NULL != pr->qe)
1583     return;
1584   if (NULL != pr->warn_task)
1585   {
1586     GNUNET_SCHEDULER_cancel (pr->warn_task);
1587     pr->warn_task = NULL;
1588   }
1589   if (NULL == (cont = pr->llc_cont))
1590     return;                     /* no continuation */
1591   pr->llc_cont = NULL;
1592   if (0 != (GSF_PRO_LOCAL_ONLY & pr->public_data.options))
1593   {
1594     if (GNUNET_BLOCK_EVALUATION_OK_LAST != pr->local_result)
1595     {
1596       /* Signal that we are done and that there won't be any
1597          additional results to allow client to clean up state. */
1598       pr->rh (pr->rh_cls,
1599                GNUNET_BLOCK_EVALUATION_OK_LAST,
1600                pr,
1601                UINT32_MAX,
1602                GNUNET_TIME_UNIT_ZERO_ABS,
1603                GNUNET_TIME_UNIT_FOREVER_ABS,
1604                GNUNET_BLOCK_TYPE_ANY,
1605                NULL, 0);
1606     }
1607     /* Finally, call our continuation to signal that we are
1608        done with local processing of this request; i.e. to
1609        start reading again from the client. */
1610     cont (pr->llc_cont_cls, NULL, GNUNET_BLOCK_EVALUATION_OK_LAST);
1611     return;
1612   }
1613
1614   cont (pr->llc_cont_cls, pr, pr->local_result);
1615 }
1616
1617
1618 /**
1619  * Is the given target a legitimate peer for forwarding the given request?
1620  *
1621  * @param pr request
1622  * @param target
1623  * @return #GNUNET_YES if this request could be forwarded to the given peer
1624  */
1625 int
1626 GSF_pending_request_test_target_ (struct GSF_PendingRequest *pr,
1627                                   const struct GNUNET_PeerIdentity *target)
1628 {
1629   struct GNUNET_PeerIdentity pi;
1630
1631   if (0 == pr->origin_pid)
1632     return GNUNET_YES;
1633   GNUNET_PEER_resolve (pr->origin_pid, &pi);
1634   return (0 ==
1635           memcmp (&pi, target,
1636                   sizeof (struct GNUNET_PeerIdentity))) ? GNUNET_NO :
1637       GNUNET_YES;
1638 }
1639
1640
1641 /**
1642  * Look up the request in the local datastore.
1643  *
1644  * @param pr the pending request to process
1645  * @param cont function to call at the end
1646  * @param cont_cls closure for @a cont
1647  */
1648 void
1649 GSF_local_lookup_ (struct GSF_PendingRequest *pr,
1650                    GSF_LocalLookupContinuation cont,
1651                    void *cont_cls)
1652 {
1653   GNUNET_assert (NULL == pr->gh);
1654   GNUNET_assert (NULL == pr->cadet_request);
1655   GNUNET_assert (NULL == pr->llc_cont);
1656   pr->llc_cont = cont;
1657   pr->llc_cont_cls = cont_cls;
1658   pr->qe_start = GNUNET_TIME_absolute_get ();
1659   pr->warn_task =
1660       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
1661                                     &warn_delay_task,
1662                                     pr);
1663 #if INSANE_STATISTICS
1664   GNUNET_STATISTICS_update (GSF_stats,
1665                             gettext_noop ("# Datastore lookups initiated"), 1,
1666                             GNUNET_NO);
1667 #endif
1668   pr->qe =
1669       GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
1670                                 &pr->public_data.query,
1671                                 pr->public_data.type ==
1672                                 GNUNET_BLOCK_TYPE_FS_DBLOCK ?
1673                                 GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
1674                                 (0 !=
1675                                  (GSF_PRO_PRIORITY_UNLIMITED & pr->
1676                                   public_data.options)) ? UINT_MAX : 1
1677                                 /* queue priority */ ,
1678                                 (0 !=
1679                                  (GSF_PRO_PRIORITY_UNLIMITED & pr->
1680                                   public_data.options)) ? UINT_MAX :
1681                                 GSF_datastore_queue_size
1682                                 /* max queue size */ ,
1683                                 &process_local_reply, pr);
1684   if (NULL != pr->qe)
1685     return;
1686   GNUNET_STATISTICS_update (GSF_stats,
1687                             gettext_noop
1688                             ("# Datastore lookups concluded (error queueing)"),
1689                             1, GNUNET_NO);
1690   GNUNET_SCHEDULER_cancel (pr->warn_task);
1691   pr->warn_task = NULL;
1692   pr->llc_cont = NULL;
1693   if (NULL != cont)
1694     cont (cont_cls, pr, pr->local_result);
1695 }
1696
1697
1698
1699 /**
1700  * Handle P2P "CONTENT" message.  Checks that the message is
1701  * well-formed and then checks if there are any pending requests for
1702  * this content and possibly passes it on (to local clients or other
1703  * peers).  Does NOT perform migration (content caching at this peer).
1704  *
1705  * @param cls the other peer involved
1706  * @param put the actual message
1707  */
1708 void
1709 handle_p2p_put (void *cls,
1710                 const struct PutMessage *put)
1711 {
1712   struct GSF_ConnectedPeer *cp = cls;
1713   uint16_t msize;
1714   size_t dsize;
1715   enum GNUNET_BLOCK_Type type;
1716   struct GNUNET_TIME_Absolute expiration;
1717   struct GNUNET_HashCode query;
1718   struct ProcessReplyClosure prq;
1719   struct GNUNET_TIME_Relative block_time;
1720   double putl;
1721   struct PutMigrationContext *pmc;
1722
1723   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1724               "Received P2P PUT from %s\n",
1725               GNUNET_i2s (GSF_get_peer_performance_data_ (cp)->peer));
1726   GSF_cover_content_count++;
1727   msize = ntohs (put->header.size);
1728   dsize = msize - sizeof (struct PutMessage);
1729   type = ntohl (put->type);
1730   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
1731   /* do not allow migrated content to live longer than 1 year */
1732   expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_YEARS),
1733                                          expiration);
1734   if (GNUNET_OK !=
1735       GNUNET_BLOCK_get_key (GSF_block_ctx,
1736                             type,
1737                             &put[1],
1738                             dsize,
1739                             &query))
1740   {
1741     GNUNET_break_op (0);
1742     return;
1743   }
1744   GNUNET_STATISTICS_update (GSF_stats,
1745                             gettext_noop ("# GAP PUT messages received"),
1746                             1,
1747                             GNUNET_NO);
1748   /* now, lookup 'query' */
1749   prq.data = (const void *) &put[1];
1750   prq.sender = cp;
1751   prq.size = dsize;
1752   prq.type = type;
1753   prq.expiration = expiration;
1754   prq.priority = 0;
1755   prq.anonymity_level = UINT32_MAX;
1756   prq.request_found = GNUNET_NO;
1757   prq.eo = GNUNET_BLOCK_EO_NONE;
1758   GNUNET_CONTAINER_multihashmap_get_multiple (pr_map,
1759                                               &query,
1760                                               &process_reply,
1761                                               &prq);
1762   if (NULL != cp)
1763   {
1764     GSF_connected_peer_change_preference_ (cp,
1765                                            CONTENT_BANDWIDTH_VALUE +
1766                                            1000 * prq.priority);
1767     GSF_get_peer_performance_data_ (cp)->respect += prq.priority;
1768   }
1769   if ((GNUNET_YES == active_to_migration) &&
1770       (NULL != cp) &&
1771       (GNUNET_NO == test_put_load_too_high (prq.priority)))
1772   {
1773     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1774                 "Replicating result for query `%s' with priority %u\n",
1775                 GNUNET_h2s (&query),
1776                 prq.priority);
1777     pmc = GNUNET_new (struct PutMigrationContext);
1778     pmc->start = GNUNET_TIME_absolute_get ();
1779     pmc->requested = prq.request_found;
1780     GNUNET_assert (0 != GSF_get_peer_performance_data_ (cp)->pid);
1781     GNUNET_PEER_resolve (GSF_get_peer_performance_data_ (cp)->pid,
1782                          &pmc->origin);
1783     if (NULL ==
1784         GNUNET_DATASTORE_put (GSF_dsh,
1785                               0,
1786                               &query,
1787                               dsize,
1788                               &put[1],
1789                               type,
1790                               prq.priority,
1791                               1 /* anonymity */ ,
1792                               0 /* replication */ ,
1793                               expiration, 1 + prq.priority,
1794                               MAX_DATASTORE_QUEUE,
1795                               &put_migration_continuation,
1796                               pmc))
1797     {
1798       put_migration_continuation (pmc,
1799                                   GNUNET_SYSERR,
1800                                   GNUNET_TIME_UNIT_ZERO_ABS,
1801                                   NULL);
1802     }
1803   }
1804   else if (NULL != cp)
1805   {
1806     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1807                 "Choosing not to keep content `%s' (%d/%d)\n",
1808                 GNUNET_h2s (&query),
1809                 active_to_migration,
1810                 test_put_load_too_high (prq.priority));
1811   }
1812   putl = GNUNET_LOAD_get_load (datastore_put_load);
1813   if ( (NULL != cp) &&
1814        (GNUNET_NO == prq.request_found) &&
1815        ( (GNUNET_YES != active_to_migration) ||
1816          (putl > 2.5 * (1 + prq.priority)) ) )
1817   {
1818     if (GNUNET_YES != active_to_migration)
1819       putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
1820     block_time =
1821         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1822                                        5000 +
1823                                        GNUNET_CRYPTO_random_u32
1824                                        (GNUNET_CRYPTO_QUALITY_WEAK,
1825                                         (unsigned int) (60000 * putl * putl)));
1826     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1827                 "Asking to stop migration for %s because of load %f and events %d/%d\n",
1828                 GNUNET_STRINGS_relative_time_to_string (block_time,
1829                                                         GNUNET_YES),
1830                 putl,
1831                 active_to_migration,
1832                 (GNUNET_NO == prq.request_found));
1833     GSF_block_peer_migration_ (cp,
1834                                GNUNET_TIME_relative_to_absolute (block_time));
1835   }
1836 }
1837
1838
1839 /**
1840  * Check if the given request is still active.
1841  *
1842  * @param pr pending request
1843  * @return #GNUNET_YES if the request is still active
1844  */
1845 int
1846 GSF_pending_request_test_active_ (struct GSF_PendingRequest *pr)
1847 {
1848   return (NULL != pr->rh) ? GNUNET_YES : GNUNET_NO;
1849 }
1850
1851
1852 /**
1853  * Setup the subsystem.
1854  */
1855 void
1856 GSF_pending_request_init_ ()
1857 {
1858   if (GNUNET_OK !=
1859       GNUNET_CONFIGURATION_get_value_number (GSF_cfg, "fs",
1860                                              "MAX_PENDING_REQUESTS",
1861                                              &max_pending_requests))
1862   {
1863     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
1864                                "fs", "MAX_PENDING_REQUESTS");
1865   }
1866   active_to_migration =
1867       GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg, "FS", "CONTENT_CACHING");
1868   datastore_put_load = GNUNET_LOAD_value_init (DATASTORE_LOAD_AUTODECLINE);
1869   pr_map = GNUNET_CONTAINER_multihashmap_create (32 * 1024, GNUNET_YES);
1870   requests_by_expiration_heap =
1871       GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1872 }
1873
1874
1875 /**
1876  * Shutdown the subsystem.
1877  */
1878 void
1879 GSF_pending_request_done_ ()
1880 {
1881   GNUNET_CONTAINER_multihashmap_iterate (pr_map,
1882                                          &clean_request,
1883                                          NULL);
1884   GNUNET_CONTAINER_multihashmap_destroy (pr_map);
1885   pr_map = NULL;
1886   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
1887   requests_by_expiration_heap = NULL;
1888   GNUNET_LOAD_value_free (datastore_put_load);
1889   datastore_put_load = NULL;
1890 }
1891
1892
1893 /* end of gnunet-service-fs_pr.c */