tolerate additional IPv4 address now available for gnunet.org
[oweals/gnunet.git] / src / fs / gnunet-service-fs_pe.c
index a17715dcf4f1e0b43c5962a1980e5786d6e71081..5e85bfdb7c470bd9246920d833c6f4c16101a353 100644 (file)
@@ -1,21 +1,21 @@
 /*
      This file is part of GNUnet.
-     (C) 2011 Christian Grothoff (and other contributing authors)
+     Copyright (C) 2011 GNUnet e.V.
 
-     GNUnet is free software; you can redistribute it and/or modify
-     it under the terms of the GNU General Public License as published
-     by the Free Software Foundation; either version 3, or (at your
-     option) any later version.
+     GNUnet is free software: you can redistribute it and/or modify it
+     under the terms of the GNU Affero General Public License as published
+     by the Free Software Foundation, either version 3 of the License,
+     or (at your option) any later version.
 
      GNUnet is distributed in the hope that it will be useful, but
      WITHOUT ANY WARRANTY; without even the implied warranty of
      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-     General Public License for more details.
+     Affero General Public License for more details.
+    
+     You should have received a copy of the GNU Affero General Public License
+     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-     You should have received a copy of the GNU General Public License
-     along with GNUnet; see the file COPYING.  If not, write to the
-     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
-     Boston, MA 02111-1307, USA.
+     SPDX-License-Identifier: AGPL3.0-or-later
 */
 
 /**
@@ -54,12 +54,12 @@ struct PeerPlan;
  * a particular pending request.
  *
  * The corresponding head and tail of the "PE" MDLL
- * are stored in a 'struct GSF_RequestPlan'. (We need
+ * are stored in a `struct GSF_RequestPlan`. (We need
  * to be able to lookup all pending requests corresponding
  * to a given plan entry.)
  *
  * Similarly head and tail of the "PR" MDLL are stored
- * with the 'struct GSF_PendingRequest'.  (We need
+ * with the `struct GSF_PendingRequest`.  (We need
  * to be able to lookup all plan entries corresponding
  * to a given pending request.)
  */
@@ -87,12 +87,14 @@ struct GSF_PendingRequestPlanBijection
   struct GSF_PendingRequestPlanBijection *prev_PE;
 
   /**
-   * Associated request plan.
+   * Associated request plan (tells us one of the peers that
+   * we plan to forward the request to).
    */
   struct GSF_RequestPlan *rp;
 
   /**
-   * Associated pending request.
+   * Associated pending request (identifies request details
+   * and one of the origins of the request).
    */
   struct GSF_PendingRequest *pr;
 
@@ -101,8 +103,8 @@ struct GSF_PendingRequestPlanBijection
 
 /**
  * Information we keep per request per peer.  This is a doubly-linked
- * list (with head and tail in the 'struct GSF_PendingRequestData')
- * with one entry in each heap of each 'struct PeerPlan'.  Each
+ * list (with head and tail in the `struct GSF_PendingRequestData`)
+ * with one entry in each heap of each `struct PeerPlan`.  Each
  * entry tracks information relevant for this request and this peer.
  */
 struct GSF_RequestPlan
@@ -129,7 +131,9 @@ struct GSF_RequestPlan
   struct PeerPlan *pp;
 
   /**
-   * Head of list of associated pending requests.
+   * Head of list of associated pending requests.  This tells us
+   * which incoming requests from other peers this plan entry
+   * corresponds to.
    */
   struct GSF_PendingRequestPlanBijection *pe_head;
 
