src/fs/gnunet-service-fs_cp.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 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_cp.c
  23  * @brief API to handle 'connected peers'
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_load_lib.h"
  28 #include "gnunet_ats_service.h"
  29 #include "gnunet-service-fs.h"
  30 #include "gnunet-service-fs_cp.h"
  31 #include "gnunet-service-fs_pe.h"
  32 #include "gnunet-service-fs_pr.h"
  33 #include "gnunet-service-fs_push.h"
  34
  35
  36 /**
  37  * Ratio for moving average delay calculation.  The previous
  38  * average goes in with a factor of (n-1) into the calculation.
  39  * Must be > 0.
  40  */
  41 #define RUNAVG_DELAY_N 16
  42
  43 /**
  44  * How often do we flush trust values to disk?
  45  */
  46 #define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
  47
  48 /**
  49  * After how long do we discard a reply?
  50  */
  51 #define REPLY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2)
  52
  53
  54 /**
  55  * Handle to cancel a transmission request.
  56  */
  57 struct GSF_PeerTransmitHandle
  58 {
  59
  60   /**
  61    * Kept in a doubly-linked list.
  62    */
  63   struct GSF_PeerTransmitHandle *next;
  64
  65   /**
  66    * Kept in a doubly-linked list.
  67    */
  68   struct GSF_PeerTransmitHandle *prev;
  69
  70   /**
  71    * Handle for an active request for transmission to this
  72    * peer, or NULL (if core queue was full).
  73    */
  74   struct GNUNET_CORE_TransmitHandle *cth;
  75
  76   /**
  77    * Time when this transmission request was issued.
  78    */
  79   struct GNUNET_TIME_Absolute transmission_request_start_time;
  80
  81   /**
  82    * Timeout for this request.
  83    */
  84   struct GNUNET_TIME_Absolute timeout;
  85
  86   /**
  87    * Task called on timeout, or 0 for none.
  88    */
  89   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
  90
  91   /**
  92    * Function to call to get the actual message.
  93    */
  94   GSF_GetMessageCallback gmc;
  95
  96   /**
  97    * Peer this request targets.
  98    */
  99   struct GSF_ConnectedPeer *cp;
 100
 101   /**
 102    * Closure for 'gmc'.
 103    */
 104   void *gmc_cls;
 105
 106   /**
 107    * Size of the message to be transmitted.
 108    */
 109   size_t size;
 110
 111   /**
 112    * Set to 1 if we're currently in the process of calling
 113    * 'GNUNET_CORE_notify_transmit_ready' (so while cth is
 114    * NULL, we should not call notify_transmit_ready for this
 115    * handle right now).
 116    */
 117   unsigned int cth_in_progress;
 118
 119   /**
 120    * GNUNET_YES if this is a query, GNUNET_NO for content.
 121    */
 122   int is_query;
 123
 124   /**
 125    * Did we get a reservation already?
 126    */
 127   int was_reserved;
 128
 129   /**
 130    * Priority of this request.
 131    */
 132   uint32_t priority;
 133
 134 };
 135
 136
 137 /**
 138  * Handle for an entry in our delay list.
 139  */
 140 struct GSF_DelayedHandle
 141 {
 142
 143   /**
 144    * Kept in a doubly-linked list.
 145    */
 146   struct GSF_DelayedHandle *next;
 147
 148   /**
 149    * Kept in a doubly-linked list.
 150    */
 151   struct GSF_DelayedHandle *prev;
 152
 153   /**
 154    * Peer this transmission belongs to.
 155    */
 156   struct GSF_ConnectedPeer *cp;
 157
 158   /**
 159    * The PUT that was delayed.
 160    */
 161   struct PutMessage *pm;
 162
 163   /**
 164    * Task for the delay.
 165    */
 166   GNUNET_SCHEDULER_TaskIdentifier delay_task;
 167
 168   /**
 169    * Size of the message.
 170    */
 171   size_t msize;
 172
 173 };
 174
 175
 176 /**
 177  * Information per peer and request.
 178  */
 179 struct PeerRequest
 180 {
 181
 182   /**
 183    * Handle to generic request.
 184    */
 185   struct GSF_PendingRequest *pr;
 186
 187   /**
 188    * Handle to specific peer.
 189    */
 190   struct GSF_ConnectedPeer *cp;
 191
 192   /**
 193    * Task for asynchronous stopping of this request.
 194    */
 195   GNUNET_SCHEDULER_TaskIdentifier kill_task;
 196
 197 };
 198
 199
 200 /**
 201  * A connected peer.
 202  */
 203 struct GSF_ConnectedPeer
 204 {
 205
 206   /**
 207    * Performance data for this peer.
 208    */
 209   struct GSF_PeerPerformanceData ppd;
 210
 211   /**
 212    * Time until when we blocked this peer from migrating
 213    * data to us.
 214    */
 215   struct GNUNET_TIME_Absolute last_migration_block;
 216
 217   /**
 218    * Task scheduled to revive migration to this peer.
 219    */
 220   GNUNET_SCHEDULER_TaskIdentifier mig_revive_task;
 221
 222   /**
 223    * Messages (replies, queries, content migration) we would like to
 224    * send to this peer in the near future.  Sorted by priority, head.
 225    */
 226   struct GSF_PeerTransmitHandle *pth_head;
 227
 228   /**
 229    * Messages (replies, queries, content migration) we would like to
 230    * send to this peer in the near future.  Sorted by priority, tail.
 231    */
 232   struct GSF_PeerTransmitHandle *pth_tail;
 233
 234   /**
 235    * Messages (replies, queries, content migration) we would like to
 236    * send to this peer in the near future.  Sorted by priority, head.
 237    */
 238   struct GSF_DelayedHandle *delayed_head;
 239
 240   /**
 241    * Messages (replies, queries, content migration) we would like to
 242    * send to this peer in the near future.  Sorted by priority, tail.
 243    */
 244   struct GSF_DelayedHandle *delayed_tail;
 245
 246   /**
 247    * Migration stop message in our queue, or NULL if we have none pending.
 248    */
 249   struct GSF_PeerTransmitHandle *migration_pth;
 250
 251   /**
 252    * Context of our GNUNET_ATS_reserve_bandwidth call (or NULL).
 253    */
 254   struct GNUNET_ATS_ReservationContext *rc;
 255
 256   /**
 257    * Task scheduled if we need to retry bandwidth reservation later.
 258    */
 259   GNUNET_SCHEDULER_TaskIdentifier rc_delay_task;
 260
 261   /**
 262    * Active requests from this neighbour, map of query to 'struct PeerRequest'.
 263    */
 264   struct GNUNET_CONTAINER_MultiHashMap *request_map;
 265
 266   /**
 267    * Increase in traffic preference still to be submitted
 268    * to the core service for this peer.
 269    */
 270   uint64_t inc_preference;
 271
 272   /**
 273    * Trust rating for this peer on disk.
 274    */
 275   uint32_t disk_trust;
 276
 277   /**
 278    * Which offset in "last_p2p_replies" will be updated next?
 279    * (we go round-robin).
 280    */
 281   unsigned int last_p2p_replies_woff;
 282
 283   /**
 284    * Which offset in "last_client_replies" will be updated next?
 285    * (we go round-robin).
 286    */
 287   unsigned int last_client_replies_woff;
 288
 289   /**
 290    * Current offset into 'last_request_times' ring buffer.
 291    */
 292   unsigned int last_request_times_off;
 293
 294   /**
 295    * GNUNET_YES if we did successfully reserve 32k bandwidth,
 296    * GNUNET_NO if not.
 297    */
 298   int did_reserve;
 299
 300 };
 301
 302
 303 /**
 304  * Map from peer identities to 'struct GSF_ConnectPeer' entries.
 305  */
 306 static struct GNUNET_CONTAINER_MultiHashMap *cp_map;
 307
 308 /**
 309  * Where do we store trust information?
 310  */
 311 static char *trustDirectory;
 312
 313 /**
 314  * Handle to ATS service.
 315  */
 316 static struct GNUNET_ATS_PerformanceHandle *ats;
 317
 318 /**
 319  * Get the filename under which we would store the GNUNET_HELLO_Message
 320  * for the given host and protocol.
 321  * @return filename of the form DIRECTORY/HOSTID
 322  */
 323 static char *
 324 get_trust_filename (const struct GNUNET_PeerIdentity *id)
 325 {
 326   struct GNUNET_CRYPTO_HashAsciiEncoded fil;
 327   char *fn;
 328
 329   GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
 330   GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
 331   return fn;
 332 }
 333
 334
 335 /**
 336  * Find latency information in 'atsi'.
 337  *
 338  * @param atsi performance data
 339  * @param atsi_count number of records in 'atsi'
 340  * @return connection latency
 341  */
 342 static struct GNUNET_TIME_Relative
 343 get_latency (const struct GNUNET_ATS_Information *atsi, unsigned int atsi_count)
 344 {
 345   unsigned int i;
 346
 347   for (i = 0; i < atsi_count; i++)
 348     if (ntohl (atsi->type) == GNUNET_ATS_QUALITY_NET_DELAY)
 349       return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
 350                                             ntohl (atsi->value));
 351   return GNUNET_TIME_UNIT_SECONDS;
 352 }
 353
 354
 355 /**
 356  * Update the performance information kept for the given peer.
 357  *
 358  * @param cp peer record to update
 359  * @param atsi transport performance data
 360  * @param atsi_count number of records in 'atsi'
 361  */
 362 static void
 363 update_atsi (struct GSF_ConnectedPeer *cp,
 364              const struct GNUNET_ATS_Information *atsi, unsigned int atsi_count)
 365 {
 366   struct GNUNET_TIME_Relative latency;
 367
 368   latency = get_latency (atsi, atsi_count);
 369   GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay, latency);
 370   /* LATER: merge atsi into cp's performance data (if we ever care...) */
 371 }
 372
 373
 374 /**
 375  * Return the performance data record for the given peer
 376  *
 377  * @param cp peer to query
 378  * @return performance data record for the peer
 379  */
 380 struct GSF_PeerPerformanceData *
 381 GSF_get_peer_performance_data_ (struct GSF_ConnectedPeer *cp)
 382 {
 383   return &cp->ppd;
 384 }
 385
 386
 387 /**
 388  * Core is ready to transmit to a peer, get the message.
 389  *
 390  * @param cls the 'struct GSF_PeerTransmitHandle' of the message
 391  * @param size number of bytes core is willing to take
 392  * @param buf where to copy the message
 393  * @return number of bytes copied to buf
 394  */
 395 static size_t
 396 peer_transmit_ready_cb (void *cls, size_t size, void *buf);
 397
 398
 399 /**
 400  * Function called by core upon success or failure of our bandwidth reservation request.
 401  *
 402  * @param cls the 'struct GSF_ConnectedPeer' of the peer for which we made the request
 403  * @param peer identifies the peer
 404  * @param amount set to the amount that was actually reserved or unreserved;
 405  *               either the full requested amount or zero (no partial reservations)
 406  * @param res_delay if the reservation could not be satisfied (amount was 0), how
 407  *        long should the client wait until re-trying?
 408  */
 409 static void
 410 ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer,
 411                       int32_t amount, struct GNUNET_TIME_Relative res_delay);
 412
 413
 414 /**
 415  * If ready (bandwidth reserved), try to schedule transmission via
 416  * core for the given handle.
 417  *
 418  * @param pth transmission handle to schedule
 419  */
 420 static void
 421 schedule_transmission (struct GSF_PeerTransmitHandle *pth)
 422 {
 423   struct GSF_ConnectedPeer *cp;
 424   struct GNUNET_PeerIdentity target;
 425
 426   if ((NULL != pth->cth) || (0 != pth->cth_in_progress))
 427     return;                     /* already done */
 428   cp = pth->cp;
 429   GNUNET_assert (0 != cp->ppd.pid);
 430   GNUNET_PEER_resolve (cp->ppd.pid, &target);
 431
 432   if (0 != cp->inc_preference)
 433   {
 434     GNUNET_ATS_change_preference (ats, &target, GNUNET_ATS_PREFERENCE_BANDWIDTH,
 435                                   (double) cp->inc_preference,
 436                                   GNUNET_ATS_PREFERENCE_END);
 437     cp->inc_preference = 0;
 438   }
 439
 440   if ((GNUNET_YES == pth->is_query) && (GNUNET_YES != pth->was_reserved))
 441   {
 442     /* query, need reservation */
 443     if (GNUNET_YES != cp->did_reserve)
 444       return;                   /* not ready */
 445     cp->did_reserve = GNUNET_NO;
 446     /* reservation already done! */
 447     pth->was_reserved = GNUNET_YES;
 448     cp->rc =
 449         GNUNET_ATS_reserve_bandwidth (ats, &target, DBLOCK_SIZE,
 450                                       &ats_reserve_callback, cp);
 451   }
 452   GNUNET_assert (pth->cth == NULL);
 453   pth->cth_in_progress++;
 454   pth->cth =
 455       GNUNET_CORE_notify_transmit_ready (GSF_core, GNUNET_YES, pth->priority,
 456                                          GNUNET_TIME_absolute_get_remaining
 457                                          (pth->timeout), &target, pth->size,
 458                                          &peer_transmit_ready_cb, pth);
 459   GNUNET_assert (0 < pth->cth_in_progress--);
 460 }
 461
 462
 463 /**
 464  * Core is ready to transmit to a peer, get the message.
 465  *
 466  * @param cls the 'struct GSF_PeerTransmitHandle' of the message
 467  * @param size number of bytes core is willing to take
 468  * @param buf where to copy the message
 469  * @return number of bytes copied to buf
 470  */
 471 static size_t
 472 peer_transmit_ready_cb (void *cls, size_t size, void *buf)
 473 {
 474   struct GSF_PeerTransmitHandle *pth = cls;
 475   struct GSF_PeerTransmitHandle *pos;
 476   struct GSF_ConnectedPeer *cp;
 477   size_t ret;
 478
 479   GNUNET_assert ((NULL == buf) || (pth->size <= size));
 480   pth->cth = NULL;
 481   if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK)
 482   {
 483     GNUNET_SCHEDULER_cancel (pth->timeout_task);
 484     pth->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 485   }
 486   cp = pth->cp;
 487   GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth);
 488   if (GNUNET_YES == pth->is_query)
 489   {
 490     cp->ppd.last_request_times[(cp->last_request_times_off++) %
 491                                MAX_QUEUE_PER_PEER] =
 492         GNUNET_TIME_absolute_get ();
 493     GNUNET_assert (0 < cp->ppd.pending_queries--);
 494   }
 495   else if (GNUNET_NO == pth->is_query)
 496   {
 497     GNUNET_assert (0 < cp->ppd.pending_replies--);
 498   }
 499   GNUNET_LOAD_update (cp->ppd.transmission_delay,
 500                       GNUNET_TIME_absolute_get_duration
 501                       (pth->transmission_request_start_time).rel_value);
 502   ret = pth->gmc (pth->gmc_cls, size, buf);
 503   GNUNET_assert (NULL == pth->cth);
 504   for (pos = cp->pth_head; pos != NULL; pos = pos->next)
 505   {
 506     GNUNET_assert (pos != pth);
 507     schedule_transmission (pos);
 508   }
 509   GNUNET_assert (pth->cth == NULL);
 510   GNUNET_assert (pth->cth_in_progress == 0);
 511   GNUNET_free (pth);
 512   return ret;
 513 }
 514
 515
 516 /**
 517  * (re)try to reserve bandwidth from the given peer.
 518  *
 519  * @param cls the 'struct GSF_ConnectedPeer' to reserve from
 520  * @param tc scheduler context
 521  */
 522 static void
 523 retry_reservation (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 524 {
 525   struct GSF_ConnectedPeer *cp = cls;
 526   struct GNUNET_PeerIdentity target;
 527
 528   GNUNET_PEER_resolve (cp->ppd.pid, &target);
 529   cp->rc_delay_task = GNUNET_SCHEDULER_NO_TASK;
 530   cp->rc =
 531       GNUNET_ATS_reserve_bandwidth (ats, &target, DBLOCK_SIZE,
 532                                     &ats_reserve_callback, cp);
 533 }
 534
 535
 536 /**
 537  * Function called by core upon success or failure of our bandwidth reservation request.
 538  *
 539  * @param cls the 'struct GSF_ConnectedPeer' of the peer for which we made the request
 540  * @param peer identifies the peer
 541  * @param amount set to the amount that was actually reserved or unreserved;
 542  *               either the full requested amount or zero (no partial reservations)
 543  * @param res_delay if the reservation could not be satisfied (amount was 0), how
 544  *        long should the client wait until re-trying?
 545  */
 546 static void
 547 ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer,
 548                       int32_t amount, struct GNUNET_TIME_Relative res_delay)
 549 {
 550   struct GSF_ConnectedPeer *cp = cls;
 551   struct GSF_PeerTransmitHandle *pth;
 552
 553   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 554               "Reserved %d bytes / need to wait %llu ms for reservation\n",
 555               (int) amount, (unsigned long long) res_delay.rel_value);
 556   cp->rc = NULL;
 557   if (0 == amount)
 558   {
 559     cp->rc_delay_task =
 560         GNUNET_SCHEDULER_add_delayed (res_delay, &retry_reservation, cp);
 561     return;
 562   }
 563   cp->did_reserve = GNUNET_YES;
 564   pth = cp->pth_head;
 565   if ((NULL != pth) && (NULL == pth->cth))
 566   {
 567     /* reservation success, try transmission now! */
 568     pth->cth_in_progress++;
 569     pth->cth =
 570         GNUNET_CORE_notify_transmit_ready (GSF_core, GNUNET_YES, pth->priority,
 571                                            GNUNET_TIME_absolute_get_remaining
 572                                            (pth->timeout), peer, pth->size,
 573                                            &peer_transmit_ready_cb, pth);
 574     GNUNET_assert (0 < pth->cth_in_progress--);
 575   }
 576 }
 577
 578
 579 /**
 580  * A peer connected to us.  Setup the connected peer
 581  * records.
 582  *
 583  * @param peer identity of peer that connected
 584  * @param atsi performance data for the connection
 585  * @param atsi_count number of records in 'atsi'
 586  * @return handle to connected peer entry
 587  */
 588 struct GSF_ConnectedPeer *
 589 GSF_peer_connect_handler_ (const struct GNUNET_PeerIdentity *peer,
 590                            const struct GNUNET_ATS_Information *atsi,
 591                            unsigned int atsi_count)
 592 {
 593   struct GSF_ConnectedPeer *cp;
 594   char *fn;
 595   uint32_t trust;
 596
 597   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connected to peer %s\n",
 598               GNUNET_i2s (peer));
 599   cp = GNUNET_malloc (sizeof (struct GSF_ConnectedPeer));
 600   cp->ppd.pid = GNUNET_PEER_intern (peer);
 601   cp->ppd.transmission_delay = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_ZERO);
 602   cp->rc =
 603       GNUNET_ATS_reserve_bandwidth (ats, peer, DBLOCK_SIZE,
 604                                     &ats_reserve_callback, cp);
 605   fn = get_trust_filename (peer);
 606   if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
 607       (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
 608     cp->disk_trust = cp->ppd.trust = ntohl (trust);
 609   GNUNET_free (fn);
 610   cp->request_map = GNUNET_CONTAINER_multihashmap_create (128);
 611   GNUNET_break (GNUNET_OK ==
 612                 GNUNET_CONTAINER_multihashmap_put (cp_map, &peer->hashPubKey,
 613                                                    cp,
 614                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 615   GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# peers connected"),
 616                          GNUNET_CONTAINER_multihashmap_size (cp_map),
 617                          GNUNET_NO);
 618   update_atsi (cp, atsi, atsi_count);
 619   GSF_push_start_ (cp);
 620   return cp;
 621 }
 622
 623
 624 /**
 625  * It may be time to re-start migrating content to this
 626  * peer.  Check, and if so, restart migration.
 627  *
 628  * @param cls the 'struct GSF_ConnectedPeer'
 629  * @param tc scheduler context
 630  */
 631 static void
 632 revive_migration (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 633 {
 634   struct GSF_ConnectedPeer *cp = cls;
 635   struct GNUNET_TIME_Relative bt;
 636
 637   cp->mig_revive_task = GNUNET_SCHEDULER_NO_TASK;
 638   bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until);
 639   if (0 != bt.rel_value)
 640   {
 641     /* still time left... */
 642     cp->mig_revive_task =
 643         GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
 644     return;
 645   }
 646   GSF_push_start_ (cp);
 647 }
 648
 649
 650 /**
 651  * Get a handle for a connected peer.
 652  *
 653  * @param peer peer's identity
 654  * @return NULL if the peer is not currently connected
 655  */
 656 struct GSF_ConnectedPeer *
 657 GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer)
 658 {
 659   if (NULL == cp_map)
 660     return NULL;
 661   return GNUNET_CONTAINER_multihashmap_get (cp_map, &peer->hashPubKey);
 662 }
 663
 664
 665 /**
 666  * Handle P2P "MIGRATION_STOP" message.
 667  *
 668  * @param cls closure, always NULL
 669  * @param other the other peer involved (sender or receiver, NULL
 670  *        for loopback messages where we are both sender and receiver)
 671  * @param message the actual message
 672  * @param atsi performance information
 673  * @param atsi_count number of records in 'atsi'
 674  * @return GNUNET_OK to keep the connection open,
 675  *         GNUNET_SYSERR to close it (signal serious error)
 676  */
 677 int
 678 GSF_handle_p2p_migration_stop_ (void *cls,
 679                                 const struct GNUNET_PeerIdentity *other,
 680                                 const struct GNUNET_MessageHeader *message,
 681                                 const struct GNUNET_ATS_Information *atsi,
 682                                 unsigned int atsi_count)
 683 {
 684   struct GSF_ConnectedPeer *cp;
 685   const struct MigrationStopMessage *msm;
 686   struct GNUNET_TIME_Relative bt;
 687
 688   msm = (const struct MigrationStopMessage *) message;
 689   cp = GSF_peer_get_ (other);
 690   if (cp == NULL)
 691   {
 692     GNUNET_break (0);
 693     return GNUNET_OK;
 694   }
 695   GNUNET_STATISTICS_update (GSF_stats,
 696                             gettext_noop ("# migration stop messages received"),
 697                             1, GNUNET_NO);
 698   bt = GNUNET_TIME_relative_ntoh (msm->duration);
 699   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 700               _("Migration of content to peer `%s' blocked for %llu ms\n"),
 701               GNUNET_i2s (other), (unsigned long long) bt.rel_value);
 702   cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt);
 703   if (cp->mig_revive_task == GNUNET_SCHEDULER_NO_TASK)
 704   {
 705     GSF_push_stop_ (cp);
 706     cp->mig_revive_task =
 707         GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
 708   }
 709   update_atsi (cp, atsi, atsi_count);
 710   return GNUNET_OK;
 711 }
 712
 713
 714 /**
 715  * Copy reply and free put message.
 716  *
 717  * @param cls the 'struct PutMessage'
 718  * @param buf_size number of bytes available in buf
 719  * @param buf where to copy the message, NULL on error (peer disconnect)
 720  * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
 721  */
 722 static size_t
 723 copy_reply (void *cls, size_t buf_size, void *buf)
 724 {
 725   struct PutMessage *pm = cls;
 726   size_t size;
 727
 728   if (buf != NULL)
 729   {
 730     GNUNET_assert (buf_size >= ntohs (pm->header.size));
 731     size = ntohs (pm->header.size);
 732     memcpy (buf, pm, size);
 733     GNUNET_STATISTICS_update (GSF_stats,
 734                               gettext_noop
 735                               ("# replies transmitted to other peers"), 1,
 736                               GNUNET_NO);
 737   }
 738   else
 739   {
 740     size = 0;
 741     GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# replies dropped"), 1,
 742                               GNUNET_NO);
 743   }
 744   GNUNET_free (pm);
 745   return size;
 746 }
 747
 748
 749 /**
 750  * Free resources associated with the given peer request.
 751  *
 752  * @param peerreq request to free
 753  * @param query associated key for the request
 754  */
 755 static void
 756 free_pending_request (struct PeerRequest *peerreq,
 757                       const GNUNET_HashCode *query)
 758 {
 759   struct GSF_ConnectedPeer *cp = peerreq->cp;
 760
 761   if (peerreq->kill_task != GNUNET_SCHEDULER_NO_TASK)
 762   {
 763     GNUNET_SCHEDULER_cancel (peerreq->kill_task);
 764     peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK;
 765   }
 766   GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"),
 767                             -1, GNUNET_NO);
 768   GNUNET_break (GNUNET_YES ==
 769                 GNUNET_CONTAINER_multihashmap_remove (cp->request_map,
 770                                                       query, peerreq));
 771   GNUNET_free (peerreq);
 772 }
 773
 774
 775 /**
 776  * Cancel all requests associated with the peer.
 777  *
 778  * @param cls unused
 779  * @param query hash code of the request
 780  * @param value the 'struct GSF_PendingRequest'
 781  * @return GNUNET_YES (continue to iterate)
 782  */
 783 static int
 784 cancel_pending_request (void *cls, const GNUNET_HashCode * query, void *value)
 785 {
 786   struct PeerRequest *peerreq = value;
 787   struct GSF_PendingRequest *pr = peerreq->pr;
 788   struct GSF_PendingRequestData *prd;
 789
 790   prd = GSF_pending_request_get_data_ (pr);
 791   GSF_pending_request_cancel_ (pr, GNUNET_NO);
 792   free_pending_request (peerreq, &prd->query);
 793   return GNUNET_OK;
 794 }
 795
 796
 797 /**
 798  * Free the given request.
 799  *
 800  * @param cls the request to free
 801  * @param tc task context
 802  */
 803 static void
 804 peer_request_destroy (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 805 {
 806   struct PeerRequest *peerreq = cls;
 807   struct GSF_PendingRequest *pr = peerreq->pr;
 808   struct GSF_PendingRequestData *prd;
 809
 810   peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK;
 811   prd = GSF_pending_request_get_data_ (pr);
 812   cancel_pending_request (NULL, &prd->query, peerreq);
 813 }
 814
 815
 816 /**
 817  * The artificial delay is over, transmit the message now.
 818  *
 819  * @param cls the 'struct GSF_DelayedHandle' with the message
 820  * @param tc scheduler context
 821  */
 822 static void
 823 transmit_delayed_now (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 824 {
 825   struct GSF_DelayedHandle *dh = cls;
 826   struct GSF_ConnectedPeer *cp = dh->cp;
 827
 828   GNUNET_CONTAINER_DLL_remove (cp->delayed_head, cp->delayed_tail, dh);
 829   if (0 != (GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason))
 830   {
 831     GNUNET_free (dh->pm);
 832     GNUNET_free (dh);
 833     return;
 834   }
 835   (void) GSF_peer_transmit_ (cp, GNUNET_NO, UINT32_MAX, REPLY_TIMEOUT,
 836                              dh->msize, &copy_reply, dh->pm);
 837   GNUNET_free (dh);
 838 }
 839
 840
 841 /**
 842  * Get the randomized delay a response should be subjected to.
 843  *
 844  * @return desired delay
 845  */
 846 static struct GNUNET_TIME_Relative
 847 get_randomized_delay ()
 848 {
 849   struct GNUNET_TIME_Relative ret;
 850
 851   ret =
 852       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
 853                                      GNUNET_CRYPTO_random_u32
 854                                      (GNUNET_CRYPTO_QUALITY_WEAK,
 855                                       2 * GSF_avg_latency.rel_value + 1));
 856   GNUNET_STATISTICS_update (GSF_stats,
 857                             gettext_noop
 858                             ("# artificial delays introduced (ms)"),
 859                             ret.rel_value, GNUNET_NO);
 860
 861   return ret;
 862 }
 863
 864
 865 /**
 866  * Handle a reply to a pending request.  Also called if a request
 867  * expires (then with data == NULL).  The handler may be called
 868  * many times (depending on the request type), but will not be
 869  * called during or after a call to GSF_pending_request_cancel
 870  * and will also not be called anymore after a call signalling
 871  * expiration.
 872  *
 873  * @param cls 'struct PeerRequest' this is an answer for
 874  * @param eval evaluation of the result
 875  * @param pr handle to the original pending request
 876  * @param reply_anonymity_level anonymity level for the reply, UINT32_MAX for "unknown"
 877  * @param expiration when does 'data' expire?
 878  * @param type type of the block
 879  * @param data response data, NULL on request expiration
 880  * @param data_len number of bytes in data
 881  */
 882 static void
 883 handle_p2p_reply (void *cls, enum GNUNET_BLOCK_EvaluationResult eval,
 884                   struct GSF_PendingRequest *pr, uint32_t reply_anonymity_level,
 885                   struct GNUNET_TIME_Absolute expiration,
 886                   enum GNUNET_BLOCK_Type type, const void *data,
 887                   size_t data_len)
 888 {
 889   struct PeerRequest *peerreq = cls;
 890   struct GSF_ConnectedPeer *cp = peerreq->cp;
 891   struct GSF_PendingRequestData *prd;
 892   struct PutMessage *pm;
 893   size_t msize;
 894
 895   GNUNET_assert (data_len + sizeof (struct PutMessage) <
 896                  GNUNET_SERVER_MAX_MESSAGE_SIZE);
 897   GNUNET_assert (peerreq->pr == pr);
 898   prd = GSF_pending_request_get_data_ (pr);
 899   if (NULL == data)
 900   {
 901     free_pending_request (peerreq, &prd->query);
 902     return;
 903   }
 904   GNUNET_break (type != GNUNET_BLOCK_TYPE_ANY);
 905   if ((prd->type != type) && (prd->type != GNUNET_BLOCK_TYPE_ANY))
 906   {
 907     GNUNET_STATISTICS_update (GSF_stats,
 908                               gettext_noop
 909                               ("# replies dropped due to type mismatch"),
 910                                 1, GNUNET_NO);
 911     return;
 912   }
 913 #if DEBUG_FS
 914   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 915               "Transmitting result for query `%s' to peer\n",
 916               GNUNET_h2s (&prd->query));
 917 #endif
 918   GNUNET_STATISTICS_update (GSF_stats,
 919                             gettext_noop ("# replies received for other peers"),
 920                             1, GNUNET_NO);
 921   msize = sizeof (struct PutMessage) + data_len;
 922   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 923   {
 924     GNUNET_break (0);
 925     return;
 926   }
 927   if ((reply_anonymity_level != UINT32_MAX) && (reply_anonymity_level > 1))
 928   {
 929     if (reply_anonymity_level - 1 > GSF_cover_content_count)
 930     {
 931       GNUNET_STATISTICS_update (GSF_stats,
 932                                 gettext_noop
 933                                 ("# replies dropped due to insufficient cover traffic"),
 934                                 1, GNUNET_NO);
 935       return;
 936     }
 937     GSF_cover_content_count -= (reply_anonymity_level - 1);
 938   }
 939
 940   pm = GNUNET_malloc (msize);
 941   pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
 942   pm->header.size = htons (msize);
 943   pm->type = htonl (type);
 944   pm->expiration = GNUNET_TIME_absolute_hton (expiration);
 945   memcpy (&pm[1], data, data_len);
 946   if ((reply_anonymity_level != UINT32_MAX) && (reply_anonymity_level != 0) &&
 947       (GSF_enable_randomized_delays == GNUNET_YES))
 948   {
 949     struct GSF_DelayedHandle *dh;
 950
 951     dh = GNUNET_malloc (sizeof (struct GSF_DelayedHandle));
 952     dh->cp = cp;
 953     dh->pm = pm;
 954     dh->msize = msize;
 955     GNUNET_CONTAINER_DLL_insert (cp->delayed_head, cp->delayed_tail, dh);
 956     dh->delay_task =
 957         GNUNET_SCHEDULER_add_delayed (get_randomized_delay (),
 958                                       &transmit_delayed_now, dh);
 959   }
 960   else
 961   {
 962     (void) GSF_peer_transmit_ (cp, GNUNET_NO, UINT32_MAX, REPLY_TIMEOUT, msize,
 963                                &copy_reply, pm);
 964   }
 965   if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST)
 966     return;
 967   if (GNUNET_SCHEDULER_NO_TASK == peerreq->kill_task)
 968   {
 969     GNUNET_STATISTICS_update (GSF_stats,
 970                               gettext_noop
 971                               ("# P2P searches destroyed due to ultimate reply"),
 972                               1, GNUNET_NO);
 973     peerreq->kill_task =
 974         GNUNET_SCHEDULER_add_now (&peer_request_destroy, peerreq);
 975   }
 976 }
 977
 978
 979 /**
 980  * Increase the host credit by a value.
 981  *
 982  * @param cp which peer to change the trust value on
 983  * @param value is the int value by which the
 984  *  host credit is to be increased or decreased
 985  * @returns the actual change in trust (positive or negative)
 986  */
 987 static int
 988 change_host_trust (struct GSF_ConnectedPeer *cp, int value)
 989 {
 990   if (value == 0)
 991     return 0;
 992   GNUNET_assert (cp != NULL);
 993   if (value > 0)
 994   {
 995     if (cp->ppd.trust + value < cp->ppd.trust)
 996     {
 997       value = UINT32_MAX - cp->ppd.trust;
 998       cp->ppd.trust = UINT32_MAX;
 999     }
1000     else
1001       cp->ppd.trust += value;
1002   }
1003   else
1004   {
1005     if (cp->ppd.trust < -value)
1006     {
1007       value = -cp->ppd.trust;
1008       cp->ppd.trust = 0;
1009     }
1010     else
1011       cp->ppd.trust += value;
1012   }
1013   return value;
1014 }
1015
1016
1017 /**
1018  * We've received a request with the specified priority.  Bound it
1019  * according to how much we trust the given peer.
1020  *
1021  * @param prio_in requested priority
1022  * @param cp the peer making the request
1023  * @return effective priority
1024  */
1025 static int32_t
1026 bound_priority (uint32_t prio_in, struct GSF_ConnectedPeer *cp)
1027 {
1028 #define N ((double)128.0)
1029   uint32_t ret;
1030   double rret;
1031   int ld;
1032
1033   ld = GSF_test_get_load_too_high_ (0);
1034   if (ld == GNUNET_SYSERR)
1035   {
1036     GNUNET_STATISTICS_update (GSF_stats,
1037                               gettext_noop
1038                               ("# requests done for free (low load)"), 1,
1039                               GNUNET_NO);
1040     return 0;                   /* excess resources */
1041   }
1042   if (prio_in > INT32_MAX)
1043     prio_in = INT32_MAX;
1044   ret = -change_host_trust (cp, -(int) prio_in);
1045   if (ret > 0)
1046   {
1047     if (ret > GSF_current_priorities + N)
1048       rret = GSF_current_priorities + N;
1049     else
1050       rret = ret;
1051     GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N;
1052   }
1053   if ((ld == GNUNET_YES) && (ret > 0))
1054   {
1055     /* try with charging */
1056     ld = GSF_test_get_load_too_high_ (ret);
1057   }
1058   if (ld == GNUNET_YES)
1059   {
1060     GNUNET_STATISTICS_update (GSF_stats,
1061                               gettext_noop
1062                               ("# request dropped, priority insufficient"), 1,
1063                               GNUNET_NO);
1064     /* undo charge */
1065     change_host_trust (cp, (int) ret);
1066     return -1;                  /* not enough resources */
1067   }
1068   else
1069   {
1070     GNUNET_STATISTICS_update (GSF_stats,
1071                               gettext_noop
1072                               ("# requests done for a price (normal load)"), 1,
1073                               GNUNET_NO);
1074   }
1075 #undef N
1076   return ret;
1077 }
1078
1079
1080 /**
1081  * The priority level imposes a bound on the maximum
1082  * value for the ttl that can be requested.
1083  *
1084  * @param ttl_in requested ttl
1085  * @param prio given priority
1086  * @return ttl_in if ttl_in is below the limit,
1087  *         otherwise the ttl-limit for the given priority
1088  */
1089 static int32_t
1090 bound_ttl (int32_t ttl_in, uint32_t prio)
1091 {
1092   unsigned long long allowed;
1093
1094   if (ttl_in <= 0)
1095     return ttl_in;
1096   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
1097   if (ttl_in > allowed)
1098   {
1099     if (allowed >= (1 << 30))
1100       return 1 << 30;
1101     return allowed;
1102   }
1103   return ttl_in;
1104 }
1105
1106
1107 /**
1108  * Handle P2P "QUERY" message.  Creates the pending request entry
1109  * and sets up all of the data structures to that we will
1110  * process replies properly.  Does not initiate forwarding or
1111  * local database lookups.
1112  *
1113  * @param other the other peer involved (sender or receiver, NULL
1114  *        for loopback messages where we are both sender and receiver)
1115  * @param message the actual message
1116  * @return pending request handle, NULL on error
1117  */
1118 struct GSF_PendingRequest *
1119 GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other,
1120                        const struct GNUNET_MessageHeader *message)
1121 {
1122   struct PeerRequest *peerreq;
1123   struct GSF_PendingRequest *pr;
1124   struct GSF_PendingRequestData *prd;
1125   struct GSF_ConnectedPeer *cp;
1126   struct GSF_ConnectedPeer *cps;
1127   const GNUNET_HashCode *namespace;
1128   const struct GNUNET_PeerIdentity *target;
1129   enum GSF_PendingRequestOptions options;
1130   uint16_t msize;
1131   const struct GetMessage *gm;
1132   unsigned int bits;
1133   const GNUNET_HashCode *opt;
1134   uint32_t bm;
1135   size_t bfsize;
1136   uint32_t ttl_decrement;
1137   int32_t priority;
1138   int32_t ttl;
1139   enum GNUNET_BLOCK_Type type;
1140   GNUNET_PEER_Id spid;
1141
1142   GNUNET_assert (other != NULL);
1143   msize = ntohs (message->size);
1144   if (msize < sizeof (struct GetMessage))
1145   {
1146     GNUNET_break_op (0);
1147     return NULL;
1148   }
1149   GNUNET_STATISTICS_update (GSF_stats,
1150                             gettext_noop
1151                             ("# GET requests received (from other peers)"), 1,
1152                             GNUNET_NO);
1153   gm = (const struct GetMessage *) message;
1154   type = ntohl (gm->type);
1155   bm = ntohl (gm->hash_bitmap);
1156   bits = 0;
1157   while (bm > 0)
1158   {
1159     if (1 == (bm & 1))
1160       bits++;
1161     bm >>= 1;
1162   }
1163   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
1164   {
1165     GNUNET_break_op (0);
1166     return NULL;
1167   }
1168   opt = (const GNUNET_HashCode *) &gm[1];
1169   bfsize = msize - sizeof (struct GetMessage) - bits * sizeof (GNUNET_HashCode);
1170   /* bfsize must be power of 2, check! */
1171   if (0 != ((bfsize - 1) & bfsize))
1172   {
1173     GNUNET_break_op (0);
1174     return NULL;
1175   }
1176   GSF_cover_query_count++;
1177   bm = ntohl (gm->hash_bitmap);
1178   bits = 0;
1179   cps = GSF_peer_get_ (other);
1180   if (NULL == cps)
1181   {
1182     /* peer must have just disconnected */
1183     GNUNET_STATISTICS_update (GSF_stats,
1184                               gettext_noop
1185                               ("# requests dropped due to initiator not being connected"),
1186                               1, GNUNET_NO);
1187     return NULL;
1188   }
1189   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1190     cp = GSF_peer_get_ ((const struct GNUNET_PeerIdentity *) &opt[bits++]);
1191   else
1192     cp = cps;
1193   if (cp == NULL)
1194   {
1195 #if DEBUG_FS
1196     if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1197       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1198                   "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
1199                   GNUNET_i2s ((const struct GNUNET_PeerIdentity *)
1200                               &opt[bits - 1]));
1201
1202     else
1203       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1204                   "Failed to find peer `%4s' in connection set. Dropping query.\n",
1205                   GNUNET_i2s (other));
1206 #endif
1207     GNUNET_STATISTICS_update (GSF_stats,
1208                               gettext_noop
1209                               ("# requests dropped due to missing reverse route"),
1210                               1, GNUNET_NO);
1211     return NULL;
1212   }
1213   /* note that we can really only check load here since otherwise
1214    * peers could find out that we are overloaded by not being
1215    * disconnected after sending us a malformed query... */
1216   priority = bound_priority (ntohl (gm->priority), cps);
1217   if (priority < 0)
1218   {
1219 #if DEBUG_FS
1220     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1221                 "Dropping query from `%s', this peer is too busy.\n",
1222                 GNUNET_i2s (other));
1223 #endif
1224     return NULL;
1225   }
1226 #if DEBUG_FS
1227   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1228               "Received request for `%s' of type %u from peer `%4s' with flags %u\n",
1229               GNUNET_h2s (&gm->query), (unsigned int) type, GNUNET_i2s (other),
1230               (unsigned int) bm);
1231 #endif
1232   namespace = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) ? &opt[bits++] : NULL;
1233   if ((type == GNUNET_BLOCK_TYPE_FS_SBLOCK) && (namespace == NULL))
1234   {
1235     GNUNET_break_op (0);
1236     return NULL;
1237   }
1238   if ((type != GNUNET_BLOCK_TYPE_FS_SBLOCK) && (namespace != NULL))
1239   {
1240     GNUNET_break_op (0);
1241     return NULL;
1242   }
1243   target =
1244       (0 !=
1245        (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? ((const struct GNUNET_PeerIdentity
1246                                                *) &opt[bits++]) : NULL;
1247   options = GSF_PRO_DEFAULTS;
1248   spid = 0;
1249   if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1 + priority))
1250       || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) >
1251           GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value * 2 +
1252           GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)))
1253   {
1254     /* don't have BW to send to peer, or would likely take longer than we have for it,
1255      * so at best indirect the query */
1256     priority = 0;
1257     options |= GSF_PRO_FORWARD_ONLY;
1258     spid = GNUNET_PEER_intern (other);
1259     GNUNET_assert (0 != spid);
1260   }
1261   ttl = bound_ttl (ntohl (gm->ttl), priority);
1262   /* decrement ttl (always) */
1263   ttl_decrement =
1264       2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1265                                                     TTL_DECREMENT);
1266   if ((ttl < 0) && (((int32_t) (ttl - ttl_decrement)) > 0))
1267   {
1268 #if DEBUG_FS
1269     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1270                 "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
1271                 GNUNET_i2s (other), ttl, ttl_decrement);
1272 #endif
1273     GNUNET_STATISTICS_update (GSF_stats,
1274                               gettext_noop
1275                               ("# requests dropped due TTL underflow"), 1,
1276                               GNUNET_NO);
1277     /* integer underflow => drop (should be very rare)! */
1278     return NULL;
1279   }
1280   ttl -= ttl_decrement;
1281
1282   /* test if the request already exists */
1283   peerreq = GNUNET_CONTAINER_multihashmap_get (cp->request_map, &gm->query);
1284   if (peerreq != NULL)
1285   {
1286     pr = peerreq->pr;
1287     prd = GSF_pending_request_get_data_ (pr);
1288     if ((prd->type == type) &&
1289         ((type != GNUNET_BLOCK_TYPE_FS_SBLOCK) ||
1290          (0 == memcmp (&prd->namespace, namespace, sizeof (GNUNET_HashCode)))))
1291     {
1292       if (prd->ttl.abs_value >= GNUNET_TIME_absolute_get ().abs_value + ttl)
1293       {
1294         /* existing request has higher TTL, drop new one! */
1295         prd->priority += priority;
1296 #if DEBUG_FS
1297         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1298                     "Have existing request with higher TTL, dropping new request.\n",
1299                     GNUNET_i2s (other));
1300 #endif
1301         GNUNET_STATISTICS_update (GSF_stats,
1302                                   gettext_noop
1303                                   ("# requests dropped due to higher-TTL request"),
1304                                   1, GNUNET_NO);
1305         return NULL;
1306       }
1307       /* existing request has lower TTL, drop old one! */
1308       priority += prd->priority;
1309       GSF_pending_request_cancel_ (pr, GNUNET_YES);
1310       free_pending_request (peerreq, &gm->query);
1311     }
1312   }
1313
1314   peerreq = GNUNET_malloc (sizeof (struct PeerRequest));
1315   peerreq->cp = cp;
1316   pr = GSF_pending_request_create_ (options, type, &gm->query, namespace,
1317                                     target,
1318                                     (bfsize >
1319                                      0) ? (const char *) &opt[bits] : NULL,
1320                                     bfsize, ntohl (gm->filter_mutator),
1321                                     1 /* anonymity */ ,
1322                                     (uint32_t) priority, ttl, spid, GNUNET_PEER_intern (other), NULL, 0,        /* replies_seen */
1323                                     &handle_p2p_reply, peerreq);
1324   GNUNET_assert (NULL != pr);
1325   peerreq->pr = pr;
1326   GNUNET_break (GNUNET_OK ==
1327                 GNUNET_CONTAINER_multihashmap_put (cp->request_map, &gm->query,
1328                                                    peerreq,
1329                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
1330   GNUNET_STATISTICS_update (GSF_stats,
1331                             gettext_noop
1332                             ("# P2P query messages received and processed"), 1,
1333                             GNUNET_NO);
1334   GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"),
1335                             1, GNUNET_NO);
1336   return pr;
1337 }
1338
1339
1340 /**
1341  * Function called if there has been a timeout trying to satisfy
1342  * a transmission request.
1343  *
1344  * @param cls the 'struct GSF_PeerTransmitHandle' of the request
1345  * @param tc scheduler context
1346  */
1347 static void
1348 peer_transmit_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1349 {
1350   struct GSF_PeerTransmitHandle *pth = cls;
1351   struct GSF_ConnectedPeer *cp;
1352
1353 #if DEBUG_FS
1354   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1355               "Timeout trying to transmit to other peer\n");
1356 #endif
1357   pth->timeout_task = GNUNET_SCHEDULER_NO_TASK;
1358   cp = pth->cp;
1359   GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth);
1360   if (GNUNET_YES == pth->is_query)
1361     GNUNET_assert (0 < cp->ppd.pending_queries--);
1362   else if (GNUNET_NO == pth->is_query)
1363     GNUNET_assert (0 < cp->ppd.pending_replies--);
1364   GNUNET_LOAD_update (cp->ppd.transmission_delay, UINT64_MAX);
1365   if (NULL != pth->cth)
1366   {
1367     GNUNET_CORE_notify_transmit_ready_cancel (pth->cth);
1368     pth->cth = NULL;
1369   }
1370   pth->gmc (pth->gmc_cls, 0, NULL);
1371   GNUNET_assert (0 == pth->cth_in_progress);
1372   GNUNET_free (pth);
1373 }
1374
1375
1376 /**
1377  * Transmit a message to the given peer as soon as possible.
1378  * If the peer disconnects before the transmission can happen,
1379  * the callback is invoked with a 'NULL' buffer.
1380  *
1381  * @param cp target peer
1382  * @param is_query is this a query (GNUNET_YES) or content (GNUNET_NO) or neither (GNUNET_SYSERR)
1383  * @param priority how important is this request?
1384  * @param timeout when does this request timeout (call gmc with error)
1385  * @param size number of bytes we would like to send to the peer
1386  * @param gmc function to call to get the message
1387  * @param gmc_cls closure for gmc
1388  * @return handle to cancel request
1389  */
1390 struct GSF_PeerTransmitHandle *
1391 GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp, int is_query,
1392                     uint32_t priority, struct GNUNET_TIME_Relative timeout,
1393                     size_t size, GSF_GetMessageCallback gmc, void *gmc_cls)
1394 {
1395   struct GSF_PeerTransmitHandle *pth;
1396   struct GSF_PeerTransmitHandle *pos;
1397   struct GSF_PeerTransmitHandle *prev;
1398
1399   pth = GNUNET_malloc (sizeof (struct GSF_PeerTransmitHandle));
1400   pth->transmission_request_start_time = GNUNET_TIME_absolute_get ();
1401   pth->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1402   pth->gmc = gmc;
1403   pth->gmc_cls = gmc_cls;
1404   pth->size = size;
1405   pth->is_query = is_query;
1406   pth->priority = priority;
1407   pth->cp = cp;
1408   /* insertion sort (by priority, descending) */
1409   prev = NULL;
1410   pos = cp->pth_head;
1411   while ((pos != NULL) && (pos->priority > priority))
1412   {
1413     prev = pos;
1414     pos = pos->next;
1415   }
1416   if (prev == NULL)
1417     GNUNET_CONTAINER_DLL_insert (cp->pth_head, cp->pth_tail, pth);
1418   else
1419     GNUNET_CONTAINER_DLL_insert_after (cp->pth_head, cp->pth_tail, prev, pth);
1420   if (GNUNET_YES == is_query)
1421     cp->ppd.pending_queries++;
1422   else if (GNUNET_NO == is_query)
1423     cp->ppd.pending_replies++;
1424   pth->timeout_task =
1425       GNUNET_SCHEDULER_add_delayed (timeout, &peer_transmit_timeout, pth);
1426   schedule_transmission (pth);
1427   return pth;
1428 }
1429
1430
1431 /**
1432  * Cancel an earlier request for transmission.
1433  *
1434  * @param pth request to cancel
1435  */
1436 void
1437 GSF_peer_transmit_cancel_ (struct GSF_PeerTransmitHandle *pth)
1438 {
1439   struct GSF_ConnectedPeer *cp;
1440
1441   if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1442   {
1443     GNUNET_SCHEDULER_cancel (pth->timeout_task);
1444     pth->timeout_task = GNUNET_SCHEDULER_NO_TASK;
1445   }
1446   if (NULL != pth->cth)
1447   {
1448     GNUNET_CORE_notify_transmit_ready_cancel (pth->cth);
1449     pth->cth = NULL;
1450   }
1451   cp = pth->cp;
1452   GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth);
1453   if (GNUNET_YES == pth->is_query)
1454     GNUNET_assert (0 < cp->ppd.pending_queries--);
1455   else if (GNUNET_NO == pth->is_query)
1456     GNUNET_assert (0 < cp->ppd.pending_replies--);
1457   GNUNET_assert (0 == pth->cth_in_progress);
1458   GNUNET_free (pth);
1459 }
1460
1461
1462 /**
1463  * Report on receiving a reply; update the performance record of the given peer.
1464  *
1465  * @param cp responding peer (will be updated)
1466  * @param request_time time at which the original query was transmitted
1467  * @param request_priority priority of the original request
1468  */
1469 void
1470 GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp,
1471                               struct GNUNET_TIME_Absolute request_time,
1472                               uint32_t request_priority)
1473 {
1474   struct GNUNET_TIME_Relative delay;
1475
1476   delay = GNUNET_TIME_absolute_get_duration (request_time);
1477   cp->ppd.avg_reply_delay.rel_value =
1478       (cp->ppd.avg_reply_delay.rel_value * (RUNAVG_DELAY_N - 1) +
1479        delay.rel_value) / RUNAVG_DELAY_N;
1480   cp->ppd.avg_priority =
1481       (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1) +
1482        request_priority) / RUNAVG_DELAY_N;
1483 }
1484
1485
1486 /**
1487  * Report on receiving a reply in response to an initiating client.
1488  * Remember that this peer is good for this client.
1489  *
1490  * @param cp responding peer (will be updated)
1491  * @param initiator_client local client on responsible for query
1492  */
1493 void
1494 GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp,
1495                                    struct GSF_LocalClient *initiator_client)
1496 {
1497   cp->ppd.last_client_replies[cp->last_client_replies_woff++ %
1498                               CS2P_SUCCESS_LIST_SIZE] = initiator_client;
1499 }
1500
1501
1502 /**
1503  * Report on receiving a reply in response to an initiating peer.
1504  * Remember that this peer is good for this initiating peer.
1505  *
1506  * @param cp responding peer (will be updated)
1507  * @param initiator_peer other peer responsible for query
1508  */
1509 void
1510 GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp,
1511                                  const struct GSF_ConnectedPeer *initiator_peer)
1512 {
1513   unsigned int woff;
1514
1515   woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE;
1516   GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1);
1517   cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid;
1518   GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1);
1519   cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE;
1520 }
1521
1522
1523 /**
1524  * A peer disconnected from us.  Tear down the connected peer
1525  * record.
1526  *
1527  * @param cls unused
1528  * @param peer identity of peer that connected
1529  */
1530 void
1531 GSF_peer_disconnect_handler_ (void *cls, const struct GNUNET_PeerIdentity *peer)
1532 {
1533   struct GSF_ConnectedPeer *cp;
1534   struct GSF_PeerTransmitHandle *pth;
1535   struct GSF_DelayedHandle *dh;
1536
1537   cp = GSF_peer_get_ (peer);
1538   if (NULL == cp)
1539     return;                     /* must have been disconnect from core with
1540                                  * 'peer' == my_id, ignore */
1541   GNUNET_assert (GNUNET_YES ==
1542                  GNUNET_CONTAINER_multihashmap_remove (cp_map,
1543                                                        &peer->hashPubKey, cp));
1544   GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# peers connected"),
1545                          GNUNET_CONTAINER_multihashmap_size (cp_map),
1546                          GNUNET_NO);
1547   if (NULL != cp->migration_pth)
1548   {
1549     GSF_peer_transmit_cancel_ (cp->migration_pth);
1550     cp->migration_pth = NULL;
1551   }
1552   if (NULL != cp->rc)
1553   {
1554     GNUNET_ATS_reserve_bandwidth_cancel (cp->rc);
1555     cp->rc = NULL;
1556   }
1557   if (GNUNET_SCHEDULER_NO_TASK != cp->rc_delay_task)
1558   {
1559     GNUNET_SCHEDULER_cancel (cp->rc_delay_task);
1560     cp->rc_delay_task = GNUNET_SCHEDULER_NO_TASK;
1561   }
1562   GNUNET_CONTAINER_multihashmap_iterate (cp->request_map,
1563                                          &cancel_pending_request, cp);
1564   GNUNET_CONTAINER_multihashmap_destroy (cp->request_map);
1565   cp->request_map = NULL;
1566   GSF_plan_notify_peer_disconnect_ (cp);
1567   GNUNET_LOAD_value_free (cp->ppd.transmission_delay);
1568   GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
1569   memset (cp->ppd.last_p2p_replies, 0, sizeof (cp->ppd.last_p2p_replies));
1570   GSF_push_stop_ (cp);
1571   while (NULL != (pth = cp->pth_head))
1572   {
1573     if (NULL != pth->cth)
1574     {
1575       GNUNET_CORE_notify_transmit_ready_cancel (pth->cth);
1576       pth->cth = NULL;
1577     }
1578     if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1579     {
1580       GNUNET_SCHEDULER_cancel (pth->timeout_task);
1581       pth->timeout_task = GNUNET_SCHEDULER_NO_TASK;
1582     }
1583     GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth);
1584     GNUNET_assert (0 == pth->cth_in_progress);
1585     pth->gmc (pth->gmc_cls, 0, NULL);
1586     GNUNET_free (pth);
1587   }
1588   while (NULL != (dh = cp->delayed_head))
1589   {
1590     GNUNET_CONTAINER_DLL_remove (cp->delayed_head, cp->delayed_tail, dh);
1591     GNUNET_SCHEDULER_cancel (dh->delay_task);
1592     GNUNET_free (dh->pm);
1593     GNUNET_free (dh);
1594   }
1595   GNUNET_PEER_change_rc (cp->ppd.pid, -1);
1596   if (GNUNET_SCHEDULER_NO_TASK != cp->mig_revive_task)
1597   {
1598     GNUNET_SCHEDULER_cancel (cp->mig_revive_task);
1599     cp->mig_revive_task = GNUNET_SCHEDULER_NO_TASK;
1600   }
1601   GNUNET_free (cp);
1602 }
1603
1604
1605 /**
1606  * Closure for 'call_iterator'.
1607  */
1608 struct IterationContext
1609 {
1610   /**
1611    * Function to call on each entry.
1612    */
1613   GSF_ConnectedPeerIterator it;
1614
1615   /**
1616    * Closure for 'it'.
1617    */
1618   void *it_cls;
1619 };
1620
1621
1622 /**
1623  * Function that calls the callback for each peer.
1624  *
1625  * @param cls the 'struct IterationContext*'
1626  * @param key identity of the peer
1627  * @param value the 'struct GSF_ConnectedPeer*'
1628  * @return GNUNET_YES to continue iteration
1629  */
1630 static int
1631 call_iterator (void *cls, const GNUNET_HashCode * key, void *value)
1632 {
1633   struct IterationContext *ic = cls;
1634   struct GSF_ConnectedPeer *cp = value;
1635
1636   ic->it (ic->it_cls, (const struct GNUNET_PeerIdentity *) key, cp, &cp->ppd);
1637   return GNUNET_YES;
1638 }
1639
1640
1641 /**
1642  * Iterate over all connected peers.
1643  *
1644  * @param it function to call for each peer
1645  * @param it_cls closure for it
1646  */
1647 void
1648 GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it, void *it_cls)
1649 {
1650   struct IterationContext ic;
1651
1652   ic.it = it;
1653   ic.it_cls = it_cls;
1654   GNUNET_CONTAINER_multihashmap_iterate (cp_map, &call_iterator, &ic);
1655 }
1656
1657
1658 /**
1659  * Obtain the identity of a connected peer.
1660  *
1661  * @param cp peer to reserve bandwidth from
1662  * @param id identity to set (written to)
1663  */
1664 void
1665 GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp,
1666                                   struct GNUNET_PeerIdentity *id)
1667 {
1668   GNUNET_assert (0 != cp->ppd.pid);
1669   GNUNET_PEER_resolve (cp->ppd.pid, id);
1670 }
1671
1672
1673 /**
1674  * Assemble a migration stop message for transmission.
1675  *
1676  * @param cls the 'struct GSF_ConnectedPeer' to use
1677  * @param size number of bytes we're allowed to write to buf
1678  * @param buf where to copy the message
1679  * @return number of bytes copied to buf
1680  */
1681 static size_t
1682 create_migration_stop_message (void *cls, size_t size, void *buf)
1683 {
1684   struct GSF_ConnectedPeer *cp = cls;
1685   struct MigrationStopMessage msm;
1686
1687   cp->migration_pth = NULL;
1688   if (NULL == buf)
1689     return 0;
1690   GNUNET_assert (size >= sizeof (struct MigrationStopMessage));
1691   msm.header.size = htons (sizeof (struct MigrationStopMessage));
1692   msm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
1693   msm.reserved = htonl (0);
1694   msm.duration =
1695       GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
1696                                  (cp->last_migration_block));
1697   memcpy (buf, &msm, sizeof (struct MigrationStopMessage));
1698   GNUNET_STATISTICS_update (GSF_stats,
1699                             gettext_noop ("# migration stop messages sent"),
1700                             1, GNUNET_NO);
1701   return sizeof (struct MigrationStopMessage);
1702 }
1703
1704
1705 /**
1706  * Ask a peer to stop migrating data to us until the given point
1707  * in time.
1708  *
1709  * @param cp peer to ask
1710  * @param block_time until when to block
1711  */
1712 void
1713 GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp,
1714                            struct GNUNET_TIME_Absolute block_time)
1715 {
1716   if (cp->last_migration_block.abs_value > block_time.abs_value)
1717   {
1718 #if DEBUG_FS
1719     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1720                 "Migration already blocked for another %llu ms\n",
1721                 (unsigned long long)
1722                 GNUNET_TIME_absolute_get_remaining
1723                 (cp->last_migration_block).rel_value);
1724 #endif
1725     return;                     /* already blocked */
1726   }
1727 #if DEBUG_FS
1728   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %llu ms\n",
1729               (unsigned long long) GNUNET_TIME_absolute_get_remaining (block_time).rel_value);
1730 #endif
1731   cp->last_migration_block = block_time;
1732   if (cp->migration_pth != NULL)
1733     GSF_peer_transmit_cancel_ (cp->migration_pth);
1734   cp->migration_pth =
1735       GSF_peer_transmit_ (cp, GNUNET_SYSERR, UINT32_MAX,
1736                           GNUNET_TIME_UNIT_FOREVER_REL,
1737                           sizeof (struct MigrationStopMessage),
1738                           &create_migration_stop_message, cp);
1739 }
1740
1741
1742 /**
1743  * Write host-trust information to a file - flush the buffer entry!
1744  *
1745  * @param cls closure, not used
1746  * @param key host identity
1747  * @param value the 'struct GSF_ConnectedPeer' to flush
1748  * @return GNUNET_OK to continue iteration
1749  */
1750 static int
1751 flush_trust (void *cls, const GNUNET_HashCode * key, void *value)
1752 {
1753   struct GSF_ConnectedPeer *cp = value;
1754   char *fn;
1755   uint32_t trust;
1756   struct GNUNET_PeerIdentity pid;
1757
1758   if (cp->ppd.trust == cp->disk_trust)
1759     return GNUNET_OK;           /* unchanged */
1760   GNUNET_assert (0 != cp->ppd.pid);
1761   GNUNET_PEER_resolve (cp->ppd.pid, &pid);
1762   fn = get_trust_filename (&pid);
1763   if (cp->ppd.trust == 0)
1764   {
1765     if ((0 != UNLINK (fn)) && (errno != ENOENT))
1766       GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
1767                                 GNUNET_ERROR_TYPE_BULK, "unlink", fn);
1768   }
1769   else
1770   {
1771     trust = htonl (cp->ppd.trust);
1772     if (sizeof (uint32_t) ==
1773         GNUNET_DISK_fn_write (fn, &trust, sizeof (uint32_t),
1774                               GNUNET_DISK_PERM_USER_READ |
1775                               GNUNET_DISK_PERM_USER_WRITE |
1776                               GNUNET_DISK_PERM_GROUP_READ |
1777                               GNUNET_DISK_PERM_OTHER_READ))
1778       cp->disk_trust = cp->ppd.trust;
1779   }
1780   GNUNET_free (fn);
1781   return GNUNET_OK;
1782 }
1783
1784
1785 /**
1786  * Notify core about a preference we have for the given peer
1787  * (to allocate more resources towards it).  The change will
1788  * be communicated the next time we reserve bandwidth with
1789  * core (not instantly).
1790  *
1791  * @param cp peer to reserve bandwidth from
1792  * @param pref preference change
1793  */
1794 void
1795 GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp,
1796                                        uint64_t pref)
1797 {
1798   cp->inc_preference += pref;
1799 }
1800
1801
1802 /**
1803  * Call this method periodically to flush trust information to disk.
1804  *
1805  * @param cls closure, not used
1806  * @param tc task context, not used
1807  */
1808 static void
1809 cron_flush_trust (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1810 {
1811
1812   if (NULL == cp_map)
1813     return;
1814   GNUNET_CONTAINER_multihashmap_iterate (cp_map, &flush_trust, NULL);
1815   if (NULL == tc)
1816     return;
1817   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1818     return;
1819   GNUNET_SCHEDULER_add_delayed (TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
1820 }
1821
1822
1823 /**
1824  * Initialize peer management subsystem.
1825  */
1826 void
1827 GSF_connected_peer_init_ ()
1828 {
1829   cp_map = GNUNET_CONTAINER_multihashmap_create (128);
1830   ats = GNUNET_ATS_performance_init (GSF_cfg, NULL, NULL);
1831   GNUNET_assert (GNUNET_OK ==
1832                  GNUNET_CONFIGURATION_get_value_filename (GSF_cfg, "fs",
1833                                                           "TRUST",
1834                                                           &trustDirectory));
1835   GNUNET_DISK_directory_create (trustDirectory);
1836   GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
1837                                       &cron_flush_trust, NULL);
1838 }
1839
1840
1841 /**
1842  * Iterator to free peer entries.
1843  *
1844  * @param cls closure, unused
1845  * @param key current key code
1846  * @param value value in the hash map (peer entry)
1847  * @return GNUNET_YES (we should continue to iterate)
1848  */
1849 static int
1850 clean_peer (void *cls, const GNUNET_HashCode * key, void *value)
1851 {
1852   GSF_peer_disconnect_handler_ (NULL, (const struct GNUNET_PeerIdentity *) key);
1853   return GNUNET_YES;
1854 }
1855
1856
1857 /**
1858  * Shutdown peer management subsystem.
1859  */
1860 void
1861 GSF_connected_peer_done_ ()
1862 {
1863   cron_flush_trust (NULL, NULL);
1864   GNUNET_CONTAINER_multihashmap_iterate (cp_map, &clean_peer, NULL);
1865   GNUNET_CONTAINER_multihashmap_destroy (cp_map);
1866   cp_map = NULL;
1867   GNUNET_free (trustDirectory);
1868   trustDirectory = NULL;
1869   GNUNET_ATS_performance_done (ats);
1870   ats = NULL;
1871 }
1872
1873
1874 /**
1875  * Iterator to remove references to LC entry.
1876  *
1877  * @param cls the 'struct GSF_LocalClient*' to look for
1878  * @param key current key code
1879  * @param value value in the hash map (peer entry)
1880  * @return GNUNET_YES (we should continue to iterate)
1881  */
1882 static int
1883 clean_local_client (void *cls, const GNUNET_HashCode * key, void *value)
1884 {
1885   const struct GSF_LocalClient *lc = cls;
1886   struct GSF_ConnectedPeer *cp = value;
1887   unsigned int i;
1888
1889   for (i = 0; i < CS2P_SUCCESS_LIST_SIZE; i++)
1890     if (cp->ppd.last_client_replies[i] == lc)
1891       cp->ppd.last_client_replies[i] = NULL;
1892   return GNUNET_YES;
1893 }
1894
1895
1896 /**
1897  * Notification that a local client disconnected.  Clean up all of our
1898  * references to the given handle.
1899  *
1900  * @param lc handle to the local client (henceforth invalid)
1901  */
1902 void
1903 GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc)
1904 {
1905   if (NULL == cp_map)
1906     return;                     /* already cleaned up */
1907   GNUNET_CONTAINER_multihashmap_iterate (cp_map, &clean_local_client,
1908                                          (void *) lc);
1909 }
1910
1911
1912 /* end of gnunet-service-fs_cp.c */