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