@@ -167,27 +171,23 @@ struct GSF_RequestPlan
 struct PeerPlan
 {
   /**
-   * Heap with pending queries (struct GSF_RequestPlan), higher weights mean higher priority.
+   * Heap with pending queries (`struct GSF_RequestPlan`), higher weights mean higher priority.
    */
   struct GNUNET_CONTAINER_Heap *priority_heap;
 
   /**
-   * Heap with pending queries (struct GSF_RequestPlan), by transmission time, lowest first.
+   * Heap with pending queries (`struct GSF_RequestPlan`), by transmission time, lowest first.
    */
   struct GNUNET_CONTAINER_Heap *delay_heap;
 
   /**
-   * Map of queries to plan entries.  All entries in the priority_heap or delay_heap
-   * should be in the plan map.  Note that it IS possible for the plan map to have
-   * multiple entries for the same query.
+   * Map of queries to plan entries.  All entries in the @e priority_heap
+   * or @e delay_heap should be in the @e plan_map.  Note that it is
+   * possible for the @e plan_map to have multiple entries for the same
+   * query.
    */
   struct GNUNET_CONTAINER_MultiHashMap *plan_map;
 
-  /**
-   * Current transmission request handle.
-   */
-  struct GSF_PeerTransmitHandle *pth;
-
   /**
    * Peer for which this is the plan.
    */
@@ -196,7 +196,13 @@ struct PeerPlan
   /**
    * Current task for executing the plan.
    */
-  GNUNET_SCHEDULER_TaskIdentifier task;
+  struct GNUNET_SCHEDULER_Task *task;
+
+  /**
+   * Current message under transmission for the plan.
+   */
+  struct GNUNET_MQ_Envelope *env;
+
 };
 
 
@@ -222,7 +228,7 @@ static unsigned long long plan_count;
  * requests for the same key and we just return _one_ of them; this
  * particular one might complete while another one might still be
  * active, hence the lifetime of the returned hash code is NOT
- * necessarily identical to that of the 'struct GSF_RequestPlan'
+ * necessarily identical to that of the `struct GSF_RequestPlan`
  * given.
  *
  * @param rp a request plan
@@ -235,17 +241,6 @@ get_rp_key (struct GSF_RequestPlan *rp)
 }
 
 
