src/fs/gnunet-service-fs_cp.c

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