src/fs/gnunet-service-fs_pr.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file 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_cp.h"
  29 #include "gnunet-service-fs_pr.h"
  30
  31
  32 /**
  33  * An active request.
  34  */
  35 struct GSF_PendingRequest
  36 {
  37   /**
  38    * Public data for the request.
  39    */
  40   struct GSF_PendingRequestData public_data;
  41
  42   /**
  43    * Function to call if we encounter a reply.
  44    */
  45   GSF_PendingRequestReplyHandler rh;
  46
  47   /**
  48    * Closure for 'rh'
  49    */
  50   void *rh_cls;
  51
  52   /**
  53    * Array of hash codes of replies we've already seen.
  54    */
  55   GNUNET_HashCode *replies_seen;
  56
  57   /**
  58    * Bloomfilter masking replies we've already seen.
  59    */
  60   struct GNUNET_CONTAINER_BloomFilter *bf;
  61
  62   /**
  63    * Entry for this pending request in the expiration heap, or NULL.
  64    */
  65   struct GNUNET_CONTAINER_HeapNode *hnode;
  66
  67   /**
  68    * Datastore queue entry for this request (or NULL for none).
  69    */
  70   struct GNUNET_DATASTORE_QueueEntry *qe;
  71
  72   /**
  73    * DHT request handle for this request (or NULL for none).
  74    */
  75   struct GNUNET_DHT_GetHandle *gh;
  76
  77   /**
  78    * Identity of the peer that we should use for the 'sender'
  79    * (recipient of the response) when forwarding (0 for none).
  80    */
  81   GNUNET_PEER_Id sender_pid;
  82
  83   /**
  84    * Number of valid entries in the 'replies_seen' array.
  85    */
  86   unsigned int replies_seen_count;
  87
  88   /**
  89    * Length of the 'replies_seen' array.
  90    */
  91   unsigned int replies_seen_size;
  92
  93   /**
  94    * Mingle value we currently use for the bf.
  95    */
  96   uint32_t mingle;
  97
  98 };
  99
 100
 101 /**
 102  * All pending requests, ordered by the query.  Entries
 103  * are of type 'struct GSF_PendingRequest*'.
 104  */
 105 static struct GNUNET_CONTAINER_MultiHashMap *pr_map;
 106
 107
 108 /**
 109  * Datastore 'PUT' load tracking.
 110  */
 111 static struct GNUNET_LOAD_Value *datastore_put_load;
 112
 113
 114 /**
 115  * Are we allowed to migrate content to this peer.
 116  */
 117 static int active_to_migration;
 118
 119
 120 /**
 121  * Heap with the request that will expire next at the top.  Contains
 122  * pointers of type "struct PendingRequest*"; these will *also* be
 123  * aliased from the "requests_by_peer" data structures and the
 124  * "requests_by_query" table.  Note that requests from our clients
 125  * don't expire and are thus NOT in the "requests_by_expiration"
 126  * (or the "requests_by_peer" tables).
 127  */
 128 static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
 129
 130
 131 /**
 132  * Maximum number of requests (from other peers, overall) that we're
 133  * willing to have pending at any given point in time.  Can be changed
 134  * via the configuration file (32k is just the default).
 135  */
 136 static unsigned long long max_pending_requests = (32 * 1024);
 137
 138
 139 /**
 140  * How many bytes should a bloomfilter be if we have already seen
 141  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
 142  * of bits set per entry.  Furthermore, we should not re-size the
 143  * filter too often (to keep it cheap).
 144  *
 145  * Since other peers will also add entries but not resize the filter,
 146  * we should generally pick a slightly larger size than what the
 147  * strict math would suggest.
 148  *
 149  * @return must be a power of two and smaller or equal to 2^15.
 150  */
 151 static size_t
 152 compute_bloomfilter_size (unsigned int entry_count)
 153 {
 154   size_t size;
 155   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
 156   uint16_t max = 1 << 15;
 157
 158   if (entry_count > max)
 159     return max;
 160   size = 8;
 161   while ((size < max) && (size < ideal))
 162     size *= 2;
 163   if (size > max)
 164     return max;
 165   return size;
 166 }
 167
 168
 169 /**
 170  * Recalculate our bloom filter for filtering replies.  This function
 171  * will create a new bloom filter from scratch, so it should only be
 172  * called if we have no bloomfilter at all (and hence can create a
 173  * fresh one of minimal size without problems) OR if our peer is the
 174  * initiator (in which case we may resize to larger than mimimum size).
 175  *
 176  * @param pr request for which the BF is to be recomputed
 177  * @return GNUNET_YES if a refresh actually happened
 178  */
 179 static int
 180 refresh_bloomfilter (struct GSF_PendingRequest *pr)
 181 {
 182   unsigned int i;
 183   size_t nsize;
 184   GNUNET_HashCode mhash;
 185
 186   nsize = compute_bloomfilter_size (pr->replies_seen_count);
 187   if ( (pr->bf != NULL) &&
 188        (nsize == GNUNET_CONTAINER_bloomfilter_get_size (pr->bf)) )
 189     return GNUNET_NO; /* size not changed */
 190   if (pr->bf != NULL)
 191     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
 192   pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 193                                          UINT32_MAX);
 194   pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 195                                               nsize,
 196                                               BLOOMFILTER_K);
 197   for (i=0;i<pr->replies_seen_count;i++)
 198     {
 199       GNUNET_BLOCK_mingle_hash (&pr->replies_seen[i],
 200                                 pr->mingle,
 201                                 &mhash);
 202       GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
 203     }
 204   return GNUNET_YES;
 205 }
 206
 207
 208 /**
 209  * Create a new pending request.
 210  *
 211  * @param options request options
 212  * @param type type of the block that is being requested
 213  * @param query key for the lookup
 214  * @param namespace namespace to lookup, NULL for no namespace
 215  * @param target preferred target for the request, NULL for none
 216  * @param bf_data raw data for bloom filter for known replies, can be NULL
 217  * @param bf_size number of bytes in bf_data
 218  * @param mingle mingle value for bf
 219  * @param anonymity_level desired anonymity level
 220  * @param priority maximum outgoing cummulative request priority to use
 221  * @param ttl current time-to-live for the request
 222  * @param sender_pid peer ID to use for the sender when forwarding, 0 for none
 223  * @param replies_seen hash codes of known local replies
 224  * @param replies_seen_count size of the 'replies_seen' array
 225  * @param rh handle to call when we get a reply
 226  * @param rh_cls closure for rh
 227  * @return handle for the new pending request
 228  */
 229 struct GSF_PendingRequest *
 230 GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
 231                              enum GNUNET_BLOCK_Type type,
 232                              const GNUNET_HashCode *query,
 233                              const GNUNET_HashCode *namespace,
 234                              const struct GNUNET_PeerIdentity *target,
 235                              const char *bf_data,
 236                              size_t bf_size,
 237                              uint32_t mingle,
 238                              uint32_t anonymity_level,
 239                              uint32_t priority,
 240                              int32_t ttl,
 241                              GNUNET_PEER_Id sender_pid,
 242                              const GNUNET_HashCode *replies_seen,
 243                              unsigned int replies_seen_count,
 244                              GSF_PendingRequestReplyHandler rh,
 245                              void *rh_cls)
 246 {
 247   struct GSF_PendingRequest *pr;
 248   struct GSF_PendingRequest *dpr;
 249
 250   pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest));
 251   pr->public_data.query = *query;
 252   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == type)
 253     {
 254       GNUNET_assert (NULL != namespace);
 255       pr->public_data.namespace = *namespace;
 256     }
 257   if (NULL != target)
 258     {
 259       pr->public_data.target = *target;
 260       pr->public_data.has_target = GNUNET_YES;
 261     }
 262   pr->public_data.anonymity_level = anonymity_level;
 263   pr->public_data.priority = priority;
 264   pr->public_data.original_priority = priority;
 265   pr->public_data.options = options;
 266   pr->public_data.type = type;
 267   pr->public_data.start_time = GNUNET_TIME_absolute_get ();
 268   pr->sender_pid = sender_pid;
 269   pr->rh = rh;
 270   pr->rh_cls = rh_cls;
 271   if (ttl >= 0)
 272     pr->public_data.ttl = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 273                                                                                            (uint32_t) ttl));
 274   else
 275     pr->public_data.ttl = GNUNET_TIME_absolute_subtract (pr->public_data.start_time,
 276                                                          GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 277                                                                                         (uint32_t) (- ttl)));
 278   if (replies_seen_count > 0)
 279     {
 280       pr->replies_seen_size = replies_seen_count;
 281       pr->replies_seen = GNUNET_malloc (sizeof (GNUNET_HashCode) * pr->replies_seen_size);
 282       memcpy (pr->replies_seen,
 283               replies_seen,
 284               replies_seen_count * sizeof (GNUNET_HashCode));
 285       pr->replies_seen_count = replies_seen_count;
 286     }
 287   if (NULL != bf_data)
 288     {
 289       pr->bf = GNUNET_CONTAINER_bloomfilter_init (bf_data,
 290                                                   bf_size,
 291                                                   BLOOMFILTER_K);
 292       pr->mingle = mingle;
 293     }
 294   else if ( (replies_seen_count > 0) &&
 295             (0 != (options & GSF_PRO_BLOOMFILTER_FULL_REFRESH)) )
 296     {
 297       GNUNET_assert (GNUNET_YES == refresh_bloomfilter (pr));
 298     }
 299   GNUNET_CONTAINER_multihashmap_put (pr_map,
 300                                      query,
 301                                      pr,
 302                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 303   if (0 != (options & GSF_PRO_REQUEST_EXPIRES))
 304     {
 305       pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
 306                                                 pr,
 307                                                 pr->public_data.ttl.abs_value);
 308       /* make sure we don't track too many requests */
 309       while (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
 310         {
 311           dpr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
 312           GNUNET_assert (dpr != NULL);
 313           if (pr == dpr)
 314             break; /* let the request live briefly... */
 315           dpr->rh (dpr->rh_cls,
 316                    dpr,
 317                    GNUNET_TIME_UNIT_FOREVER_ABS,
 318                    NULL, 0,
 319                    GNUNET_SYSERR);
 320           GSF_pending_request_cancel_ (dpr);
 321         }
 322     }
 323   return pr;
 324 }
 325
 326
 327 /**
 328  * Obtain the public data associated with a pending request
 329  *
 330  * @param pr pending request
 331  * @return associated public data
 332  */
 333 struct GSF_PendingRequestData *
 334 GSF_pending_request_get_data_ (struct GSF_PendingRequest *pr)
 335 {
 336   return &pr->public_data;
 337 }
 338
 339
 340 /**
 341  * Update a given pending request with additional replies
 342  * that have been seen.
 343  *
 344  * @param pr request to update
 345  * @param replies_seen hash codes of replies that we've seen
 346  * @param replies_seen_count size of the replies_seen array
 347  */
 348 void
 349 GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
 350                              const GNUNET_HashCode *replies_seen,
 351                              unsigned int replies_seen_count)
 352 {
 353   unsigned int i;
 354   GNUNET_HashCode mhash;
 355
 356   if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
 357     return; /* integer overflow */
 358   if (0 != (pr->public_data.options & GSF_PRO_BLOOMFILTER_FULL_REFRESH))
 359     {
 360       /* we're responsible for the BF, full refresh */
 361       if (replies_seen_count + pr->replies_seen_count > pr->replies_seen_size)
 362         GNUNET_array_grow (pr->replies_seen,
 363                            pr->replies_seen_size,
 364                            replies_seen_count + pr->replies_seen_count);
 365       memcpy (&pr->replies_seen[pr->replies_seen_count],
 366               replies_seen,
 367               sizeof (GNUNET_HashCode) * replies_seen_count);
 368       pr->replies_seen_count += replies_seen_count;
 369       if (GNUNET_NO == refresh_bloomfilter (pr))
 370         {
 371           /* bf not recalculated, simply extend it with new bits */
 372           for (i=0;i<pr->replies_seen_count;i++)
 373             {
 374               GNUNET_BLOCK_mingle_hash (&replies_seen[i],
 375                                         pr->mingle,
 376                                         &mhash);
 377               GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
 378             }
 379         }
 380     }
 381   else
 382     {
 383       if (NULL == pr->bf)
 384         {
 385           /* we're not the initiator, but the initiator did not give us
 386              any bloom-filter, so we need to create one on-the-fly */
 387           pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 388                                                  UINT32_MAX);
 389           pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 390                                                       compute_bloomfilter_size (replies_seen_count),
 391                                                       BLOOMFILTER_K);
 392         }
 393       for (i=0;i<pr->replies_seen_count;i++)
 394         {
 395           GNUNET_BLOCK_mingle_hash (&replies_seen[i],
 396                                     pr->mingle,
 397                                     &mhash);
 398           GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
 399         }
 400     }
 401 }
 402
 403
 404 /**
 405  * Generate the message corresponding to the given pending request for
 406  * transmission to other peers (or at least determine its size).
 407  *
 408  * @param pr request to generate the message for
 409  * @param do_route are we routing the reply
 410  * @param buf_size number of bytes available in buf
 411  * @param buf where to copy the message (can be NULL)
 412  * @return number of bytes needed (if > buf_size) or used
 413  */
 414 size_t
 415 GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
 416                                   int do_route,
 417                                   size_t buf_size,
 418                                   void *buf)
 419 {
 420   char lbuf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
 421   struct GetMessage *gm;
 422   GNUNET_HashCode *ext;
 423   size_t msize;
 424   unsigned int k;
 425   uint32_t bm;
 426   uint32_t prio;
 427   size_t bf_size;
 428   struct GNUNET_TIME_Absolute now;
 429   int64_t ttl;
 430
 431   k = 0;
 432   bm = 0;
 433   if (GNUNET_YES != do_route)
 434     {
 435       bm |= GET_MESSAGE_BIT_RETURN_TO;
 436       k++;
 437     }
 438   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
 439     {
 440       bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
 441       k++;
 442     }
 443   if (GNUNET_YES == pr->public_data.has_target)
 444     {
 445       bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
 446       k++;
 447     }
 448   bf_size = GNUNET_CONTAINER_bloomfilter_get_size (pr->bf);
 449   msize = sizeof (struct GetMessage) + bf_size + k * sizeof(GNUNET_HashCode);
 450   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
 451   if (buf_size < msize)
 452     return msize;
 453   gm = (struct GetMessage*) lbuf;
 454   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
 455   gm->header.size = htons (msize);
 456   gm->type = htonl (pr->public_data.type);
 457   if (GNUNET_YES == do_route)
 458     prio = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 459                                      pr->public_data.priority + 1);
 460   else
 461     prio = 0;
 462   pr->public_data.priority -= prio;
 463   gm->priority = htonl (prio);
 464   now = GNUNET_TIME_absolute_get ();
 465   ttl = (int64_t) (pr->public_data.ttl.abs_value - now.abs_value);
 466   gm->ttl = htonl (ttl / 1000);
 467   gm->filter_mutator = htonl(pr->mingle);
 468   gm->hash_bitmap = htonl (bm);
 469   gm->query = pr->public_data.query;
 470   ext = (GNUNET_HashCode*) &gm[1];
 471   k = 0;
 472   if (GNUNET_YES != do_route)
 473     GNUNET_PEER_resolve (pr->sender_pid,
 474                          (struct GNUNET_PeerIdentity*) &ext[k++]);
 475   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
 476     memcpy (&ext[k++],
 477             &pr->public_data.namespace,
 478             sizeof (GNUNET_HashCode));
 479   if (GNUNET_YES == pr->public_data.has_target)
 480     GNUNET_PEER_resolve (pr->sender_pid,
 481                          (struct GNUNET_PeerIdentity*) &ext[k++]);
 482   if (pr->bf != NULL)
 483     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
 484                                                (char*) &ext[k],
 485                                                bf_size);
 486   memcpy (buf, gm, msize);
 487   return msize;
 488 }
 489
 490
 491 /**
 492  * Iterator to free pending requests.
 493  *
 494  * @param cls closure, unused
 495  * @param key current key code
 496  * @param value value in the hash map (pending request)
 497  * @return GNUNET_YES (we should continue to iterate)
 498  */
 499 static int
 500 clean_request (void *cls,
 501                const GNUNET_HashCode * key,
 502                void *value)
 503 {
 504   struct GSF_PendingRequest *pr = value;
 505
 506   GNUNET_free_non_null (pr->replies_seen);
 507   if (NULL != pr->bf)
 508     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
 509   GNUNET_PEER_change_rc (pr->sender_pid, -1);
 510   if (NULL != pr->hnode)
 511     GNUNET_CONTAINER_heap_remove_node (requests_by_expiration_heap,
 512                                        pr->hnode);
 513   if (NULL != pr->qe)
 514     GNUNET_DATASTORE_cancel (pr->qe);
 515   if (NULL != pr->gh)
 516     GNUNET_DHT_get_stop (pr->gh);
 517   GNUNET_free (pr);
 518   return GNUNET_YES;
 519 }
 520
 521
 522 /**
 523  * Explicitly cancel a pending request.
 524  *
 525  * @param pr request to cancel
 526  */
 527 void
 528 GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr)
 529 {
 530   GNUNET_assert (GNUNET_OK ==
 531                  GNUNET_CONTAINER_multihashmap_remove (pr_map,
 532                                                        &pr->public_data.query,
 533                                                        pr));
 534   GNUNET_assert (GNUNET_YES ==
 535                  clean_request (NULL, &pr->public_data.query, pr));
 536 }
 537
 538
 539 /**
 540  * Iterate over all pending requests.
 541  *
 542  * @param it function to call for each request
 543  * @param cls closure for it
 544  */
 545 void
 546 GSF_iterate_pending_requests_ (GSF_PendingRequestIterator it,
 547                                void *cls)
 548 {
 549   GNUNET_CONTAINER_multihashmap_iterate (pr_map,
 550                                          (GNUNET_CONTAINER_HashMapIterator) it,
 551                                          cls);
 552 }
 553
 554
 555
 556
 557 /**
 558  * Closure for "process_reply" function.
 559  */
 560 struct ProcessReplyClosure
 561 {
 562   /**
 563    * The data for the reply.
 564    */
 565   const void *data;
 566
 567   /**
 568    * Who gave us this reply? NULL for local host (or DHT)
 569    */
 570   struct GSF_ConnectedPeer *sender;
 571
 572   /**
 573    * When the reply expires.
 574    */
 575   struct GNUNET_TIME_Absolute expiration;
 576
 577   /**
 578    * Size of data.
 579    */
 580   size_t size;
 581
 582   /**
 583    * Type of the block.
 584    */
 585   enum GNUNET_BLOCK_Type type;
 586
 587   /**
 588    * How much was this reply worth to us?
 589    */
 590   uint32_t priority;
 591
 592   /**
 593    * Anonymity requirements for this reply.
 594    */
 595   uint32_t anonymity_level;
 596
 597   /**
 598    * Evaluation result (returned).
 599    */
 600   enum GNUNET_BLOCK_EvaluationResult eval;
 601
 602   /**
 603    * Did we finish processing the associated request?
 604    */
 605   int finished;
 606
 607   /**
 608    * Did we find a matching request?
 609    */
 610   int request_found;
 611 };
 612
 613
 614 /**
 615  * Update the performance data for the sender (if any) since
 616  * the sender successfully answered one of our queries.
 617  *
 618  * @param prq information about the sender
 619  * @param pr request that was satisfied
 620  */
 621 static void
 622 update_request_performance_data (struct ProcessReplyClosure *prq,
 623                                  struct GSF_PendingRequest *pr)
 624 {
 625   if (prq->sender == NULL)
 626     return;
 627   GSF_peer_update_performance_ (prq->sender,
 628                                 pr->public_data.start_time,
 629                                 prq->priority);
 630 }
 631
 632
 633 /**
 634  * We have received a reply; handle it!
 635  *
 636  * @param cls response (struct ProcessReplyClosure)
 637  * @param key our query
 638  * @param value value in the hash map (info about the query)
 639  * @return GNUNET_YES (we should continue to iterate)
 640  */
 641 static int
 642 process_reply (void *cls,
 643                const GNUNET_HashCode * key,
 644                void *value)
 645 {
 646   struct ProcessReplyClosure *prq = cls;
 647   struct GSF_PendingRequest *pr = value;
 648   GNUNET_HashCode chash;
 649
 650 #if DEBUG_FS
 651   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 652               "Matched result (type %u) for query `%s' with pending request\n",
 653               (unsigned int) prq->type,
 654               GNUNET_h2s (key));
 655 #endif
 656   GNUNET_STATISTICS_update (GSF_stats,
 657                             gettext_noop ("# replies received and matched"),
 658                             1,
 659                             GNUNET_NO);
 660   prq->eval = GNUNET_BLOCK_evaluate (GSF_block_ctx,
 661                                      prq->type,
 662                                      key,
 663                                      &pr->bf,
 664                                      pr->mingle,
 665                                      &pr->public_data.namespace,
 666                                      (prq->type == GNUNET_BLOCK_TYPE_FS_SBLOCK) ? sizeof (GNUNET_HashCode) : 0,
 667                                      prq->data,
 668                                      prq->size);
 669   switch (prq->eval)
 670     {
 671     case GNUNET_BLOCK_EVALUATION_OK_MORE:
 672       update_request_performance_data (prq, pr);
 673       break;
 674     case GNUNET_BLOCK_EVALUATION_OK_LAST:
 675       /* short cut: stop processing early, no BF-update, etc. */
 676       update_request_performance_data (prq, pr);
 677       GNUNET_LOAD_update (GSF_rt_entry_lifetime,
 678                           GNUNET_TIME_absolute_get_duration (pr->public_data.start_time).rel_value);
 679       /* pass on to other peers / local clients */
 680       pr->rh (pr->rh_cls,
 681               pr,
 682               prq->expiration,
 683               prq->data, prq->size,
 684               GNUNET_NO);
 685       /* destroy request, we're done */
 686       GSF_pending_request_cancel_ (pr);
 687       return GNUNET_YES;
 688     case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
 689       GNUNET_STATISTICS_update (GSF_stats,
 690                                 gettext_noop ("# duplicate replies discarded (bloomfilter)"),
 691                                 1,
 692                                 GNUNET_NO);
 693 #if DEBUG_FS && 0
 694       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 695                   "Duplicate response `%s', discarding.\n",
 696                   GNUNET_h2s (&mhash));
 697 #endif
 698       return GNUNET_YES; /* duplicate */
 699     case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
 700       return GNUNET_YES; /* wrong namespace */
 701     case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
 702       GNUNET_break (0);
 703       return GNUNET_YES;
 704     case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
 705       GNUNET_break (0);
 706       return GNUNET_YES;
 707     case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
 708       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 709                   _("Unsupported block type %u\n"),
 710                   prq->type);
 711       return GNUNET_NO;
 712     }
 713   /* update bloomfilter */
 714   GNUNET_CRYPTO_hash (prq->data,
 715                       prq->size,
 716                       &chash);
 717   GSF_pending_request_update_ (pr, &chash, 1);
 718   if (NULL == prq->sender)
 719     {
 720 #if DEBUG_FS
 721       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 722                   "Found result for query `%s' in local datastore\n",
 723                   GNUNET_h2s (key));
 724 #endif
 725       GNUNET_STATISTICS_update (GSF_stats,
 726                                 gettext_noop ("# results found locally"),
 727                                 1,
 728                                 GNUNET_NO);
 729     }
 730   else
 731     {
 732       GSF_dht_lookup_ (pr);
 733     }
 734   prq->priority += pr->public_data.original_priority;
 735   pr->public_data.priority = 0;
 736   pr->public_data.original_priority = 0;
 737   pr->public_data.results_found++;
 738   prq->request_found = GNUNET_YES;
 739   /* finally, pass on to other peer / local client */
 740   pr->rh (pr->rh_cls,
 741           pr,
 742           prq->expiration,
 743           prq->data, prq->size,
 744           GNUNET_YES);
 745   return GNUNET_YES;
 746 }
 747
 748
 749 /**
 750  * Continuation called to notify client about result of the
 751  * operation.
 752  *
 753  * @param cls closure
 754  * @param success GNUNET_SYSERR on failure
 755  * @param msg NULL on success, otherwise an error message
 756  */
 757 static void
 758 put_migration_continuation (void *cls,
 759                             int success,
 760                             const char *msg)
 761 {
 762   struct GNUNET_TIME_Absolute *start = cls;
 763   struct GNUNET_TIME_Relative delay;
 764
 765   delay = GNUNET_TIME_absolute_get_duration (*start);
 766   GNUNET_free (start);
 767   /* FIXME: should we really update the load value on failure? */
 768   GNUNET_LOAD_update (datastore_put_load,
 769                       delay.rel_value);
 770   if (GNUNET_OK == success)
 771     return;
 772   GNUNET_STATISTICS_update (GSF_stats,
 773                             gettext_noop ("# datastore 'put' failures"),
 774                             1,
 775                             GNUNET_NO);
 776 }
 777
 778
 779 /**
 780  * Test if the DATABASE (PUT) load on this peer is too high
 781  * to even consider processing the query at
 782  * all.
 783  *
 784  * @return GNUNET_YES if the load is too high to do anything (load high)
 785  *         GNUNET_NO to process normally (load normal or low)
 786  */
 787 static int
 788 test_put_load_too_high (uint32_t priority)
 789 {
 790   double ld;
 791
 792   if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
 793     return GNUNET_NO; /* very fast */
 794   ld = GNUNET_LOAD_get_load (datastore_put_load);
 795   if (ld < 2.0 * (1 + priority))
 796     return GNUNET_NO;
 797   GNUNET_STATISTICS_update (GSF_stats,
 798                             gettext_noop ("# storage requests dropped due to high load"),
 799                             1,
 800                             GNUNET_NO);
 801   return GNUNET_YES;
 802 }
 803
 804
 805 /**
 806  * Iterator called on each result obtained for a DHT
 807  * operation that expects a reply
 808  *
 809  * @param cls closure
 810  * @param exp when will this value expire
 811  * @param key key of the result
 812  * @param get_path NULL-terminated array of pointers
 813  *                 to the peers on reverse GET path (or NULL if not recorded)
 814  * @param put_path NULL-terminated array of pointers
 815  *                 to the peers on the PUT path (or NULL if not recorded)
 816  * @param type type of the result
 817  * @param size number of bytes in data
 818  * @param data pointer to the result data
 819  */
 820 static void
 821 handle_dht_reply (void *cls,
 822                   struct GNUNET_TIME_Absolute exp,
 823                   const GNUNET_HashCode *key,
 824                   const struct GNUNET_PeerIdentity * const *get_path,
 825                   const struct GNUNET_PeerIdentity * const *put_path,
 826                   enum GNUNET_BLOCK_Type type,
 827                   size_t size,
 828                   const void *data)
 829 {
 830   struct GSF_PendingRequest *pr = cls;
 831   struct ProcessReplyClosure prq;
 832   struct GNUNET_TIME_Absolute *start;
 833
 834   memset (&prq, 0, sizeof (prq));
 835   prq.data = data;
 836   prq.expiration = exp;
 837   prq.size = size;
 838   prq.type = type;
 839   process_reply (&prq, key, pr);
 840   if ( (GNUNET_YES == active_to_migration) &&
 841        (GNUNET_NO == test_put_load_too_high (prq.priority)) )
 842     {
 843 #if DEBUG_FS
 844       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 845                   "Replicating result for query `%s' with priority %u\n",
 846                   GNUNET_h2s (&query),
 847                   prq.priority);
 848 #endif
 849       start = GNUNET_malloc (sizeof (struct GNUNET_TIME_Absolute));
 850       *start = GNUNET_TIME_absolute_get ();
 851       GNUNET_DATASTORE_put (GSF_dsh,
 852                             0, key, size, data,
 853                             type, prq.priority, 1 /* anonymity */,
 854                             exp,
 855                             1 + prq.priority, MAX_DATASTORE_QUEUE,
 856                             GNUNET_CONSTANTS_SERVICE_TIMEOUT,
 857                             &put_migration_continuation,
 858                             start);
 859     }
 860 }
 861
 862
 863 /**
 864  * Consider looking up the data in the DHT (anonymity-level permitting).
 865  *
 866  * @param pr the pending request to process
 867  */
 868 void
 869 GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
 870 {
 871   const void *xquery;
 872   size_t xquery_size;
 873   struct GNUNET_PeerIdentity pi;
 874   char buf[sizeof (GNUNET_HashCode) * 2];
 875
 876   if (0 != pr->public_data.anonymity_level)
 877     return;
 878   if (NULL != pr->gh)
 879     {
 880       GNUNET_DHT_get_stop (pr->gh);
 881       pr->gh = NULL;
 882     }
 883   xquery = NULL;
 884   xquery_size = 0;
 885   if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
 886     {
 887       xquery = buf;
 888       memcpy (buf, &pr->public_data.namespace, sizeof (GNUNET_HashCode));
 889       xquery_size = sizeof (GNUNET_HashCode);
 890     }
 891   if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
 892     {
 893       GNUNET_PEER_resolve (pr->sender_pid,
 894                            &pi);
 895       memcpy (&buf[xquery_size], &pi, sizeof (struct GNUNET_PeerIdentity));
 896       xquery_size += sizeof (struct GNUNET_PeerIdentity);
 897     }
 898   pr->gh = GNUNET_DHT_get_start (GSF_dht,
 899                                  GNUNET_TIME_UNIT_FOREVER_REL,
 900                                  pr->public_data.type,
 901                                  &pr->public_data.query,
 902                                  DEFAULT_GET_REPLICATION,
 903                                  GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
 904                                  pr->bf,
 905                                  pr->mingle,
 906                                  xquery,
 907                                  xquery_size,
 908                                  &handle_dht_reply,
 909                                  pr);
 910 }
 911
 912
 913 /**
 914  * We're processing (local) results for a search request
 915  * from another peer.  Pass applicable results to the
 916  * peer and if we are done either clean up (operation
 917  * complete) or forward to other peers (more results possible).
 918  *
 919  * @param cls our closure (struct LocalGetContext)
 920  * @param key key for the content
 921  * @param size number of bytes in data
 922  * @param data content stored
 923  * @param type type of the content
 924  * @param priority priority of the content
 925  * @param anonymity anonymity-level for the content
 926  * @param expiration expiration time for the content
 927  * @param uid unique identifier for the datum;
 928  *        maybe 0 if no unique identifier is available
 929  */
 930 static void
 931 process_local_reply (void *cls,
 932                      const GNUNET_HashCode * key,
 933                      size_t size,
 934                      const void *data,
 935                      enum GNUNET_BLOCK_Type type,
 936                      uint32_t priority,
 937                      uint32_t anonymity,
 938                      struct GNUNET_TIME_Absolute expiration,
 939                      uint64_t uid)
 940 {
 941 #if FIXME
 942   struct PendingRequest *pr = cls;
 943   struct ProcessReplyClosure prq;
 944   struct CheckDuplicateRequestClosure cdrc;
 945   GNUNET_HashCode query;
 946   unsigned int old_rf;
 947
 948   if (NULL == key)
 949     {
 950 #if DEBUG_FS > 1
 951       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 952                   "Done processing local replies, forwarding request to other peers.\n");
 953 #endif
 954       pr->qe = NULL;
 955       if (pr->client_request_list != NULL)
 956         {
 957           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
 958                                       GNUNET_YES);
 959           /* Figure out if this is a duplicate request and possibly
 960              merge 'struct PendingRequest' entries */
 961           cdrc.have = NULL;
 962           cdrc.pr = pr;
 963           GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
 964                                                       &pr->query,
 965                                                       &check_duplicate_request_client,
 966                                                       &cdrc);
 967           if (cdrc.have != NULL)
 968             {
 969 #if DEBUG_FS
 970               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 971                           "Received request for block `%s' twice from client, will only request once.\n",
 972                           GNUNET_h2s (&pr->query));
 973 #endif
 974
 975               destroy_pending_request (pr);
 976               return;
 977             }
 978         }
 979       if (pr->local_only == GNUNET_YES)
 980         {
 981           destroy_pending_request (pr);
 982           return;
 983         }
 984       /* no more results */
 985       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
 986         pr->task = GNUNET_SCHEDULER_add_now (&forward_request_task,
 987                                              pr);
 988       return;
 989     }
 990 #if DEBUG_FS
 991   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 992               "New local response to `%s' of type %u.\n",
 993               GNUNET_h2s (key),
 994               type);
 995 #endif
 996   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
 997     {
 998 #if DEBUG_FS
 999       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1000                   "Found ONDEMAND block, performing on-demand encoding\n");
