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