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., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, 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", peer, "respect",
 662                                 GNUNET_TIME_UNIT_FOREVER_REL, &peer_respect_cb,
 663                                 cp);
 664 }
 665
 666
 667 /**
 668  * It may be time to re-start migrating content to this
 669  * peer.  Check, and if so, restart migration.
 670  *
 671  * @param cls the `struct GSF_ConnectedPeer`
 672  * @param tc scheduler context
 673  */
 674 static void
 675 revive_migration (void *cls,
 676                   const struct GNUNET_SCHEDULER_TaskContext *tc)
 677 {
 678   struct GSF_ConnectedPeer *cp = cls;
 679   struct GNUNET_TIME_Relative bt;
 680
 681   cp->mig_revive_task = NULL;
 682   bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until);
 683   if (0 != bt.rel_value_us)
 684   {
 685     /* still time left... */
 686     cp->mig_revive_task =
 687         GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
 688     return;
 689   }
 690   GSF_push_start_ (cp);
 691 }
 692
 693
 694 /**
 695  * Get a handle for a connected peer.
 696  *
 697  * @param peer peer's identity
 698  * @return NULL if the peer is not currently connected
 699  */
 700 struct GSF_ConnectedPeer *
 701 GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer)
 702 {
 703   if (NULL == cp_map)
 704     return NULL;
 705   return GNUNET_CONTAINER_multipeermap_get (cp_map, peer);
 706 }
 707
 708
 709 /**
 710  * Handle P2P "MIGRATION_STOP" message.
 711  *
 712  * @param cls closure, always NULL
 713  * @param other the other peer involved (sender or receiver, NULL
 714  *        for loopback messages where we are both sender and receiver)
 715  * @param message the actual message
 716  * @return #GNUNET_OK to keep the connection open,
 717  *         #GNUNET_SYSERR to close it (signal serious error)
 718  */
 719 int
 720 GSF_handle_p2p_migration_stop_ (void *cls,
 721                                 const struct GNUNET_PeerIdentity *other,
 722                                 const struct GNUNET_MessageHeader *message)
 723 {
 724   struct GSF_ConnectedPeer *cp;
 725   const struct MigrationStopMessage *msm;
 726   struct GNUNET_TIME_Relative bt;
 727
 728   msm = (const struct MigrationStopMessage *) message;
 729   cp = GSF_peer_get_ (other);
 730   if (NULL == cp)
 731   {
 732     GNUNET_break (0);
 733     return GNUNET_OK;
 734   }
 735   GNUNET_STATISTICS_update (GSF_stats,
 736                             gettext_noop ("# migration stop messages received"),
 737                             1, GNUNET_NO);
 738   bt = GNUNET_TIME_relative_ntoh (msm->duration);
 739   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 740               _("Migration of content to peer `%s' blocked for %s\n"),
 741               GNUNET_i2s (other),
 742               GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES));
 743   cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt);
 744   if ( (NULL == cp->mig_revive_task) &&
 745        (NULL == cp->respect_iterate_req) )
 746   {
 747     GSF_push_stop_ (cp);
 748     cp->mig_revive_task =
 749         GNUNET_SCHEDULER_add_delayed (bt,
 750                                       &revive_migration, cp);
 751   }
 752   return GNUNET_OK;
 753 }
 754
 755
 756 /**
 757  * Copy reply and free put message.
 758  *
 759  * @param cls the `struct PutMessage`
 760  * @param buf_size number of bytes available in @a buf
 761  * @param buf where to copy the message, NULL on error (peer disconnect)
 762  * @return number of bytes copied to @a buf, can be 0 (without indicating an error)
 763  */
 764 static size_t
 765 copy_reply (void *cls,
 766             size_t buf_size,
 767             void *buf)
 768 {
 769   struct PutMessage *pm = cls;
 770   size_t size;
 771
 772   if (NULL != buf)
 773   {
 774     GNUNET_assert (buf_size >= ntohs (pm->header.size));
 775     size = ntohs (pm->header.size);
 776     memcpy (buf, pm, size);
 777     GNUNET_STATISTICS_update (GSF_stats,
 778                               gettext_noop ("# replies transmitted to other peers"),
 779                               1,
 780                               GNUNET_NO);
 781   }
 782   else
 783   {
 784     size = 0;
 785     GNUNET_STATISTICS_update (GSF_stats,
 786                               gettext_noop ("# replies dropped"),
 787                               1,
 788                               GNUNET_NO);
 789   }
 790   GNUNET_free (pm);
 791   return size;
 792 }
 793
 794
 795 /**
 796  * Free resources associated with the given peer request.
 797  *
 798  * @param peerreq request to free
 799  * @param query associated key for the request
 800  */
 801 static void
 802 free_pending_request (struct PeerRequest *peerreq,
 803                       const struct GNUNET_HashCode *query)
 804 {
 805   struct GSF_ConnectedPeer *cp = peerreq->cp;
 806
 807   if (NULL != peerreq->kill_task)
 808   {
 809     GNUNET_SCHEDULER_cancel (peerreq->kill_task);
 810     peerreq->kill_task = NULL;
 811   }
 812   GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"),
 813                             -1, GNUNET_NO);
 814   GNUNET_break (GNUNET_YES ==
 815                 GNUNET_CONTAINER_multihashmap_remove (cp->request_map,
 816                                                       query, peerreq));
 817   GNUNET_free (peerreq);
 818 }
 819
 820
 821 /**
 822  * Cancel all requests associated with the peer.
 823  *
 824  * @param cls unused
 825  * @param query hash code of the request
 826  * @param value the `struct GSF_PendingRequest`
 827  * @return #GNUNET_YES (continue to iterate)
 828  */
 829 static int
 830 cancel_pending_request (void *cls,
 831                         const struct GNUNET_HashCode *query,
 832                         void *value)
 833 {
 834   struct PeerRequest *peerreq = value;
 835   struct GSF_PendingRequest *pr = peerreq->pr;
 836   struct GSF_PendingRequestData *prd;
 837
 838   prd = GSF_pending_request_get_data_ (pr);
 839   GSF_pending_request_cancel_ (pr, GNUNET_NO);
 840   free_pending_request (peerreq, &prd->query);
 841   return GNUNET_OK;
 842 }
 843
 844
 845 /**
 846  * Free the given request.
 847  *
 848  * @param cls the request to free
 849  * @param tc task context
 850  */
 851 static void
 852 peer_request_destroy (void *cls,
 853                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 854 {
 855   struct PeerRequest *peerreq = cls;
 856   struct GSF_PendingRequest *pr = peerreq->pr;
 857   struct GSF_PendingRequestData *prd;
 858
 859   peerreq->kill_task = NULL;
 860   prd = GSF_pending_request_get_data_ (pr);
 861   cancel_pending_request (NULL, &prd->query, peerreq);
 862 }
 863
 864
 865 /**
 866  * The artificial delay is over, transmit the message now.
 867  *
 868  * @param cls the `struct GSF_DelayedHandle` with the message
 869  * @param tc scheduler context
 870  */
 871 static void
 872 transmit_delayed_now (void *cls,
 873                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 874 {
 875   struct GSF_DelayedHandle *dh = cls;
 876   struct GSF_ConnectedPeer *cp = dh->cp;
 877
 878   GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
 879                                cp->delayed_tail,
 880                                dh);
 881   cp->delay_queue_size--;
 882   if (0 != (GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason))
 883   {
 884     GNUNET_free (dh->pm);
 885     GNUNET_free (dh);
 886     return;
 887   }
 888   (void) GSF_peer_transmit_ (cp,
 889                              GNUNET_NO,
 890                              UINT32_MAX,
 891                              REPLY_TIMEOUT,
 892                              dh->msize,
 893                              &copy_reply,
 894                              dh->pm);
 895   GNUNET_free (dh);
 896 }
 897
 898
 899 /**
 900  * Get the randomized delay a response should be subjected to.
 901  *
 902  * @return desired delay
 903  */
 904 static struct GNUNET_TIME_Relative
 905 get_randomized_delay ()
 906 {
 907   struct GNUNET_TIME_Relative ret;
 908
 909   ret =
 910       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
 911                                      GNUNET_CRYPTO_random_u32
 912                                      (GNUNET_CRYPTO_QUALITY_WEAK,
 913                                       2 * GSF_avg_latency.rel_value_us + 1));
 914 #if INSANE_STATISTICS
 915   GNUNET_STATISTICS_update (GSF_stats,
 916                             gettext_noop
 917                             ("# artificial delays introduced (ms)"),
 918                             ret.rel_value_us / 1000LL, GNUNET_NO);
 919 #endif
 920   return ret;
 921 }
 922
 923
 924 /**
 925  * Handle a reply to a pending request.  Also called if a request
 926  * expires (then with data == NULL).  The handler may be called
 927  * many times (depending on the request type), but will not be
 928  * called during or after a call to GSF_pending_request_cancel
 929  * and will also not be called anymore after a call signalling
 930  * expiration.
 931  *
 932  * @param cls `struct PeerRequest` this is an answer for
 933  * @param eval evaluation of the result
 934  * @param pr handle to the original pending request
 935  * @param reply_anonymity_level anonymity level for the reply, UINT32_MAX for "unknown"
 936  * @param expiration when does @a data expire?
 937  * @param last_transmission when did we last transmit a request for this block
 938  * @param type type of the block
 939  * @param data response data, NULL on request expiration
 940  * @param data_len number of bytes in @a data
 941  */
 942 static void
 943 handle_p2p_reply (void *cls,
 944                   enum GNUNET_BLOCK_EvaluationResult eval,
 945                   struct GSF_PendingRequest *pr,
 946                   uint32_t reply_anonymity_level,
 947                   struct GNUNET_TIME_Absolute expiration,
 948                   struct GNUNET_TIME_Absolute last_transmission,
 949                   enum GNUNET_BLOCK_Type type,
 950                   const void *data,
 951                   size_t data_len)
 952 {
 953   struct PeerRequest *peerreq = cls;
 954   struct GSF_ConnectedPeer *cp = peerreq->cp;
 955   struct GSF_PendingRequestData *prd;
 956   struct PutMessage *pm;
 957   size_t msize;
 958
 959   GNUNET_assert (data_len + sizeof (struct PutMessage) <
 960                  GNUNET_SERVER_MAX_MESSAGE_SIZE);
 961   GNUNET_assert (peerreq->pr == pr);
 962   prd = GSF_pending_request_get_data_ (pr);
 963   if (NULL == data)
 964   {
 965     free_pending_request (peerreq, &prd->query);
 966     return;
 967   }
 968   GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type);
 969   if ((prd->type != type) && (GNUNET_BLOCK_TYPE_ANY != prd->type))
 970   {
 971     GNUNET_STATISTICS_update (GSF_stats,
 972                               gettext_noop
 973                               ("# replies dropped due to type mismatch"),
 974                                 1, GNUNET_NO);
 975     return;
 976   }
 977   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 978               "Transmitting result for query `%s' to peer\n",
 979               GNUNET_h2s (&prd->query));
 980   GNUNET_STATISTICS_update (GSF_stats,
 981                             gettext_noop ("# replies received for other peers"),
 982                             1, GNUNET_NO);
 983   msize = sizeof (struct PutMessage) + data_len;
 984   if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 985   {
 986     GNUNET_break (0);
 987     return;
 988   }
 989   if ((UINT32_MAX != reply_anonymity_level) && (reply_anonymity_level > 1))
 990   {
 991     if (reply_anonymity_level - 1 > GSF_cover_content_count)
 992     {
 993       GNUNET_STATISTICS_update (GSF_stats,
 994                                 gettext_noop
 995                                 ("# replies dropped due to insufficient cover traffic"),
 996                                 1, GNUNET_NO);
 997       return;
 998     }
 999     GSF_cover_content_count -= (reply_anonymity_level - 1);
1000   }
1001
1002   pm = GNUNET_malloc (msize);
1003   pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
1004   pm->header.size = htons (msize);
1005   pm->type = htonl (type);
1006   pm->expiration = GNUNET_TIME_absolute_hton (expiration);
1007   memcpy (&pm[1], data, data_len);
1008   if ( (UINT32_MAX != reply_anonymity_level) &&
1009        (0 != reply_anonymity_level) &&
1010        (GNUNET_YES == GSF_enable_randomized_delays) )
1011   {
1012     struct GSF_DelayedHandle *dh;
1013
1014     dh = GNUNET_new (struct GSF_DelayedHandle);
1015     dh->cp = cp;
1016     dh->pm = pm;
1017     dh->msize = msize;
1018     GNUNET_CONTAINER_DLL_insert (cp->delayed_head,
1019                                  cp->delayed_tail,
1020                                  dh);
1021     cp->delay_queue_size++;
1022     dh->delay_task =
1023         GNUNET_SCHEDULER_add_delayed (get_randomized_delay (),
1024                                       &transmit_delayed_now,
1025                                       dh);
1026   }
1027   else
1028   {
1029     (void) GSF_peer_transmit_ (cp,
1030                                GNUNET_NO,
1031                                UINT32_MAX,
1032                                REPLY_TIMEOUT,
1033                                msize,
1034                                &copy_reply,
1035                                pm);
1036   }
1037   if (GNUNET_BLOCK_EVALUATION_OK_LAST != eval)
1038     return;
1039   if (NULL == peerreq->kill_task)
1040   {
1041     GNUNET_STATISTICS_update (GSF_stats,
1042                               gettext_noop
1043                               ("# P2P searches destroyed due to ultimate reply"),
1044                               1, GNUNET_NO);
1045     peerreq->kill_task =
1046         GNUNET_SCHEDULER_add_now (&peer_request_destroy, peerreq);
1047   }
1048 }
1049
1050
1051 /**
1052  * Increase the peer's respect by a value.
1053  *
1054  * @param cp which peer to change the respect value on
1055  * @param value is the int value by which the
1056  *  peer's credit is to be increased or decreased
1057  * @returns the actual change in respect (positive or negative)
1058  */
1059 static int
1060 change_peer_respect (struct GSF_ConnectedPeer *cp, int value)
1061 {
1062   if (0 == value)
1063     return 0;
1064   GNUNET_assert (NULL != cp);
1065   if (value > 0)
1066   {
1067     if (cp->ppd.respect + value < cp->ppd.respect)
1068     {
1069       value = UINT32_MAX - cp->ppd.respect;
1070       cp->ppd.respect = UINT32_MAX;
1071     }
1072     else
1073       cp->ppd.respect += value;
1074   }
1075   else
1076   {
1077     if (cp->ppd.respect < -value)
1078     {
1079       value = -cp->ppd.respect;
1080       cp->ppd.respect = 0;
1081     }
1082     else
1083       cp->ppd.respect += value;
1084   }
1085   return value;
1086 }
1087
1088
1089 /**
1090  * We've received a request with the specified priority.  Bound it
1091  * according to how much we respect the given peer.
1092  *
1093  * @param prio_in requested priority
1094  * @param cp the peer making the request
1095  * @return effective priority
1096  */
1097 static int32_t
1098 bound_priority (uint32_t prio_in, struct GSF_ConnectedPeer *cp)
1099 {
1100 #define N ((double)128.0)
1101   uint32_t ret;
1102   double rret;
1103   int ld;
1104
1105   ld = GSF_test_get_load_too_high_ (0);
1106   if (GNUNET_SYSERR == ld)
1107   {
1108 #if INSANE_STATISTICS
1109     GNUNET_STATISTICS_update (GSF_stats,
1110                               gettext_noop
1111                               ("# requests done for free (low load)"), 1,
1112                               GNUNET_NO);
1113 #endif
1114     return 0;                   /* excess resources */
1115   }
1116   if (prio_in > INT32_MAX)
1117     prio_in = INT32_MAX;
1118   ret = -change_peer_respect (cp, -(int) prio_in);
1119   if (ret > 0)
1120   {
1121     if (ret > GSF_current_priorities + N)
1122       rret = GSF_current_priorities + N;
1123     else
1124       rret = ret;
1125     GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N;
1126   }
1127   if ((GNUNET_YES == ld) && (ret > 0))
1128   {
1129     /* try with charging */
1130     ld = GSF_test_get_load_too_high_ (ret);
1131   }
1132   if (GNUNET_YES == ld)
1133   {
1134     GNUNET_STATISTICS_update (GSF_stats,
1135                               gettext_noop
1136                               ("# request dropped, priority insufficient"), 1,
1137                               GNUNET_NO);
1138     /* undo charge */
1139     change_peer_respect (cp, (int) ret);
1140     return -1;                  /* not enough resources */
1141   }
1142   else
1143   {
1144     GNUNET_STATISTICS_update (GSF_stats,
1145                               gettext_noop
1146                               ("# requests done for a price (normal load)"), 1,
1147                               GNUNET_NO);
1148   }
1149 #undef N
1150   return ret;
1151 }
1152
1153
1154 /**
1155  * The priority level imposes a bound on the maximum
1156  * value for the ttl that can be requested.
1157  *
1158  * @param ttl_in requested ttl
1159  * @param prio given priority
1160  * @return ttl_in if ttl_in is below the limit,
1161  *         otherwise the ttl-limit for the given priority
1162  */
1163 static int32_t
1164 bound_ttl (int32_t ttl_in, uint32_t prio)
1165 {
1166   unsigned long long allowed;
1167
1168   if (ttl_in <= 0)
1169     return ttl_in;
1170   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
1171   if (ttl_in > allowed)
1172   {
1173     if (allowed >= (1 << 30))
1174       return 1 << 30;
1175     return allowed;
1176   }
1177   return ttl_in;
1178 }
1179
1180
1181 /**
1182  * Handle P2P "QUERY" message.  Creates the pending request entry
1183  * and sets up all of the data structures to that we will
1184  * process replies properly.  Does not initiate forwarding or
1185  * local database lookups.
1186  *
1187  * @param other the other peer involved (sender or receiver, NULL
1188  *        for loopback messages where we are both sender and receiver)
1189  * @param message the actual message
1190  * @return pending request handle, NULL on error
1191  */
1192 struct GSF_PendingRequest *
1193 GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other,
1194                        const struct GNUNET_MessageHeader *message)
1195 {
1196   struct PeerRequest *peerreq;
1197   struct GSF_PendingRequest *pr;
1198   struct GSF_PendingRequestData *prd;
1199   struct GSF_ConnectedPeer *cp;
1200   struct GSF_ConnectedPeer *cps;
1201   const struct GNUNET_PeerIdentity *target;
1202   enum GSF_PendingRequestOptions options;
1203   uint16_t msize;
1204   const struct GetMessage *gm;
1205   unsigned int bits;
1206   const struct GNUNET_PeerIdentity *opt;
1207   uint32_t bm;
1208   size_t bfsize;
1209   uint32_t ttl_decrement;
1210   int32_t priority;
1211   int32_t ttl;
1212   enum GNUNET_BLOCK_Type type;
1213   GNUNET_PEER_Id spid;
1214
1215   msize = ntohs (message->size);
1216   if (msize < sizeof (struct GetMessage))
1217   {
1218     GNUNET_break_op (0);
1219     return NULL;
1220   }
1221   GNUNET_STATISTICS_update (GSF_stats,
1222                             gettext_noop
1223                             ("# GET requests received (from other peers)"),
1224                             1,
1225                             GNUNET_NO);
1226   gm = (const struct GetMessage *) message;
1227   type = ntohl (gm->type);
1228   bm = ntohl (gm->hash_bitmap);
1229   bits = 0;
1230   while (bm > 0)
1231   {
1232     if (1 == (bm & 1))
1233       bits++;
1234     bm >>= 1;
1235   }
1236   if (msize < sizeof (struct GetMessage) + bits * sizeof (struct GNUNET_PeerIdentity))
1237   {
1238     GNUNET_break_op (0);
1239     return NULL;
1240   }
1241   opt = (const struct GNUNET_PeerIdentity *) &gm[1];
1242   bfsize = msize - sizeof (struct GetMessage) - bits * sizeof (struct GNUNET_PeerIdentity);
1243   /* bfsize must be power of 2, check! */
1244   if (0 != ((bfsize - 1) & bfsize))
1245   {
1246     GNUNET_break_op (0);
1247     return NULL;
1248   }
1249   GSF_cover_query_count++;
1250   bm = ntohl (gm->hash_bitmap);
1251   bits = 0;
1252   cps = GSF_peer_get_ (other);
1253   if (NULL == cps)
1254   {
1255     /* peer must have just disconnected */
1256     GNUNET_STATISTICS_update (GSF_stats,
1257                               gettext_noop
1258                               ("# requests dropped due to initiator not being connected"),
1259                               1, GNUNET_NO);
1260     return NULL;
1261   }
1262   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1263     cp = GSF_peer_get_ (&opt[bits++]);
1264   else
1265     cp = cps;
1266   if (NULL == cp)
1267   {
1268     if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1269       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1270                   "Failed to find RETURN-TO peer `%s' in connection set. Dropping query.\n",
1271                   GNUNET_i2s (&opt[bits - 1]));
1272
1273     else
1274       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1275                   "Failed to find peer `%s' in connection set. Dropping query.\n",
1276                   GNUNET_i2s (other));
1277     GNUNET_STATISTICS_update (GSF_stats,
1278                               gettext_noop
1279                               ("# requests dropped due to missing reverse route"),
1280                               1,
1281                               GNUNET_NO);
1282     return NULL;
1283   }
1284   if (cp->ppd.pending_replies + cp->delay_queue_size > MAX_QUEUE_PER_PEER)
1285   {
1286     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1287                 "Peer `%s' has too many replies queued already. Dropping query.\n",
1288                 GNUNET_i2s (other));
1289     GNUNET_STATISTICS_update (GSF_stats,
1290                               gettext_noop ("# requests dropped due to full reply queue"),
1291                               1,
1292                               GNUNET_NO);
1293     return NULL;
1294   }
1295   /* note that we can really only check load here since otherwise
1296    * peers could find out that we are overloaded by not being
1297    * disconnected after sending us a malformed query... */
1298   priority = bound_priority (ntohl (gm->priority),
1299                              cps);
1300   if (priority < 0)
1301   {
1302     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1303                 "Dropping query from `%s', this peer is too busy.\n",
1304                 GNUNET_i2s (other));
1305     return NULL;
1306   }
1307   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1308               "Received request for `%s' of type %u from peer `%s' with flags %u\n",
1309               GNUNET_h2s (&gm->query),
1310               (unsigned int) type,
1311               GNUNET_i2s (other),
1312               (unsigned int) bm);
1313   target =
1314       (0 !=
1315        (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? (&opt[bits++]) : NULL;
1316   options = GSF_PRO_DEFAULTS;
1317   spid = 0;
1318   if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1 + priority))
1319       || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) >
1320           GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value_us * 2 +
1321           GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)))
1322   {
1323     /* don't have BW to send to peer, or would likely take longer than we have for it,
1324      * so at best indirect the query */
1325     priority = 0;
1326     options |= GSF_PRO_FORWARD_ONLY;
1327     spid = GNUNET_PEER_intern (other);
1328     GNUNET_assert (0 != spid);
1329   }
1330   ttl = bound_ttl (ntohl (gm->ttl), priority);
1331   /* decrement ttl (always) */
1332   ttl_decrement =
1333       2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1334                                                     TTL_DECREMENT);
1335   if ((ttl < 0) && (((int32_t) (ttl - ttl_decrement)) > 0))
1336   {
1337     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1338                 "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
1339                 GNUNET_i2s (other), ttl, ttl_decrement);
1340     GNUNET_STATISTICS_update (GSF_stats,
1341                               gettext_noop
1342                               ("# requests dropped due TTL underflow"), 1,
1343                               GNUNET_NO);
1344     /* integer underflow => drop (should be very rare)! */
1345     return NULL;
1346   }
1347   ttl -= ttl_decrement;
1348
1349   /* test if the request already exists */
1350   peerreq = GNUNET_CONTAINER_multihashmap_get (cp->request_map,
1351                                                &gm->query);
1352   if (NULL != peerreq)
1353   {
1354     pr = peerreq->pr;
1355     prd = GSF_pending_request_get_data_ (pr);
1356     if (prd->type == type)
1357     {
1358       if (prd->ttl.abs_value_us >= GNUNET_TIME_absolute_get ().abs_value_us + ttl * 1000LL)
1359       {
1360         /* existing request has higher TTL, drop new one! */
1361         prd->priority += priority;
1362         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1363                     "Have existing request with higher TTL, dropping new request.\n",
1364                     GNUNET_i2s (other));
1365         GNUNET_STATISTICS_update (GSF_stats,
1366                                   gettext_noop
1367                                   ("# requests dropped due to higher-TTL request"),
1368                                   1, GNUNET_NO);
1369         return NULL;
1370       }
1371       /* existing request has lower TTL, drop old one! */
1372       priority += prd->priority;
1373       GSF_pending_request_cancel_ (pr, GNUNET_YES);
1374       free_pending_request (peerreq, &gm->query);
1375     }
1376   }
1377
1378   peerreq = GNUNET_new (struct PeerRequest);
1379   peerreq->cp = cp;
1380   pr = GSF_pending_request_create_ (options,
1381                                     type,
1382                                     &gm->query,
1383                                     target,
1384                                     (bfsize > 0)
1385                                     ? (const char *) &opt[bits]
1386                                     : NULL,
1387                                     bfsize,
1388                                     ntohl (gm->filter_mutator),
1389                                     1 /* anonymity */,
1390                                     (uint32_t) priority,
1391                                     ttl,
1392                                     spid,
1393                                     GNUNET_PEER_intern (other),
1394                                     NULL, 0,        /* replies_seen */
1395                                     &handle_p2p_reply, peerreq);
1396   GNUNET_assert (NULL != pr);
1397   peerreq->pr = pr;
1398   GNUNET_break (GNUNET_OK ==
1399                 GNUNET_CONTAINER_multihashmap_put (cp->request_map, &gm->query,
1400                                                    peerreq,
1401                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
1402   GNUNET_STATISTICS_update (GSF_stats,
1403                             gettext_noop
1404                             ("# P2P query messages received and processed"), 1,
1405                             GNUNET_NO);
1406   GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"),
1407                             1, GNUNET_NO);
1408   return pr;
1409 }
1410
1411
1412 /**
1413  * Function called if there has been a timeout trying to satisfy
1414  * a transmission request.
1415  *
1416  * @param cls the `struct GSF_PeerTransmitHandle` of the request
1417  * @param tc scheduler context
1418  */
1419 static void
1420 peer_transmit_timeout (void *cls,
1421                        const struct GNUNET_SCHEDULER_TaskContext *tc)
1422 {
1423   struct GSF_PeerTransmitHandle *pth = cls;
1424   struct GSF_ConnectedPeer *cp;
1425
1426   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1427               "Timeout trying to transmit to other peer\n");
1428   pth->timeout_task = NULL;
1429   cp = pth->cp;
1430   GNUNET_CONTAINER_DLL_remove (cp->pth_head,
1431                                cp->pth_tail,
1432                                pth);
1433   if (GNUNET_YES == pth->is_query)
1434     GNUNET_assert (0 < cp->ppd.pending_queries--);
1435   else if (GNUNET_NO == pth->is_query)
1436     GNUNET_assert (0 < cp->ppd.pending_replies--);
1437   GNUNET_LOAD_update (cp->ppd.transmission_delay, UINT64_MAX);
1438   if (NULL != cp->cth)
1439   {
1440     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1441     cp->cth = NULL;
1442   }
1443   pth->gmc (pth->gmc_cls, 0, NULL);
1444   GNUNET_assert (0 == cp->cth_in_progress);
1445   GNUNET_free (pth);
1446 }
1447
1448
1449 /**
1450  * Transmit a message to the given peer as soon as possible.
1451  * If the peer disconnects before the transmission can happen,
1452  * the callback is invoked with a `NULL` @a buffer.
1453  *
1454  * @param cp target peer
1455  * @param is_query is this a query (#GNUNET_YES) or content (#GNUNET_NO) or neither (#GNUNET_SYSERR)
1456  * @param priority how important is this request?
1457  * @param timeout when does this request timeout (call gmc with error)
1458  * @param size number of bytes we would like to send to the peer
1459  * @param gmc function to call to get the message
1460  * @param gmc_cls closure for @a gmc
1461  * @return handle to cancel request
1462  */
1463 struct GSF_PeerTransmitHandle *
1464 GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp,
1465                     int is_query,
1466                     uint32_t priority,
1467                     struct GNUNET_TIME_Relative timeout,
1468                     size_t size,
1469                     GSF_GetMessageCallback gmc, void *gmc_cls)
1470 {
1471   struct GSF_PeerTransmitHandle *pth;
1472   struct GSF_PeerTransmitHandle *pos;
1473   struct GSF_PeerTransmitHandle *prev;
1474
1475   pth = GNUNET_new (struct GSF_PeerTransmitHandle);
1476   pth->transmission_request_start_time = GNUNET_TIME_absolute_get ();
1477   pth->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1478   pth->gmc = gmc;
1479   pth->gmc_cls = gmc_cls;
1480   pth->size = size;
1481   pth->is_query = is_query;
1482   pth->priority = priority;
1483   pth->cp = cp;
1484   /* insertion sort (by priority, descending) */
1485   prev = NULL;
1486   pos = cp->pth_head;
1487   while ((NULL != pos) && (pos->priority > priority))
1488   {
1489     prev = pos;
1490     pos = pos->next;
1491   }
1492   GNUNET_CONTAINER_DLL_insert_after (cp->pth_head,
1493                                      cp->pth_tail,
1494                                      prev,
1495                                      pth);
1496   if (GNUNET_YES == is_query)
1497     cp->ppd.pending_queries++;
1498   else if (GNUNET_NO == is_query)
1499     cp->ppd.pending_replies++;
1500   pth->timeout_task
1501     = GNUNET_SCHEDULER_add_delayed (timeout,
1502                                     &peer_transmit_timeout,
1503                                     pth);
1504   schedule_transmission (pth);
1505   return pth;
1506 }
1507
1508
1509 /**
1510  * Cancel an earlier request for transmission.
1511  *
1512  * @param pth request to cancel
1513  */
1514 void
1515 GSF_peer_transmit_cancel_ (struct GSF_PeerTransmitHandle *pth)
1516 {
1517   struct GSF_ConnectedPeer *cp;
1518
1519   if (NULL != pth->timeout_task)
1520   {
1521     GNUNET_SCHEDULER_cancel (pth->timeout_task);
1522     pth->timeout_task = NULL;
1523   }
1524   cp = pth->cp;
1525   GNUNET_CONTAINER_DLL_remove (cp->pth_head,
1526                                cp->pth_tail,
1527                                pth);
1528   if (GNUNET_YES == pth->is_query)
1529     GNUNET_assert (0 < cp->ppd.pending_queries--);
1530   else if (GNUNET_NO == pth->is_query)
1531     GNUNET_assert (0 < cp->ppd.pending_replies--);
1532   GNUNET_free (pth);
1533 }
1534
1535
1536 /**
1537  * Report on receiving a reply; update the performance record of the given peer.
1538  *
1539  * @param cp responding peer (will be updated)
1540  * @param request_time time at which the original query was transmitted
1541  * @param request_priority priority of the original request
1542  */
1543 void
1544 GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp,
1545                               struct GNUNET_TIME_Absolute request_time,
1546                               uint32_t request_priority)
1547 {
1548   struct GNUNET_TIME_Relative delay;
1549
1550   delay = GNUNET_TIME_absolute_get_duration (request_time);
1551   cp->ppd.avg_reply_delay.rel_value_us =
1552       (cp->ppd.avg_reply_delay.rel_value_us * (RUNAVG_DELAY_N - 1) +
1553        delay.rel_value_us) / RUNAVG_DELAY_N;
1554   cp->ppd.avg_priority =
1555       (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1) +
1556        request_priority) / RUNAVG_DELAY_N;
1557 }
1558
1559
1560 /**
1561  * Report on receiving a reply in response to an initiating client.
1562  * Remember that this peer is good for this client.
1563  *
1564  * @param cp responding peer (will be updated)
1565  * @param initiator_client local client on responsible for query
1566  */
1567 void
1568 GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp,
1569                                    struct GSF_LocalClient *initiator_client)
1570 {
1571   cp->ppd.last_client_replies[cp->last_client_replies_woff++ %
1572                               CS2P_SUCCESS_LIST_SIZE] = initiator_client;
1573 }
1574
1575
1576 /**
1577  * Report on receiving a reply in response to an initiating peer.
1578  * Remember that this peer is good for this initiating peer.
1579  *
1580  * @param cp responding peer (will be updated)
1581  * @param initiator_peer other peer responsible for query
1582  */
1583 void
1584 GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp,
1585                                  const struct GSF_ConnectedPeer *initiator_peer)
1586 {
1587   unsigned int woff;
1588
1589   woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE;
1590   GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1);
1591   cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid;
1592   GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1);
1593   cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE;
1594 }
1595
1596
1597 /**
1598  * Write peer-respect information to a file - flush the buffer entry!
1599  *
1600  * @param cls unused
1601  * @param key peer identity
1602  * @param value the `struct GSF_ConnectedPeer` to flush
1603  * @return #GNUNET_OK to continue iteration
1604  */
1605 static int
1606 flush_respect (void *cls,
1607                const struct GNUNET_PeerIdentity *key,
1608                void *value)
1609 {
1610   struct GSF_ConnectedPeer *cp = value;
1611   struct GNUNET_PeerIdentity pid;
1612
1613   if (cp->ppd.respect == cp->disk_respect)
1614     return GNUNET_OK;           /* unchanged */
1615   GNUNET_assert (0 != cp->ppd.pid);
1616   GNUNET_PEER_resolve (cp->ppd.pid, &pid);
1617   GNUNET_PEERSTORE_store (peerstore, "fs", &pid, "respect", &cp->ppd.respect,
1618                           sizeof (cp->ppd.respect),
1619                           GNUNET_TIME_UNIT_FOREVER_ABS,
1620                           GNUNET_PEERSTORE_STOREOPTION_REPLACE, NULL, NULL);
1621   return GNUNET_OK;
1622 }
1623
1624
1625 /**
1626  * A peer disconnected from us.  Tear down the connected peer
1627  * record.
1628  *
1629  * @param cls unused
1630  * @param peer identity of peer that connected
1631  */
1632 void
1633 GSF_peer_disconnect_handler_ (void *cls,
1634                               const struct GNUNET_PeerIdentity *peer)
1635 {
1636   struct GSF_ConnectedPeer *cp;
1637   struct GSF_PeerTransmitHandle *pth;
1638   struct GSF_DelayedHandle *dh;
1639
1640   cp = GSF_peer_get_ (peer);
1641   if (NULL == cp)
1642     return;                     /* must have been disconnect from core with
1643                                  * 'peer' == my_id, ignore */
1644   flush_respect (NULL, peer, cp);
1645   GNUNET_assert (GNUNET_YES ==
1646                  GNUNET_CONTAINER_multipeermap_remove (cp_map,
1647                                                        peer,
1648                                                        cp));
1649   GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# peers connected"),
1650                          GNUNET_CONTAINER_multipeermap_size (cp_map),
1651                          GNUNET_NO);
1652   if (NULL != cp->respect_iterate_req)
1653   {
1654     GNUNET_PEERSTORE_iterate_cancel (cp->respect_iterate_req);
1655     cp->respect_iterate_req = NULL;
1656   }
1657   if (NULL != cp->migration_pth)
1658   {
1659     GSF_peer_transmit_cancel_ (cp->migration_pth);
1660     cp->migration_pth = NULL;
1661   }
1662   if (NULL != cp->rc)
1663   {
1664     GNUNET_ATS_reserve_bandwidth_cancel (cp->rc);
1665     cp->rc = NULL;
1666   }
1667   if (NULL != cp->rc_delay_task)
1668   {
1669     GNUNET_SCHEDULER_cancel (cp->rc_delay_task);
1670     cp->rc_delay_task = NULL;
1671   }
1672   GNUNET_CONTAINER_multihashmap_iterate (cp->request_map,
1673                                          &cancel_pending_request, cp);
1674   GNUNET_CONTAINER_multihashmap_destroy (cp->request_map);
1675   cp->request_map = NULL;
1676   GSF_plan_notify_peer_disconnect_ (cp);
1677   GNUNET_LOAD_value_free (cp->ppd.transmission_delay);
1678   GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
1679   memset (cp->ppd.last_p2p_replies, 0, sizeof (cp->ppd.last_p2p_replies));
1680   GSF_push_stop_ (cp);
1681   if (NULL != cp->cth)
1682   {
1683     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1684     cp->cth = NULL;
1685   }
1686   GNUNET_assert (0 == cp->cth_in_progress);
1687   while (NULL != (pth = cp->pth_head))
1688   {
1689     if (pth->timeout_task != NULL)
1690     {
1691       GNUNET_SCHEDULER_cancel (pth->timeout_task);
1692       pth->timeout_task = NULL;
1693     }
1694     GNUNET_CONTAINER_DLL_remove (cp->pth_head,
1695                                  cp->pth_tail,
1696                                  pth);
1697     if (GNUNET_YES == pth->is_query)
1698       GNUNET_assert (0 < cp->ppd.pending_queries--);
1699     else if (GNUNET_NO == pth->is_query)
1700       GNUNET_assert (0 < cp->ppd.pending_replies--);
1701     pth->gmc (pth->gmc_cls, 0, NULL);
1702     GNUNET_free (pth);
1703   }
1704   while (NULL != (dh = cp->delayed_head))
1705   {
1706     GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
1707                                  cp->delayed_tail,
1708                                  dh);
1709     cp->delay_queue_size--;
1710     GNUNET_SCHEDULER_cancel (dh->delay_task);
1711     GNUNET_free (dh->pm);
1712     GNUNET_free (dh);
1713   }
1714   GNUNET_PEER_change_rc (cp->ppd.pid, -1);
1715   if (NULL != cp->mig_revive_task)
1716   {
1717     GNUNET_SCHEDULER_cancel (cp->mig_revive_task);
1718     cp->mig_revive_task = NULL;
1719   }
1720   GNUNET_break (0 == cp->ppd.pending_queries);
1721   GNUNET_break (0 == cp->ppd.pending_replies);
1722   GNUNET_free (cp);
1723 }
1724
1725
1726 /**
1727  * Closure for #call_iterator().
1728  */
1729 struct IterationContext
1730 {
1731   /**
1732    * Function to call on each entry.
1733    */
1734   GSF_ConnectedPeerIterator it;
1735
1736   /**
1737    * Closure for @e it.
1738    */
1739   void *it_cls;
1740 };
1741
1742
1743 /**
1744  * Function that calls the callback for each peer.
1745  *
1746  * @param cls the `struct IterationContext *`
1747  * @param key identity of the peer
1748  * @param value the `struct GSF_ConnectedPeer *`
1749  * @return #GNUNET_YES to continue iteration
1750  */
1751 static int
1752 call_iterator (void *cls,
1753                const struct GNUNET_PeerIdentity *key,
1754                void *value)
1755 {
1756   struct IterationContext *ic = cls;
1757   struct GSF_ConnectedPeer *cp = value;
1758
1759   ic->it (ic->it_cls,
1760           key, cp,
1761           &cp->ppd);
1762   return GNUNET_YES;
1763 }
1764
1765
1766 /**
1767  * Iterate over all connected peers.
1768  *
1769  * @param it function to call for each peer
1770  * @param it_cls closure for @a it
1771  */
1772 void
1773 GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it,
1774                               void *it_cls)
1775 {
1776   struct IterationContext ic;
1777
1778   ic.it = it;
1779   ic.it_cls = it_cls;
1780   GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1781                                          &call_iterator,
1782                                          &ic);
1783 }
1784
1785
1786 /**
1787  * Obtain the identity of a connected peer.
1788  *
1789  * @param cp peer to get identity of
1790  * @param id identity to set (written to)
1791  */
1792 void
1793 GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp,
1794                                   struct GNUNET_PeerIdentity *id)
1795 {
1796   GNUNET_assert (0 != cp->ppd.pid);
1797   GNUNET_PEER_resolve (cp->ppd.pid, id);
1798 }
1799
1800
1801 /**
1802  * Obtain the identity of a connected peer.
1803  *
1804  * @param cp peer to get identity of
1805  * @return reference to peer identity, valid until peer disconnects (!)
1806  */
1807 const struct GNUNET_PeerIdentity *
1808 GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp)
1809 {
1810   GNUNET_assert (0 != cp->ppd.pid);
1811   return GNUNET_PEER_resolve2 (cp->ppd.pid);
1812 }
1813
1814
1815 /**
1816  * Assemble a migration stop message for transmission.
1817  *
1818  * @param cls the `struct GSF_ConnectedPeer` to use
1819  * @param size number of bytes we're allowed to write to @a buf
1820  * @param buf where to copy the message
1821  * @return number of bytes copied to @a buf
1822  */
1823 static size_t
1824 create_migration_stop_message (void *cls,
1825                                size_t size,
1826                                void *buf)
1827 {
1828   struct GSF_ConnectedPeer *cp = cls;
1829   struct MigrationStopMessage msm;
1830
1831   cp->migration_pth = NULL;
1832   if (NULL == buf)
1833     return 0;
1834   GNUNET_assert (size >= sizeof (struct MigrationStopMessage));
1835   msm.header.size = htons (sizeof (struct MigrationStopMessage));
1836   msm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
1837   msm.reserved = htonl (0);
1838   msm.duration =
1839       GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
1840                                  (cp->last_migration_block));
1841   memcpy (buf, &msm, sizeof (struct MigrationStopMessage));
1842   GNUNET_STATISTICS_update (GSF_stats,
1843                             gettext_noop ("# migration stop messages sent"),
1844                             1, GNUNET_NO);
1845   return sizeof (struct MigrationStopMessage);
1846 }
1847
1848
1849 /**
1850  * Ask a peer to stop migrating data to us until the given point
1851  * in time.
1852  *
1853  * @param cp peer to ask
1854  * @param block_time until when to block
1855  */
1856 void
1857 GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp,
1858                            struct GNUNET_TIME_Absolute block_time)
1859 {
1860   if (cp->last_migration_block.abs_value_us > block_time.abs_value_us)
1861   {
1862     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1863                 "Migration already blocked for another %s\n",
1864                 GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining
1865                                                         (cp->last_migration_block), GNUNET_YES));
1866     return;                     /* already blocked */
1867   }
1868   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %s\n",
1869               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (block_time),
1870                                                       GNUNET_YES));
1871   cp->last_migration_block = block_time;
1872   if (NULL != cp->migration_pth)
1873     GSF_peer_transmit_cancel_ (cp->migration_pth);
1874   cp->migration_pth =
1875       GSF_peer_transmit_ (cp, GNUNET_SYSERR, UINT32_MAX,
1876                           GNUNET_TIME_UNIT_FOREVER_REL,
1877                           sizeof (struct MigrationStopMessage),
1878                           &create_migration_stop_message, cp);
1879 }
1880
1881
1882 /**
1883  * Notify core about a preference we have for the given peer
1884  * (to allocate more resources towards it).  The change will
1885  * be communicated the next time we reserve bandwidth with
1886  * core (not instantly).
1887  *
1888  * @param cp peer to reserve bandwidth from
1889  * @param pref preference change
1890  */
1891 void
1892 GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp,
1893                                        uint64_t pref)
1894 {
1895   cp->inc_preference += pref;
1896 }
1897
1898
1899 /**
1900  * Call this method periodically to flush respect information to disk.
1901  *
1902  * @param cls closure, not used
1903  * @param tc task context, not used
1904  */
1905 static void
1906 cron_flush_respect (void *cls,
1907                     const struct GNUNET_SCHEDULER_TaskContext *tc)
1908 {
1909   if (NULL == cp_map)
1910     return;
1911   GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1912                                          &flush_respect, NULL);
1913   if (NULL == tc)
1914     return;
1915   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1916     return;
1917   GNUNET_SCHEDULER_add_delayed_with_priority (RESPECT_FLUSH_FREQ,
1918                                               GNUNET_SCHEDULER_PRIORITY_HIGH,
1919                                               &cron_flush_respect, NULL);
1920 }
1921
1922
1923 /**
1924  * Initialize peer management subsystem.
1925  */
1926 void
1927 GSF_connected_peer_init_ ()
1928 {
1929   cp_map = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES);
1930   peerstore = GNUNET_PEERSTORE_connect (GSF_cfg);
1931   GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
1932                                       &cron_flush_respect, NULL);
1933 }
1934
1935
1936 /**
1937  * Iterator to free peer entries.
1938  *
1939  * @param cls closure, unused
1940  * @param key current key code
1941  * @param value value in the hash map (peer entry)
1942  * @return #GNUNET_YES (we should continue to iterate)
1943  */
1944 static int
1945 clean_peer (void *cls,
1946             const struct GNUNET_PeerIdentity *key,
1947             void *value)
1948 {
1949   GSF_peer_disconnect_handler_ (NULL, key);
1950   return GNUNET_YES;
1951 }
1952
1953
1954 /**
1955  * Shutdown peer management subsystem.
1956  */
1957 void
1958 GSF_connected_peer_done_ ()
1959 {
1960   cron_flush_respect (NULL, NULL);
1961   GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1962                                          &clean_peer, NULL);
1963   GNUNET_CONTAINER_multipeermap_destroy (cp_map);
1964   cp_map = NULL;
1965   GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_YES);
1966 }
1967
1968
1969 /**
1970  * Iterator to remove references to LC entry.
1971  *
1972  * @param cls the `struct GSF_LocalClient *` to look for
1973  * @param key current key code
1974  * @param value value in the hash map (peer entry)
1975  * @return #GNUNET_YES (we should continue to iterate)
1976  */
1977 static int
1978 clean_local_client (void *cls,
1979                     const struct GNUNET_PeerIdentity *key,
1980                     void *value)
1981 {
1982   const struct GSF_LocalClient *lc = cls;
1983   struct GSF_ConnectedPeer *cp = value;
1984   unsigned int i;
1985
1986   for (i = 0; i < CS2P_SUCCESS_LIST_SIZE; i++)
1987     if (cp->ppd.last_client_replies[i] == lc)
1988       cp->ppd.last_client_replies[i] = NULL;
1989   return GNUNET_YES;
1990 }
1991
1992
1993 /**
1994  * Notification that a local client disconnected.  Clean up all of our
1995  * references to the given handle.
1996  *
1997  * @param lc handle to the local client (henceforth invalid)
1998  */
1999 void
2000 GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc)
2001 {
2002   if (NULL == cp_map)
2003     return;                     /* already cleaned up */
2004   GNUNET_CONTAINER_multipeermap_iterate (cp_map, &clean_local_client,
2005                                          (void *) lc);
2006 }
2007
2008
2009 /* end of gnunet-service-fs_cp.c */