-/**
- * Figure out when and how to transmit to the given peer.
- *
- * @param cls the 'struct GSF_ConnectedPeer' for transmission
- * @param tc scheduler context
- */
-static void
-schedule_peer_transmission (void *cls,
-                            const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
 /**
  * Insert the given request plan into the heap with the appropriate weight.
  *
@@ -253,7 +248,8 @@ schedule_peer_transmission (void *cls,
  * @param rp request to plan
  */
 static void
-plan (struct PeerPlan *pp, struct GSF_RequestPlan *rp)
+plan (struct PeerPlan *pp,
+      struct GSF_RequestPlan *rp)
 {
 #define N ((double)128.0)
   /**
@@ -325,21 +321,22 @@ plan (struct PeerPlan *pp, struct GSF_RequestPlan *rp)
   rp->earliest_transmission = GNUNET_TIME_relative_to_absolute (delay);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Earliest (re)transmission for `%s' in %us\n",
-              GNUNET_h2s (&prd->query), rp->transmission_counter);
+              GNUNET_h2s (&prd->query),
+             rp->transmission_counter);
   GNUNET_assert (rp->hn == NULL);
   if (0 == GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission).rel_value_us)
-    rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, rp, rp->priority);
+    rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap,
+                                          rp,
+                                          rp->priority);
   else
     rp->hn =
-        GNUNET_CONTAINER_heap_insert (pp->delay_heap, rp,
+        GNUNET_CONTAINER_heap_insert (pp->delay_heap,
+                                     rp,
                                       rp->earliest_transmission.abs_value_us);
   GNUNET_assert (GNUNET_YES ==
                  GNUNET_CONTAINER_multihashmap_contains_value (pp->plan_map,
                                                                get_rp_key (rp),
                                                                rp));
-  if (GNUNET_SCHEDULER_NO_TASK != pp->task)
-    GNUNET_SCHEDULER_cancel (pp->task);
-  pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
 #undef N
 }
 
@@ -355,107 +352,49 @@ get_latest (const struct GSF_RequestPlan *rp)
 {
   struct GSF_PendingRequest *ret;
   struct GSF_PendingRequestPlanBijection *bi;
+  const struct GSF_PendingRequestData *rprd;
+  const struct GSF_PendingRequestData *prd;
 
   bi = rp->pe_head;
   if (NULL == bi)
     return NULL; /* should never happen */
   ret = bi->pr;
-  bi = bi->next_PE;
-  while (NULL != bi)
+  rprd = GSF_pending_request_get_data_ (ret);
+  for (bi = bi->next_PE; NULL != bi; bi = bi->next_PE)
   {
-    if (GSF_pending_request_get_data_ (bi->pr)->ttl.abs_value_us >
-        GSF_pending_request_get_data_ (ret)->ttl.abs_value_us)
+    GNUNET_break (GNUNET_YES ==
+                  GSF_pending_request_test_active_ (bi->pr));
+    prd = GSF_pending_request_get_data_ (bi->pr);
+    if (prd->ttl.abs_value_us > rprd->ttl.abs_value_us)
+    {
       ret = bi->pr;
-    bi = bi->next_PE;
+      rprd = prd;
+    }
   }
   return ret;
 }
 
 
-/**
- * Function called to get a message for transmission.
- *
- * @param cls closure
- * @param buf_size number of bytes available in buf
- * @param buf where to copy the message, NULL on error (peer disconnect)
- * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
- */
-static size_t
-transmit_message_callback (void *cls, size_t buf_size, void *buf)
-{
-  struct PeerPlan *pp = cls;
-  struct GSF_RequestPlan *rp;
-  size_t msize;
-
-  pp->pth = NULL;
-  if (NULL == buf)
-  {
-    /* failed, try again... */
-    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
-      GNUNET_SCHEDULER_cancel (pp->task);
-
-    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
-    GNUNET_STATISTICS_update (GSF_stats,
-                              gettext_noop
-                              ("# transmission failed (core has no bandwidth)"),
-                              1, GNUNET_NO);
-    return 0;
-  }
-  rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap);
-  if (NULL == rp)
-  {
-    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
-      GNUNET_SCHEDULER_cancel (pp->task);
-    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
-    return 0;
-  }
-  msize = GSF_pending_request_get_message_ (get_latest (rp), buf_size, buf);
-  if (msize > buf_size)
-  {
-    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
-      GNUNET_SCHEDULER_cancel (pp->task);
-    /* buffer to small (message changed), try again */
-    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
-    return 0;
-  }
-  /* remove from root, add again elsewhere... */
-  GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->priority_heap));
-  rp->hn = NULL;
-  rp->last_transmission = GNUNET_TIME_absolute_get ();
-  rp->transmission_counter++;
-  total_delay++;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Executing plan %p executed %u times, planning retransmission\n",
-              rp, rp->transmission_counter);
-  plan (pp, rp);
-  GNUNET_STATISTICS_update (GSF_stats,
-                            gettext_noop
-                            ("# query messages sent to other peers"), 1,
-                            GNUNET_NO);
-  return msize;
-}
-
-
 /**
  * Figure out when and how to transmit to the given peer.
  *
  * @param cls the `struct PeerPlan`
- * @param tc scheduler context
  */
 static void
-schedule_peer_transmission (void *cls,
-                            const struct GNUNET_SCHEDULER_TaskContext *tc)
+schedule_peer_transmission (void *cls)
 {
   struct PeerPlan *pp = cls;
   struct GSF_RequestPlan *rp;
-  size_t msize;
   struct GNUNET_TIME_Relative delay;
 
-  pp->task = GNUNET_SCHEDULER_NO_TASK;
-  if (NULL != pp->pth)
+  if (NULL != pp->task)
+  {
+    pp->task = NULL;
+  }
+  else
   {
-    GSF_peer_transmit_cancel_ (pp->pth);
-    pp->pth = NULL;
+    GNUNET_assert (NULL != pp->env);
+    pp->env = NULL;
   }
   /* move ready requests to priority queue */
   while ((NULL != (rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap))) &&
@@ -463,7 +402,9 @@ schedule_peer_transmission (void *cls,
           (rp->earliest_transmission).rel_value_us))
   {
     GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->delay_heap));
