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