src/fs/gnunet-service-fs_cp.c

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