src/fs/gnunet-service-fs_pe.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2011 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 /**
  20  * @file fs/gnunet-service-fs_pe.c
  21  * @brief API to manage query plan
  22  * @author Christian Grothoff
  23  */
  24 #include "platform.h"
  25 #include "gnunet-service-fs.h"
  26 #include "gnunet-service-fs_cp.h"
  27 #include "gnunet-service-fs_pe.h"
  28 #include "gnunet-service-fs_pr.h"
  29
  30 /**
  31  * Collect an instane number of statistics?  May cause excessive IPC.
  32  */
  33 #define INSANE_STATISTICS GNUNET_NO
  34
  35 /**
  36  * List of GSF_PendingRequests this request plan
  37  * participates with.
  38  */
  39 struct PendingRequestList;
  40
  41 /**
  42  * Transmission plan for a peer.
  43  */
  44 struct PeerPlan;
  45
  46
  47 /**
  48  * M:N binding of plans to pending requests.
  49  * Each pending request can be in a number of plans,
  50  * and each plan can have a number of pending requests.
  51  * Objects of this type indicate a mapping of a plan to
  52  * a particular pending request.
  53  *
  54  * The corresponding head and tail of the "PE" MDLL
  55  * are stored in a `struct GSF_RequestPlan`. (We need
  56  * to be able to lookup all pending requests corresponding
  57  * to a given plan entry.)
  58  *
  59  * Similarly head and tail of the "PR" MDLL are stored
  60  * with the `struct GSF_PendingRequest`.  (We need
  61  * to be able to lookup all plan entries corresponding
  62  * to a given pending request.)
  63  */
  64 struct GSF_PendingRequestPlanBijection
  65 {
  66
  67   /**
  68    * This is a doubly-linked list.
  69    */
  70   struct GSF_PendingRequestPlanBijection *next_PR;
  71
  72   /**
  73    * This is a doubly-linked list.
  74    */
  75   struct GSF_PendingRequestPlanBijection *prev_PR;
  76
  77   /**
  78    * This is a doubly-linked list.
  79    */
  80   struct GSF_PendingRequestPlanBijection *next_PE;
  81
  82   /**
  83    * This is a doubly-linked list.
  84    */
  85   struct GSF_PendingRequestPlanBijection *prev_PE;
  86
  87   /**
  88    * Associated request plan (tells us one of the peers that
  89    * we plan to forward the request to).
  90    */
  91   struct GSF_RequestPlan *rp;
  92
  93   /**
  94    * Associated pending request (identifies request details
  95    * and one of the origins of the request).
  96    */
  97   struct GSF_PendingRequest *pr;
  98
  99 };
 100
 101
 102 /**
 103  * Information we keep per request per peer.  This is a doubly-linked
 104  * list (with head and tail in the `struct GSF_PendingRequestData`)
 105  * with one entry in each heap of each `struct PeerPlan`.  Each
 106  * entry tracks information relevant for this request and this peer.
 107  */
 108 struct GSF_RequestPlan
 109 {
 110
 111   /**
 112    * This is a doubly-linked list.
 113    */
 114   struct GSF_RequestPlan *next;
 115
 116   /**
 117    * This is a doubly-linked list.
 118    */
 119   struct GSF_RequestPlan *prev;
 120
 121   /**
 122    * Heap node associated with this request and this peer.
 123    */
 124   struct GNUNET_CONTAINER_HeapNode *hn;
 125
 126   /**
 127    * The transmission plan for a peer that this request is associated with.
 128    */
 129   struct PeerPlan *pp;
 130
 131   /**
 132    * Head of list of associated pending requests.  This tells us
 133    * which incoming requests from other peers this plan entry
 134    * corresponds to.
 135    */
 136   struct GSF_PendingRequestPlanBijection *pe_head;
 137
 138   /**
 139    * Tail of list of associated pending requests.
 140    */
 141   struct GSF_PendingRequestPlanBijection *pe_tail;
 142
 143   /**
 144    * Earliest time we'd be happy to (re)transmit this request.
 145    */
 146   struct GNUNET_TIME_Absolute earliest_transmission;
 147
 148   /**
 149    * When was the last time we transmitted this request to this peer? 0 for never.
 150    */
 151   struct GNUNET_TIME_Absolute last_transmission;
 152
 153   /**
 154    * Current priority for this request for this target.
 155    */
 156   uint64_t priority;
 157
 158   /**
 159    * How often did we transmit this request to this peer?
 160    */
 161   unsigned int transmission_counter;
 162
 163 };
 164
 165
 166 /**
 167  * Transmission plan for a peer.
 168  */
 169 struct PeerPlan
 170 {
 171   /**
 172    * Heap with pending queries (`struct GSF_RequestPlan`), higher weights mean higher priority.
 173    */
 174   struct GNUNET_CONTAINER_Heap *priority_heap;
 175
 176   /**
 177    * Heap with pending queries (`struct GSF_RequestPlan`), by transmission time, lowest first.
 178    */
 179   struct GNUNET_CONTAINER_Heap *delay_heap;
 180
 181   /**
 182    * Map of queries to plan entries.  All entries in the @e priority_heap
 183    * or @e delay_heap should be in the @e plan_map.  Note that it is
 184    * possible for the @e plan_map to have multiple entries for the same
 185    * query.
 186    */
 187   struct GNUNET_CONTAINER_MultiHashMap *plan_map;
 188
 189   /**
 190    * Peer for which this is the plan.
 191    */
 192   struct GSF_ConnectedPeer *cp;
 193
 194   /**
 195    * Current task for executing the plan.
 196    */
 197   struct GNUNET_SCHEDULER_Task *task;
 198
 199   /**
 200    * Current message under transmission for the plan.
 201    */
 202   struct GNUNET_MQ_Envelope *env;
 203
 204 };
 205
 206
 207 /**
 208  * Hash map from peer identities to PeerPlans.
 209  */
 210 static struct GNUNET_CONTAINER_MultiPeerMap *plans;
 211
 212 /**
 213  * Sum of all transmission counters (equals total delay for all plan entries).
 214  */
 215 static unsigned long long total_delay;
 216
 217 /**
 218  * Number of plan entries.
 219  */
 220 static unsigned long long plan_count;
 221
 222
 223 /**
 224  * Return the query (key in the plan_map) for the given request plan.
 225  * Note that this key may change as there can be multiple pending
 226  * requests for the same key and we just return _one_ of them; this
 227  * particular one might complete while another one might still be
 228  * active, hence the lifetime of the returned hash code is NOT
 229  * necessarily identical to that of the `struct GSF_RequestPlan`
 230  * given.
 231  *
 232  * @param rp a request plan
 233  * @return the associated query
 234  */
 235 static const struct GNUNET_HashCode *
 236 get_rp_key (struct GSF_RequestPlan *rp)
 237 {
 238   return &GSF_pending_request_get_data_ (rp->pe_head->pr)->query;
 239 }
 240
 241
 242 /**
 243  * Insert the given request plan into the heap with the appropriate weight.
 244  *
 245  * @param pp associated peer's plan
 246  * @param rp request to plan
 247  */
 248 static void
 249 plan (struct PeerPlan *pp,
 250       struct GSF_RequestPlan *rp)
 251 {
 252 #define N ((double)128.0)
 253   /**
 254    * Running average delay we currently impose.
 255    */
 256   static double avg_delay;
 257
 258   struct GSF_PendingRequestData *prd;
 259   struct GNUNET_TIME_Relative delay;
 260
 261   GNUNET_assert (rp->pp == pp);
 262   GNUNET_STATISTICS_set (GSF_stats,
 263                          gettext_noop ("# average retransmission delay (ms)"),
 264                          total_delay * 1000LL / plan_count, GNUNET_NO);
 265   prd = GSF_pending_request_get_data_ (rp->pe_head->pr);
 266
 267   if (rp->transmission_counter < 8)
 268     delay =
 269         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 270                                        rp->transmission_counter);
 271   else if (rp->transmission_counter < 32)
 272     delay =
 273         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 274                                        8 +
 275                                        (1LL << (rp->transmission_counter - 8)));
 276   else
 277     delay =
 278         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 279                                        8 + (1LL << 24));
 280   delay.rel_value_us =
 281     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 282                               delay.rel_value_us + 1);
 283   /* Add 0.01 to avg_delay to avoid division-by-zero later */
 284   avg_delay = (((avg_delay * (N - 1.0)) + delay.rel_value_us) / N) + 0.01;
 285
 286   /*
 287    * For the priority, we need to consider a few basic rules:
 288    * 1) if we just started requesting (delay is small), we should
 289    * virtually always have a priority of zero.
 290    * 2) for requests with average latency, our priority should match
 291    * the average priority observed on the network
 292    * 3) even the longest-running requests should not be WAY out of
 293    * the observed average (thus we bound by a factor of 2)
 294    * 4) we add +1 to the observed average priority to avoid everyone
 295    * staying put at zero (2 * 0 = 0...).
 296    *
 297    * Using the specific calculation below, we get:
 298    *
 299    * delay = 0 => priority = 0;
 300    * delay = avg delay => priority = running-average-observed-priority;
 301    * delay >> avg_delay => priority = 2 * running-average-observed-priority;
 302    *
 303    * which satisfies all of the rules above.
 304    *
 305    * Note: M_PI_4 = PI/4 = arctan(1)
 306    */
 307   rp->priority =
 308       round ((GSF_current_priorities +
 309               1.0) * atan (delay.rel_value_us / avg_delay)) / M_PI_4;
 310   /* Note: usage of 'round' and 'atan' requires -lm */
 311
 312   if (rp->transmission_counter != 0)
 313     delay.rel_value_us += TTL_DECREMENT * 1000;
 314   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 315               "Considering (re)transmission number %u in %s\n",
 316               (unsigned int) rp->transmission_counter,
 317               GNUNET_STRINGS_relative_time_to_string (delay,
 318                                                       GNUNET_YES));
 319   rp->earliest_transmission = GNUNET_TIME_relative_to_absolute (delay);
 320   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 321               "Earliest (re)transmission for `%s' in %us\n",
 322               GNUNET_h2s (&prd->query),
 323               rp->transmission_counter);
 324   GNUNET_assert (rp->hn == NULL);
 325   if (0 == GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission).rel_value_us)
 326     rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap,
 327                                            rp,
 328                                            rp->priority);
 329   else
 330     rp->hn =
 331         GNUNET_CONTAINER_heap_insert (pp->delay_heap,
 332                                       rp,
 333                                       rp->earliest_transmission.abs_value_us);
 334   GNUNET_assert (GNUNET_YES ==
 335                  GNUNET_CONTAINER_multihashmap_contains_value (pp->plan_map,
 336                                                                get_rp_key (rp),
 337                                                                rp));
 338 #undef N
 339 }
 340
 341
 342 /**
 343  * Get the pending request with the highest TTL from the given plan.
 344  *
 345  * @param rp plan to investigate
 346  * @return pending request with highest TTL
 347  */
 348 struct GSF_PendingRequest *
 349 get_latest (const struct GSF_RequestPlan *rp)
 350 {
 351   struct GSF_PendingRequest *ret;
 352   struct GSF_PendingRequestPlanBijection *bi;
 353   const struct GSF_PendingRequestData *rprd;
 354   const struct GSF_PendingRequestData *prd;
 355
 356   bi = rp->pe_head;
 357   if (NULL == bi)
 358     return NULL; /* should never happen */
 359   ret = bi->pr;
 360   rprd = GSF_pending_request_get_data_ (ret);
 361   for (bi = bi->next_PE; NULL != bi; bi = bi->next_PE)
 362   {
 363     GNUNET_break (GNUNET_YES ==
 364                   GSF_pending_request_test_active_ (bi->pr));
 365     prd = GSF_pending_request_get_data_ (bi->pr);
 366     if (prd->ttl.abs_value_us > rprd->ttl.abs_value_us)
 367     {
 368       ret = bi->pr;
 369       rprd = prd;
 370     }
 371   }
 372   return ret;
 373 }
 374
 375
 376 /**
 377  * Figure out when and how to transmit to the given peer.
 378  *
 379  * @param cls the `struct PeerPlan`
 380  */
 381 static void
 382 schedule_peer_transmission (void *cls)
 383 {
 384   struct PeerPlan *pp = cls;
 385   struct GSF_RequestPlan *rp;
 386   struct GNUNET_TIME_Relative delay;
 387
 388   if (NULL != pp->task)
 389   {
 390     pp->task = NULL;
 391   }
 392   else
 393   {
 394     GNUNET_assert (NULL != pp->env);
 395     pp->env = NULL;
 396   }
 397   /* move ready requests to priority queue */
 398   while ((NULL != (rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap))) &&
 399          (0 == GNUNET_TIME_absolute_get_remaining
 400           (rp->earliest_transmission).rel_value_us))
 401   {
 402     GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->delay_heap));
 403     rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap,
 404                                            rp,
 405                                            rp->priority);
 406   }
 407   if (0 == GNUNET_CONTAINER_heap_get_size (pp->priority_heap))
 408   {
 409     /* priority heap (still) empty, check for delay... */
 410     rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap);
 411     if (NULL == rp)
 412     {
 413       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 414                   "No active requests for plan %p.\n",
 415                   pp);
 416       return;                   /* both queues empty */
 417     }
 418     delay = GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission);
 419     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 420                 "Sleeping for %s before retrying requests on plan %p.\n",
 421                 GNUNET_STRINGS_relative_time_to_string (delay,
 422                                                         GNUNET_YES),
 423                 pp);
 424     GNUNET_STATISTICS_set (GSF_stats,
 425                            gettext_noop ("# delay heap timeout (ms)"),
 426                            delay.rel_value_us / 1000LL, GNUNET_NO);
 427
 428     pp->task
 429       = GNUNET_SCHEDULER_add_at (rp->earliest_transmission,
 430                                  &schedule_peer_transmission,
 431                                  pp);
 432     return;
 433   }
 434 #if INSANE_STATISTICS
 435   GNUNET_STATISTICS_update (GSF_stats,
 436                             gettext_noop ("# query plans executed"),
 437                             1,
 438                             GNUNET_NO);
 439 #endif
 440   /* process from priority heap */
 441   rp = GNUNET_CONTAINER_heap_remove_root (pp->priority_heap);
 442   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 443               "Executing query plan %p\n",
 444               rp);
 445   GNUNET_assert (NULL != rp);
 446   rp->hn = NULL;
 447   rp->last_transmission = GNUNET_TIME_absolute_get ();
 448   rp->transmission_counter++;
 449   total_delay++;
 450   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 451               "Executing plan %p executed %u times, planning retransmission\n",
 452               rp,
 453               rp->transmission_counter);
 454   GNUNET_assert (NULL == pp->env);
 455   pp->env = GSF_pending_request_get_message_ (get_latest (rp));
 456   GNUNET_MQ_notify_sent (pp->env,
 457                          &schedule_peer_transmission,
 458                          pp);
 459   GSF_peer_transmit_ (pp->cp,
 460                       GNUNET_YES,
 461                       rp->priority,
 462                       pp->env);
 463   GNUNET_STATISTICS_update (GSF_stats,
 464                             gettext_noop ("# query messages sent to other peers"),
 465                             1,
 466                             GNUNET_NO);
 467   plan (pp,
 468         rp);
 469 }
 470
 471
 472 /**
 473  * Closure for merge_pr().
 474  */
 475 struct MergeContext
 476 {
 477
 478   /**
 479    * Request we are trying to merge.
 480    */
 481   struct GSF_PendingRequest *pr;
 482
 483   /**
 484    * Set to #GNUNET_YES if we succeeded to merge.
 485    */
 486   int merged;
 487
 488 };
 489
 490
 491 /**
 492  * Iterator that checks if an equivalent request is already
 493  * present for this peer.
 494  *
 495  * @param cls closure
 496  * @param query the query
 497  * @param element request plan stored at the node
 498  * @return #GNUNET_YES if we should continue to iterate,
 499  *         #GNUNET_NO if not (merge success)
 500  */
 501 static int
 502 merge_pr (void *cls,
 503           const struct GNUNET_HashCode *query,
 504           void *element)
 505 {
 506   struct MergeContext *mpr = cls;
 507   struct GSF_RequestPlan *rp = element;
 508   struct GSF_PendingRequestData *prd;
 509   struct GSF_PendingRequestPlanBijection *bi;
 510   struct GSF_PendingRequest *latest;
 511
 512   GNUNET_break (GNUNET_YES ==
 513                 GSF_pending_request_test_active_ (mpr->pr));
 514   if (GNUNET_OK !=
 515       GSF_pending_request_is_compatible_ (mpr->pr,
 516                                           rp->pe_head->pr))
 517     return GNUNET_YES;
 518   /* merge new request with existing request plan */
 519   bi = GNUNET_new (struct GSF_PendingRequestPlanBijection);
 520   bi->rp = rp;
 521   bi->pr = mpr->pr;
 522   prd = GSF_pending_request_get_data_ (mpr->pr);
 523   GNUNET_CONTAINER_MDLL_insert (PR,
 524                                 prd->pr_head,
 525                                 prd->pr_tail,
 526                                 bi);
 527   GNUNET_CONTAINER_MDLL_insert (PE,
 528                                 rp->pe_head,
 529                                 rp->pe_tail,
 530                                 bi);
 531   mpr->merged = GNUNET_YES;
 532 #if INSANE_STATISTICS
 533   GNUNET_STATISTICS_update (GSF_stats,
 534                             gettext_noop ("# requests merged"),
 535                             1,
 536                             GNUNET_NO);
 537 #endif
 538   latest = get_latest (rp);
 539   if (GSF_pending_request_get_data_ (latest)->ttl.abs_value_us <
 540       prd->ttl.abs_value_us)
 541   {
 542 #if INSANE_STATISTICS
 543     GNUNET_STATISTICS_update (GSF_stats,
 544                               gettext_noop ("# requests refreshed"),
 545                               1,
 546                               GNUNET_NO);
 547 #endif
 548     rp->transmission_counter = 0;       /* reset */
 549   }
 550   return GNUNET_NO;
 551 }
 552
 553
 554 /**
 555  * Create a new query plan entry.
 556  *
 557  * @param cp peer with the entry
 558  * @param pr request with the entry
 559  */
 560 void
 561 GSF_plan_add_ (struct GSF_ConnectedPeer *cp,
 562                struct GSF_PendingRequest *pr)
 563 {
 564   const struct GNUNET_PeerIdentity *id;
 565   struct PeerPlan *pp;
 566   struct GSF_PendingRequestData *prd;
 567   struct GSF_RequestPlan *rp;
 568   struct GSF_PendingRequestPlanBijection *bi;
 569   struct MergeContext mpc;
 570
 571   GNUNET_assert (GNUNET_YES ==
 572                  GSF_pending_request_test_active_ (pr));
 573   GNUNET_assert (NULL != cp);
 574   id = GSF_connected_peer_get_identity2_ (cp);
 575   pp = GNUNET_CONTAINER_multipeermap_get (plans, id);
 576   if (NULL == pp)
 577   {
 578     pp = GNUNET_new (struct PeerPlan);
 579     pp->plan_map = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_NO);
 580     pp->priority_heap =
 581         GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MAX);
 582     pp->delay_heap =
 583         GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
 584     pp->cp = cp;
 585     GNUNET_assert (GNUNET_OK ==
 586                    GNUNET_CONTAINER_multipeermap_put (plans,
 587                                                       id,
 588                                                       pp,
 589                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 590     pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission,
 591                                          pp);
 592   }
 593   mpc.merged = GNUNET_NO;
 594   mpc.pr = pr;
 595   prd = GSF_pending_request_get_data_ (pr);
 596   GNUNET_CONTAINER_multihashmap_get_multiple (pp->plan_map,
 597                                               &prd->query,
 598                                               &merge_pr,
 599                                               &mpc);
 600   if (GNUNET_NO != mpc.merged)
 601     return;
 602   plan_count++;
 603   GNUNET_STATISTICS_update (GSF_stats,
 604                             gettext_noop ("# query plan entries"),
 605                             1,
 606                             GNUNET_NO);
 607   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 608               "Planning transmission of query `%s' to peer `%s'\n",
 609               GNUNET_h2s (&prd->query),
 610               GNUNET_i2s (id));
 611   rp = GNUNET_new (struct GSF_RequestPlan);
 612   bi = GNUNET_new (struct GSF_PendingRequestPlanBijection);
 613   bi->rp = rp;
 614   bi->pr = pr;
 615   GNUNET_CONTAINER_MDLL_insert (PR,
 616                                 prd->pr_head,
 617                                 prd->pr_tail,
 618                                 bi);
 619   GNUNET_CONTAINER_MDLL_insert (PE,
 620                                 rp->pe_head,
 621                                 rp->pe_tail,
 622                                 bi);
 623   rp->pp = pp;
 624   GNUNET_assert (GNUNET_YES ==
 625                  GNUNET_CONTAINER_multihashmap_put (pp->plan_map,
 626                                                     get_rp_key (rp),
 627                                                     rp,
 628                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 629   plan (pp,
 630         rp);
 631 }
 632
 633
 634 /**
 635  * Notify the plan about a peer being no longer available;
 636  * destroy all entries associated with this peer.
 637  *
 638  * @param cp connected peer
 639  */
 640 void
 641 GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp)
 642 {
 643   const struct GNUNET_PeerIdentity *id;
 644   struct PeerPlan *pp;
 645   struct GSF_RequestPlan *rp;
 646   struct GSF_PendingRequestData *prd;
 647   struct GSF_PendingRequestPlanBijection *bi;
 648
 649   id = GSF_connected_peer_get_identity2_ (cp);
 650   pp = GNUNET_CONTAINER_multipeermap_get (plans, id);
 651   if (NULL == pp)
 652     return;                     /* nothing was ever planned for this peer */
 653   GNUNET_assert (GNUNET_YES ==
 654                  GNUNET_CONTAINER_multipeermap_remove (plans, id,
 655                                                        pp));
 656   if (NULL != pp->task)
 657   {
 658     GNUNET_SCHEDULER_cancel (pp->task);
 659     pp->task = NULL;
 660   }
 661   while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->priority_heap)))
 662   {
 663     GNUNET_break (GNUNET_YES ==
 664                   GNUNET_CONTAINER_multihashmap_remove (pp->plan_map,
 665                                                         get_rp_key (rp),
 666                                                         rp));
 667     while (NULL != (bi = rp->pe_head))
 668     {
 669       GNUNET_CONTAINER_MDLL_remove (PE,
 670                                     rp->pe_head,
 671                                     rp->pe_tail,
 672                                     bi);
 673       prd = GSF_pending_request_get_data_ (bi->pr);
 674       GNUNET_CONTAINER_MDLL_remove (PR,
 675                                     prd->pr_head,
 676                                     prd->pr_tail,
 677                                     bi);
 678       GNUNET_free (bi);
 679     }
 680     plan_count--;
 681     GNUNET_free (rp);
 682   }
 683   GNUNET_CONTAINER_heap_destroy (pp->priority_heap);
 684   while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->delay_heap)))
 685   {
 686     GNUNET_break (GNUNET_YES ==
 687                   GNUNET_CONTAINER_multihashmap_remove (pp->plan_map,
 688                                                         get_rp_key (rp),
 689                                                         rp));
 690     while (NULL != (bi = rp->pe_head))
 691     {
 692       prd = GSF_pending_request_get_data_ (bi->pr);
 693       GNUNET_CONTAINER_MDLL_remove (PE,
 694                                     rp->pe_head,
 695                                     rp->pe_tail,
 696                                     bi);
 697       GNUNET_CONTAINER_MDLL_remove (PR,
 698                                     prd->pr_head,
 699                                     prd->pr_tail,
 700                                     bi);
 701       GNUNET_free (bi);
 702     }
 703     plan_count--;
 704     GNUNET_free (rp);
 705   }
 706   GNUNET_STATISTICS_set (GSF_stats,
 707                          gettext_noop ("# query plan entries"),
 708                          plan_count,
 709                          GNUNET_NO);
 710   GNUNET_CONTAINER_heap_destroy (pp->delay_heap);
 711   GNUNET_CONTAINER_multihashmap_destroy (pp->plan_map);
 712   GNUNET_free (pp);
 713 }
 714
 715
 716 /**
 717  * Get the last transmission attempt time for the request plan list
 718  * referenced by @a pr_head, that was sent to @a sender
 719  *
 720  * @param pr_head request plan reference list to check.
 721  * @param sender the peer that we've sent the request to.
 722  * @param result the timestamp to fill, set to #GNUNET_TIME_UNIT_FOREVER_ABS if never transmitted
 723  * @return #GNUNET_YES if @a result was changed, #GNUNET_NO otherwise.
 724  */
 725 int
 726 GSF_request_plan_reference_get_last_transmission_ (struct GSF_PendingRequestPlanBijection *pr_head,
 727                                                    struct GSF_ConnectedPeer *sender,
 728                                                    struct GNUNET_TIME_Absolute *result)
 729 {
 730   struct GSF_PendingRequestPlanBijection *bi;
 731
 732   for (bi = pr_head; NULL != bi; bi = bi->next_PR)
 733   {
 734     if (bi->rp->pp->cp == sender)
 735     {
 736       if (0 == bi->rp->last_transmission.abs_value_us)
 737         *result = GNUNET_TIME_UNIT_FOREVER_ABS;
 738       else
 739         *result = bi->rp->last_transmission;
 740       return GNUNET_YES;
 741     }
 742   }
 743   return GNUNET_NO;
 744 }
 745
 746
 747 /**
 748  * Notify the plan about a request being done; destroy all entries
 749  * associated with this request.
 750  *
 751  * @param pr request that is done
 752  */
 753 void
 754 GSF_plan_notify_request_done_ (struct GSF_PendingRequest *pr)
 755 {
 756   struct GSF_RequestPlan *rp;
 757   struct GSF_PendingRequestData *prd;
 758   struct GSF_PendingRequestPlanBijection *bi;
 759
 760   prd = GSF_pending_request_get_data_ (pr);
 761   while (NULL != (bi = prd->pr_head))
 762   {
 763     rp = bi->rp;
 764     GNUNET_CONTAINER_MDLL_remove (PR,
 765                                   prd->pr_head,
 766                                   prd->pr_tail,
 767                                   bi);
 768     GNUNET_CONTAINER_MDLL_remove (PE,
 769                                   rp->pe_head,
 770                                   rp->pe_tail,
 771                                   bi);
 772     GNUNET_assert (bi->pr == pr);
 773     if (NULL == rp->pe_head)
 774     {
 775       GNUNET_CONTAINER_heap_remove_node (rp->hn);
 776       plan_count--;
 777       GNUNET_break (GNUNET_YES ==
 778                     GNUNET_CONTAINER_multihashmap_remove (rp->pp->plan_map,
 779                                                           &prd->query,
 780                                                           rp));
 781       GNUNET_free (rp);
 782     }
 783     GNUNET_free (bi);
 784   }
 785   GNUNET_STATISTICS_set (GSF_stats,
 786                          gettext_noop ("# query plan entries"),
 787                          plan_count,
 788                          GNUNET_NO);
 789 }
 790
 791
 792 /**
 793  * Initialize plan subsystem.
 794  */
 795 void
 796 GSF_plan_init ()
 797 {
 798   plans = GNUNET_CONTAINER_multipeermap_create (256,
 799                                                 GNUNET_YES);
 800 }
 801
 802
 803 /**
 804  * Shutdown plan subsystem.
 805  */
 806 void
 807 GSF_plan_done ()
 808 {
 809   GNUNET_assert (0 == GNUNET_CONTAINER_multipeermap_size (plans));
 810   GNUNET_CONTAINER_multipeermap_destroy (plans);
 811 }
 812
 813
 814
 815 /* end of gnunet-service-fs_pe.h */