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