-    rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, rp, rp->priority);
+    rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap,
+                                           rp,
+                                           rp->priority);
   }
   if (0 == GNUNET_CONTAINER_heap_get_size (pp->priority_heap))
   {
@@ -471,7 +412,8 @@ schedule_peer_transmission (void *cls,
     rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap);
     if (NULL == rp)
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No active requests for plan %p.\n",
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "No active requests for plan %p.\n",
                   pp);
       return;                   /* both queues empty */
     }
@@ -481,27 +423,51 @@ schedule_peer_transmission (void *cls,
                 GNUNET_STRINGS_relative_time_to_string (delay,
                                                        GNUNET_YES),
                pp);
-    GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# delay heap timeout (ms)"),
+    GNUNET_STATISTICS_set (GSF_stats,
+                           gettext_noop ("# delay heap timeout (ms)"),
                            delay.rel_value_us / 1000LL, GNUNET_NO);
 
-    pp->task =
-        GNUNET_SCHEDULER_add_delayed (delay, &schedule_peer_transmission, pp);
+    pp->task
+      = GNUNET_SCHEDULER_add_at (rp->earliest_transmission,
+                                 &schedule_peer_transmission,
+                                 pp);
     return;
   }
 #if INSANE_STATISTICS
-  GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# query plans executed"),
-                            1, GNUNET_NO);
+  GNUNET_STATISTICS_update (GSF_stats,
+                           gettext_noop ("# query plans executed"),
+                            1,
+                           GNUNET_NO);
 #endif
   /* process from priority heap */
-  rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap);
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Executing query plan %p\n", rp);
+  rp = GNUNET_CONTAINER_heap_remove_root (pp->priority_heap);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Executing query plan %p\n",
+              rp);
   GNUNET_assert (NULL != rp);
-  msize = GSF_pending_request_get_message_ (get_latest (rp), 0, NULL);
-  pp->pth =
-      GSF_peer_transmit_ (pp->cp, GNUNET_YES, rp->priority,
-                          GNUNET_TIME_UNIT_FOREVER_REL, msize,
-                          &transmit_message_callback, pp);
-  GNUNET_assert (NULL != pp->pth);
+  rp->hn = NULL;
+  rp->last_transmission = GNUNET_TIME_absolute_get ();
+  rp->transmission_counter++;
+  total_delay++;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Executing plan %p executed %u times, planning retransmission\n",
+              rp,
+             rp->transmission_counter);
+  GNUNET_assert (NULL == pp->env);
+  pp->env = GSF_pending_request_get_message_ (get_latest (rp));
+  GNUNET_MQ_notify_sent (pp->env,
+                        &schedule_peer_transmission,
+                        pp);
+  GSF_peer_transmit_ (pp->cp,
+                     GNUNET_YES,
+                     rp->priority,
+                     pp->env);
+  GNUNET_STATISTICS_update (GSF_stats,
+                            gettext_noop ("# query messages sent to other peers"),
+                            1,
+                            GNUNET_NO);
+  plan (pp,
+       rp);
 }
 
 
