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