src/fs/gnunet-service-fs_pr.c

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