@@ -511,8 +477,14 @@ schedule_peer_transmission (void *cls,
 struct MergeContext
 {
 
+  /**
+   * Request we are trying to merge.
+   */
   struct GSF_PendingRequest *pr;
 
+  /**
+   * Set to #GNUNET_YES if we succeeded to merge.
+   */
   int merged;
 
 };
@@ -525,11 +497,13 @@ struct MergeContext
  * @param cls closure
  * @param query the query
  * @param element request plan stored at the node
- * @return GNUNET_YES if we should continue to iterate,
- *         GNUNET_NO if not (merge success)
+ * @return #GNUNET_YES if we should continue to iterate,
+ *         #GNUNET_NO if not (merge success)
  */
 static int
-merge_pr (void *cls, const struct GNUNET_HashCode * query, void *element)
+merge_pr (void *cls,
+          const struct GNUNET_HashCode *query,
+          void *element)
 {
   struct MergeContext *mpr = cls;
   struct GSF_RequestPlan *rp = element;
@@ -537,19 +511,30 @@ merge_pr (void *cls, const struct GNUNET_HashCode * query, void *element)
   struct GSF_PendingRequestPlanBijection *bi;
   struct GSF_PendingRequest *latest;
 
+  GNUNET_break (GNUNET_YES ==
+                GSF_pending_request_test_active_ (mpr->pr));
   if (GNUNET_OK !=
-      GSF_pending_request_is_compatible_ (mpr->pr, rp->pe_head->pr))
+      GSF_pending_request_is_compatible_ (mpr->pr,
+                                          rp->pe_head->pr))
     return GNUNET_YES;
   /* merge new request with existing request plan */
   bi = GNUNET_new (struct GSF_PendingRequestPlanBijection);
   bi->rp = rp;
   bi->pr = mpr->pr;
   prd = GSF_pending_request_get_data_ (mpr->pr);
-  GNUNET_CONTAINER_MDLL_insert (PR, prd->pr_head, prd->pr_tail, bi);
-  GNUNET_CONTAINER_MDLL_insert (PE, rp->pe_head, rp->pe_tail, bi);
+  GNUNET_CONTAINER_MDLL_insert (PR,
+                                prd->pr_head,
+                                prd->pr_tail,
+                                bi);
+  GNUNET_CONTAINER_MDLL_insert (PE,
+                                rp->pe_head,
+                                rp->pe_tail,
+                                bi);
   mpr->merged = GNUNET_YES;
 #if INSANE_STATISTICS
-  GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# requests merged"), 1,
+  GNUNET_STATISTICS_update (GSF_stats,
+                            gettext_noop ("# requests merged"),
+                            1,
                             GNUNET_NO);
 #endif
   latest = get_latest (rp);
@@ -557,8 +542,10 @@ merge_pr (void *cls, const struct GNUNET_HashCode * query, void *element)
       prd->ttl.abs_value_us)
   {
 #if INSANE_STATISTICS
-    GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# requests refreshed"),
-                              1, GNUNET_NO);
+    GNUNET_STATISTICS_update (GSF_stats,
+                              gettext_noop ("# requests refreshed"),
+                              1,
+                              GNUNET_NO);
 #endif
     rp->transmission_counter = 0;       /* reset */
   }
@@ -583,6 +570,8 @@ GSF_plan_add_ (struct GSF_ConnectedPeer *cp,
   struct GSF_PendingRequestPlanBijection *bi;
   struct MergeContext mpc;
 
+  GNUNET_assert (GNUNET_YES ==
+                 GSF_pending_request_test_active_ (pr));
   GNUNET_assert (NULL != cp);
   id = GSF_connected_peer_get_identity2_ (cp);
   pp = GNUNET_CONTAINER_multipeermap_get (plans, id);
@@ -597,40 +586,50 @@ GSF_plan_add_ (struct GSF_ConnectedPeer *cp,
     pp->cp = cp;
     GNUNET_assert (GNUNET_OK ==
                    GNUNET_CONTAINER_multipeermap_put (plans,
-                                                      id, pp,
+                                                      id,
+                                                      pp,
                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission,
+                                        pp);
   }
   mpc.merged = GNUNET_NO;
   mpc.pr = pr;
+  prd = GSF_pending_request_get_data_ (pr);
   GNUNET_CONTAINER_multihashmap_get_multiple (pp->plan_map,
-                                              &GSF_pending_request_get_data_
-                                              (pr)->query, &merge_pr, &mpc);
-  if (GNUNET_NO != mpc.merged)
-    return;
-  GNUNET_CONTAINER_multihashmap_get_multiple (pp->plan_map,
-                                              &GSF_pending_request_get_data_
-                                              (pr)->query, &merge_pr, &mpc);
+                                              &prd->query,
+                                              &merge_pr,
+                                              &mpc);
   if (GNUNET_NO != mpc.merged)
     return;
   plan_count++;
-  GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# query plan entries"), 1,
+  GNUNET_STATISTICS_update (GSF_stats,
+                            gettext_noop ("# query plan entries"),
+                            1,
                             GNUNET_NO);
-  prd = GSF_pending_request_get_data_ (pr);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Planning transmission of query `%s' to peer `%s'\n",
-              GNUNET_h2s (&prd->query), GNUNET_i2s (id));
+              GNUNET_h2s (&prd->query),
+              GNUNET_i2s (id));
   rp = GNUNET_new (struct GSF_RequestPlan);
   bi = GNUNET_new (struct GSF_PendingRequestPlanBijection);
   bi->rp = rp;
   bi->pr = pr;
