src/fs/gnunet-service-fs_cp.c

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