1001 #endif
1002       GNUNET_STATISTICS_update (stats,
1003                                 gettext_noop ("# on-demand blocks matched requests"),
1004                                 1,
1005                                 GNUNET_NO);
1006       if (GNUNET_OK !=
1007           GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
1008                                             anonymity, expiration, uid,
1009                                             &process_local_reply,
1010                                             pr))
1011       if (pr->qe != NULL)
1012         {
1013           GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1014         }
1015       return;
1016     }
1017   old_rf = pr->results_found;
1018   memset (&prq, 0, sizeof (prq));
1019   prq.data = data;
1020   prq.expiration = expiration;
1021   prq.size = size;
1022   if (GNUNET_OK !=
1023       GNUNET_BLOCK_get_key (block_ctx,
1024                             type,
1025                             data,
1026                             size,
1027                             &query))
1028     {
1029       GNUNET_break (0);
1030       GNUNET_DATASTORE_remove (dsh,
1031                                key,
1032                                size, data,
1033                                -1, -1,
1034                                GNUNET_TIME_UNIT_FOREVER_REL,
1035                                NULL, NULL);
1036       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1037       return;
1038     }
1039   prq.type = type;
1040   prq.priority = priority;
1041   prq.finished = GNUNET_NO;
1042   prq.request_found = GNUNET_NO;
1043   prq.anonymity_level = anonymity;
1044   if ( (old_rf == 0) &&
1045        (pr->results_found == 0) )
1046     update_datastore_delays (pr->start_time);
1047   process_reply (&prq, key, pr);
1048   if (prq.finished == GNUNET_YES)
1049     return;
1050   if (pr->qe == NULL)
1051     return; /* done here */
1052   if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
1053     {
1054       pr->local_only = GNUNET_YES; /* do not forward */
1055       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1056       return;
1057     }
1058   if ( (pr->client_request_list == NULL) &&
1059        ( (GNUNET_YES == test_get_load_too_high (0)) ||
1060          (pr->results_found > 5 + 2 * pr->priority) ) )
1061     {
1062 #if DEBUG_FS > 2
1063       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1064                   "Load too high, done with request\n");
1065 #endif
1066       GNUNET_STATISTICS_update (stats,
1067                                 gettext_noop ("# processing result set cut short due to load"),
1068                                 1,
1069                                 GNUNET_NO);
1070       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1071       return;
1072     }
1073   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1074 #endif
1075 }
1076
1077
1078 /**
1079  * Look up the request in the local datastore.
1080  *
1081  * @param pr the pending request to process
1082  * @param cont function to call at the end
1083  * @param cont_cls closure for cont
1084  */
1085 void
1086 GSF_local_lookup_ (struct GSF_PendingRequest *pr,
1087                    GSF_LocalLookupContinuation cont,
1088                    void *cont_cls)
1089 {
1090   // FIXME: fix process_local_reply / cont!
1091   GNUNET_assert (NULL == pr->gh);
1092   pr->qe = GNUNET_DATASTORE_get (GSF_dsh,
1093                                  &pr->public_data.query,
1094                                  pr->public_data.type,
1095                                  1 /* queue priority */,
1096                                  1 /* max queue size */,
1097                                  GNUNET_TIME_UNIT_FOREVER_REL,
1098                                  &process_local_reply,
1099                                  pr);
1100 }
1101
1102
1103
1104 /**
1105  * Handle P2P "CONTENT" message.  Checks that the message is
1106  * well-formed and then checks if there are any pending requests for
1107  * this content and possibly passes it on (to local clients or other
1108  * peers).  Does NOT perform migration (content caching at this peer).
1109  *
1110  * @param cp the other peer involved (sender or receiver, NULL
1111  *        for loopback messages where we are both sender and receiver)
1112  * @param message the actual message
1113  * @return GNUNET_OK if the message was well-formed,
1114  *         GNUNET_SYSERR if the message was malformed (close connection,
1115  *         do not cache under any circumstances)
1116  */
1117 int
1118 GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
1119                          const struct GNUNET_MessageHeader *message)
1120 {
1121   const struct PutMessage *put;
1122   uint16_t msize;
1123   size_t dsize;
1124   enum GNUNET_BLOCK_Type type;
1125   struct GNUNET_TIME_Absolute expiration;
1126   GNUNET_HashCode query;
1127   struct ProcessReplyClosure prq;
1128   struct GNUNET_TIME_Relative block_time;
1129   double putl;
1130   struct GNUNET_TIME_Absolute *start;
1131
1132   msize = ntohs (message->size);
1133   if (msize < sizeof (struct PutMessage))
1134     {
1135       GNUNET_break_op(0);
1136       return GNUNET_SYSERR;
1137     }
1138   put = (const struct PutMessage*) message;
1139   dsize = msize - sizeof (struct PutMessage);
1140   type = ntohl (put->type);
1141   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
1142   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1143     return GNUNET_SYSERR;
1144   if (GNUNET_OK !=
1145       GNUNET_BLOCK_get_key (GSF_block_ctx,
1146                             type,
1147                             &put[1],
1148                             dsize,
1149                             &query))
1150     {
1151       GNUNET_break_op (0);
1152       return GNUNET_SYSERR;
1153     }
1154   /* now, lookup 'query' */
1155   prq.data = (const void*) &put[1];
1156   if (NULL != cp)
1157     prq.sender = cp;
1158   else
1159     prq.sender = NULL;
1160   prq.size = dsize;
1161   prq.type = type;
1162   prq.expiration = expiration;
1163   prq.priority = 0;
1164   prq.anonymity_level = 1;
1165   prq.finished = GNUNET_NO;
1166   prq.request_found = GNUNET_NO;
1167   GNUNET_CONTAINER_multihashmap_get_multiple (pr_map,
1168                                               &query,
1169                                               &process_reply,
1170                                               &prq);
1171   if (NULL != cp)
1172     {
1173       GSF_connected_peer_change_preference_ (cp, CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority);
1174       GSF_get_peer_performance_data_ (cp)->trust += prq.priority;
1175     }
1176   if ( (GNUNET_YES == active_to_migration) &&
1177        (GNUNET_NO == test_put_load_too_high (prq.priority)) )
1178     {
1179 #if DEBUG_FS
1180       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1181                   "Replicating result for query `%s' with priority %u\n",
1182                   GNUNET_h2s (&query),
1183                   prq.priority);
1184 #endif
1185       start = GNUNET_malloc (sizeof (struct GNUNET_TIME_Absolute));
1186       *start = GNUNET_TIME_absolute_get ();
1187       GNUNET_DATASTORE_put (GSF_dsh,
1188                             0, &query, dsize, &put[1],
1189                             type, prq.priority, 1 /* anonymity */,
1190                             expiration,
1191                             1 + prq.priority, MAX_DATASTORE_QUEUE,
1192                             GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1193                             &put_migration_continuation,
1194                             start);
1195     }
1196   putl = GNUNET_LOAD_get_load (datastore_put_load);
1197   if ( (NULL != (cp = prq.sender)) &&
1198        (GNUNET_NO == prq.request_found) &&
1199        ( (GNUNET_YES != active_to_migration) ||
1200          (putl > 2.5 * (1 + prq.priority)) ) )
1201     {
1202       if (GNUNET_YES != active_to_migration)
1203         putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
1204       block_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1205                                                   5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1206                                                                                    (unsigned int) (60000 * putl * putl)));
1207       GSF_block_peer_migration_ (cp, block_time);
1208     }
1209   return GNUNET_OK;
1210 }
1211
1212
1213 /**
1214  * Setup the subsystem.
1215  */
1216 void
1217 GSF_pending_request_init_ ()
1218 {
1219   if (GNUNET_OK !=
1220       GNUNET_CONFIGURATION_get_value_number (GSF_cfg,
1221                                              "fs",
1222                                              "MAX_PENDING_REQUESTS",
1223                                              &max_pending_requests))
1224     {
1225       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1226                   _("Configuration fails to specify `%s', assuming default value."),
1227                   "MAX_PENDING_REQUESTS");
1228     }
1229   pr_map = GNUNET_CONTAINER_multihashmap_create (32 * 1024);
1230   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1231 }
1232
1233
1234 /**
1235  * Shutdown the subsystem.
1236  */
1237 void
1238 GSF_pending_request_done_ ()
1239 {
1240   GNUNET_CONTAINER_multihashmap_iterate (pr_map,
1241                                          &clean_request,
1242                                          NULL);
1243   GNUNET_CONTAINER_multihashmap_destroy (pr_map);
1244   pr_map = NULL;
1245   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
1246   requests_by_expiration_heap = NULL;
1247 }
1248
1249
1250 /* end of gnunet-service-fs_pr.c */