-  GNUNET_CONTAINER_MDLL_insert (PR, prd->pr_head, prd->pr_tail, bi);
-  GNUNET_CONTAINER_MDLL_insert (PE, rp->pe_head, rp->pe_tail, bi);
+  GNUNET_CONTAINER_MDLL_insert (PR,
+                                prd->pr_head,
+                                prd->pr_tail,
+                                bi);
+  GNUNET_CONTAINER_MDLL_insert (PE,
+                                rp->pe_head,
+                                rp->pe_tail,
+                                bi);
   rp->pp = pp;
   GNUNET_assert (GNUNET_YES ==
                  GNUNET_CONTAINER_multihashmap_put (pp->plan_map,
-                                                    get_rp_key (rp), rp,
+                                                    get_rp_key (rp),
+                                                    rp,
                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
-  plan (pp, rp);
+  plan (pp,
+        rp);
 }
 
 
@@ -656,26 +655,28 @@ GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp)
   GNUNET_assert (GNUNET_YES ==
                  GNUNET_CONTAINER_multipeermap_remove (plans, id,
                                                        pp));
-  if (NULL != pp->pth)
-  {
-    GSF_peer_transmit_cancel_ (pp->pth);
-    pp->pth = NULL;
-  }
-  if (GNUNET_SCHEDULER_NO_TASK != pp->task)
+  if (NULL != pp->task)
   {
     GNUNET_SCHEDULER_cancel (pp->task);
-    pp->task = GNUNET_SCHEDULER_NO_TASK;
+    pp->task = NULL;
   }
   while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->priority_heap)))
   {
     GNUNET_break (GNUNET_YES ==
                   GNUNET_CONTAINER_multihashmap_remove (pp->plan_map,
-                                                        get_rp_key (rp), rp));
+                                                        get_rp_key (rp),
+                                                        rp));
     while (NULL != (bi = rp->pe_head))
     {
-      GNUNET_CONTAINER_MDLL_remove (PE, rp->pe_head, rp->pe_tail, bi);
+      GNUNET_CONTAINER_MDLL_remove (PE,
+                                    rp->pe_head,
+                                    rp->pe_tail,
+                                    bi);
       prd = GSF_pending_request_get_data_ (bi->pr);
-      GNUNET_CONTAINER_MDLL_remove (PR, prd->pr_head, prd->pr_tail, bi);
+      GNUNET_CONTAINER_MDLL_remove (PR,
+                                    prd->pr_head,
+                                    prd->pr_tail,
+                                    bi);
       GNUNET_free (bi);
     }
     plan_count--;
@@ -686,19 +687,28 @@ GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp)
   {
     GNUNET_break (GNUNET_YES ==
                   GNUNET_CONTAINER_multihashmap_remove (pp->plan_map,
-                                                        get_rp_key (rp), rp));
+                                                        get_rp_key (rp),
+                                                        rp));
     while (NULL != (bi = rp->pe_head))
     {
       prd = GSF_pending_request_get_data_ (bi->pr);
-      GNUNET_CONTAINER_MDLL_remove (PE, rp->pe_head, rp->pe_tail, bi);
-      GNUNET_CONTAINER_MDLL_remove (PR, prd->pr_head, prd->pr_tail, bi);
+      GNUNET_CONTAINER_MDLL_remove (PE,
+                                    rp->pe_head,
+                                    rp->pe_tail,
+                                    bi);
+      GNUNET_CONTAINER_MDLL_remove (PR,
+                                    prd->pr_head,
+                                    prd->pr_tail,
+                                    bi);
       GNUNET_free (bi);
     }
     plan_count--;
     GNUNET_free (rp);
   }
