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