-  GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# query plan entries"),
-                         plan_count, GNUNET_NO);
+  GNUNET_STATISTICS_set (GSF_stats,
+                         gettext_noop ("# query plan entries"),
+                         plan_count,
+                         GNUNET_NO);
   GNUNET_CONTAINER_heap_destroy (pp->delay_heap);
   GNUNET_CONTAINER_multihashmap_destroy (pp->plan_map);
   GNUNET_free (pp);
@@ -707,17 +717,17 @@ GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp)
 
 /**
  * Get the last transmission attempt time for the request plan list
- * referenced by 'pr_head', that was sent to 'sender'
+ * referenced by @a pr_head, that was sent to @a sender
  *
  * @param pr_head request plan reference list to check.
  * @param sender the peer that we've sent the request to.
- * @param result the timestamp to fill, set to "FOREVER" if never transmitted
- * @return GNUNET_YES if 'result' was changed, GNUNET_NO otherwise.
+ * @param result the timestamp to fill, set to #GNUNET_TIME_UNIT_FOREVER_ABS if never transmitted
+ * @return #GNUNET_YES if @a result was changed, #GNUNET_NO otherwise.
  */
 int
-GSF_request_plan_reference_get_last_transmission_ (
-    struct GSF_PendingRequestPlanBijection *pr_head, struct GSF_ConnectedPeer *sender,
-    struct GNUNET_TIME_Absolute *result)
+GSF_request_plan_reference_get_last_transmission_ (struct GSF_PendingRequestPlanBijection *pr_head,
+                                                   struct GSF_ConnectedPeer *sender,
+                                                   struct GNUNET_TIME_Absolute *result)
 {
   struct GSF_PendingRequestPlanBijection *bi;
 
@@ -753,23 +763,31 @@ GSF_plan_notify_request_done_ (struct GSF_PendingRequest *pr)
   while (NULL != (bi = prd->pr_head))
   {
     rp = bi->rp;
-    GNUNET_CONTAINER_MDLL_remove (PR, prd->pr_head, prd->pr_tail, bi);
-    GNUNET_CONTAINER_MDLL_remove (PE, rp->pe_head, rp->pe_tail, bi);
+    GNUNET_CONTAINER_MDLL_remove (PR,
+                                  prd->pr_head,
+                                  prd->pr_tail,
+                                  bi);
+    GNUNET_CONTAINER_MDLL_remove (PE,
+                                  rp->pe_head,
+                                  rp->pe_tail,
+                                  bi);
+    GNUNET_assert (bi->pr == pr);
     if (NULL == rp->pe_head)
     {
       GNUNET_CONTAINER_heap_remove_node (rp->hn);
       plan_count--;
       GNUNET_break (GNUNET_YES ==
                     GNUNET_CONTAINER_multihashmap_remove (rp->pp->plan_map,
-                                                         &GSF_pending_request_get_data_
-                                                         (bi->pr)->query,
+                                                         &prd->query,
                                                           rp));
       GNUNET_free (rp);
     }
     GNUNET_free (bi);
   }
-  GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# query plan entries"),
-                         plan_count, GNUNET_NO);
+  GNUNET_STATISTICS_set (GSF_stats,
+                         gettext_noop ("# query plan entries"),
+                         plan_count,
+                         GNUNET_NO);
 }
 
 
@@ -779,7 +797,8 @@ GSF_plan_notify_request_done_ (struct GSF_PendingRequest *pr)
 void
 GSF_plan_init ()
 {
-  plans = GNUNET_CONTAINER_multipeermap_create (256, GNUNET_YES);
+  plans = GNUNET_CONTAINER_multipeermap_create (256,
+                                                GNUNET_YES);
 }