src/cadet/gnunet-service-cadet_peer.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2001-2017 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 cadet/gnunet-service-cadet_peer.c
  23  * @brief Information we track per peer.
  24  * @author Bartlomiej Polot
  25  * @author Christian Grothoff
  26  *
  27  * TODO:
  28  * - optimize stopping/restarting DHT search to situations
  29  *   where we actually need it (i.e. not if we have a direct connection,
  30  *   or if we already have plenty of good short ones, or maybe even
  31  *   to take a break if we have some connections and have searched a lot (?))
  32  */
  33 #include "platform.h"
  34 #include "gnunet_util_lib.h"
  35 #include "gnunet_hello_lib.h"
  36 #include "gnunet_signatures.h"
  37 #include "gnunet_transport_service.h"
  38 #include "gnunet_ats_service.h"
  39 #include "gnunet_core_service.h"
  40 #include "gnunet_statistics_service.h"
  41 #include "cadet_protocol.h"
  42 #include "gnunet-service-cadet_connection.h"
  43 #include "gnunet-service-cadet_dht.h"
  44 #include "gnunet-service-cadet_peer.h"
  45 #include "gnunet-service-cadet_paths.h"
  46 #include "gnunet-service-cadet_tunnels.h"
  47
  48
  49 #define LOG(level, ...) GNUNET_log_from(level,"cadet-per",__VA_ARGS__)
  50
  51
  52 /**
  53  * How long do we wait until tearing down an idle peer?
  54  */
  55 #define IDLE_PEER_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 5)
  56
  57 /**
  58  * How long do we keep paths around if we no longer care about the peer?
  59  */
  60 #define IDLE_PATH_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
  61
  62 /**
  63  * Queue size when we start dropping OOO messages.
  64  */
  65 #define MAX_OOO_QUEUE_SIZE  100
  66
  67
  68 /**
  69  * Data structure used to track whom we have to notify about changes
  70  * to our message queue.
  71  */
  72 struct GCP_MessageQueueManager
  73 {
  74
  75   /**
  76    * Kept in a DLL.
  77    */
  78   struct GCP_MessageQueueManager *next;
  79
  80   /**
  81    * Kept in a DLL.
  82    */
  83   struct GCP_MessageQueueManager *prev;
  84
  85   /**
  86    * Function to call with updated message queue object.
  87    */
  88   GCP_MessageQueueNotificationCallback cb;
  89
  90   /**
  91    * Closure for @e cb.
  92    */
  93   void *cb_cls;
  94
  95   /**
  96    * The peer this is for.
  97    */
  98   struct CadetPeer *cp;
  99
 100   /**
 101    * Envelope this manager would like to transmit once it is its turn.
 102    */
 103   struct GNUNET_MQ_Envelope *env;
 104
 105 };
 106
 107
 108 /**
 109  * Struct containing all information regarding a given peer
 110  */
 111 struct CadetPeer
 112 {
 113   /**
 114    * ID of the peer
 115    */
 116   struct GNUNET_PeerIdentity pid;
 117
 118   /**
 119    * Last time we heard from this peer (currently not used!)
 120    */
 121   struct GNUNET_TIME_Absolute last_contactXXX;
 122
 123   /**
 124    * Array of DLLs of paths traversing the peer, organized by the
 125    * offset of the peer on the larger path.
 126    */
 127   struct CadetPeerPathEntry **path_heads;
 128
 129   /**
 130    * Array of DLL of paths traversing the peer, organized by the
 131    * offset of the peer on the larger path.
 132    */
 133   struct CadetPeerPathEntry **path_tails;
 134
 135   /**
 136    * Notifications to call when @e core_mq changes.
 137    */
 138   struct GCP_MessageQueueManager *mqm_head;
 139
 140   /**
 141    * Notifications to call when @e core_mq changes.
 142    */
 143   struct GCP_MessageQueueManager *mqm_tail;
 144
 145   /**
 146    * Pointer to first "ready" entry in @e mqm_head.
 147    */
 148   struct GCP_MessageQueueManager *mqm_ready_ptr;
 149
 150   /**
 151    * MIN-heap of paths owned by this peer (they also end at this
 152    * peer).  Ordered by desirability.
 153    */
 154   struct GNUNET_CONTAINER_Heap *path_heap;
 155
 156   /**
 157    * Handle to stop the DHT search for paths to this peer
 158    */
 159   struct GCD_search_handle *search_h;
 160
 161   /**
 162    * Task to clean up @e path_heap asynchronously.
 163    */
 164   struct GNUNET_SCHEDULER_Task *heap_cleanup_task;
 165
 166   /**
 167    * Task to destroy this entry.
 168    */
 169   struct GNUNET_SCHEDULER_Task *destroy_task;
 170
 171   /**
 172    * Tunnel to this peer, if any.
 173    */
 174   struct CadetTunnel *t;
 175
 176   /**
 177    * Connections that go through this peer; indexed by tid.
 178    */
 179   struct GNUNET_CONTAINER_MultiShortmap *connections;
 180
 181   /**
 182    * Handle for core transmissions.
 183    */
 184   struct GNUNET_MQ_Handle *core_mq;
 185
 186   /**
 187    * Hello message of the peer.
 188    */
 189   struct GNUNET_HELLO_Message *hello;
 190
 191   /**
 192    * Handle to us offering the HELLO to the transport.
 193    */
 194   struct GNUNET_TRANSPORT_OfferHelloHandle *hello_offer;
 195
 196   /**
 197    * Handle to our ATS request asking ATS to suggest an address
 198    * to TRANSPORT for this peer (to establish a direct link).
 199    */
 200   struct GNUNET_ATS_ConnectivitySuggestHandle *connectivity_suggestion;
 201
 202   /**
 203    * How many messages are in the queue to this peer.
 204    */
 205   unsigned int queue_n;
 206
 207   /**
 208    * How many paths do we have to this peer (in all @e path_heads DLLs combined).
 209    */
 210   unsigned int num_paths;
 211
 212   /**
 213    * Sum over all of the offsets of all of the paths in the @a path_heads DLLs.
 214    * Used to speed-up @GCP_get_desirability_of_path() calculation.
 215    */
 216   unsigned int off_sum;
 217
 218   /**
 219    * Number of message queue managers of this peer that have a message in waiting.
 220    *
 221    * Used to quickly see if we need to bother scanning the @e msm_head DLL.
 222    * TODO: could be replaced by another DLL that would then allow us to avoid
 223    * the O(n)-scan of the DLL for ready entries!
 224    */
 225   unsigned int mqm_ready_counter;
 226
 227   /**
 228    * Current length of the @e path_heads and @path_tails arrays.
 229    * The arrays should be grown as needed.
 230    */
 231   unsigned int path_dll_length;
 232
 233 };
 234
 235
 236 /**
 237  * Get the static string for a peer ID.
 238  *
 239  * @param cp Peer.
 240  * @return Static string for it's ID.
 241  */
 242 const char *
 243 GCP_2s (const struct CadetPeer *cp)
 244 {
 245   static char buf[5];
 246   char *ret;
 247
 248   ret = GNUNET_CRYPTO_eddsa_public_key_to_string (&cp->pid.public_key);
 249   strncpy (buf,
 250            ret,
 251            sizeof (buf) - 1);
 252   GNUNET_free (ret);
 253   buf[4] = '\0';
 254   return buf;
 255 }
 256
 257
 258 /**
 259  * Calculate how desirable a path is for @a cp if @a cp
 260  * is at offset @a off.
 261  *
 262  * The 'desirability_table.c' program can be used to compute a list of
 263  * sample outputs for different scenarios.  Basically, we score paths
 264  * lower if there are many alternatives, and higher if they are
 265  * shorter than average, and very high if they are much shorter than
 266  * average and without many alternatives.
 267  *
 268  * @param cp a peer reachable via a path
 269  * @param off offset of @a cp in the path
 270  * @return score how useful a path is to reach @a cp,
 271  *         positive scores mean path is more desirable
 272  */
 273 double
 274 GCP_get_desirability_of_path (struct CadetPeer *cp,
 275                               unsigned int off)
 276 {
 277   unsigned int num_alts = cp->num_paths;
 278   unsigned int off_sum;
 279   double avg_sum;
 280   double path_delta;
 281   double weight_alts;
 282
 283   GNUNET_assert (num_alts >= 1); /* 'path' should be in there! */
 284   GNUNET_assert (0 != cp->path_dll_length);
 285
 286   /* We maintain 'off_sum' in 'peer' and thereby
 287      avoid the SLOW recalculation each time. Kept here
 288      just to document what is going on. */
 289 #if SLOW
 290   off_sum = 0;
 291   for (unsigned int j=0;j<cp->path_dll_length;j++)
 292     for (struct CadetPeerPathEntry *pe = cp->path_heads[j];
 293          NULL != pe;
 294          pe = pe->next)
 295       off_sum += j;
 296   GNUNET_assert (off_sum == cp->off_sum);
 297 #else
 298   off_sum = cp->off_sum;
 299 #endif
 300   avg_sum = off_sum * 1.0 / cp->path_dll_length;
 301   path_delta = off - avg_sum;
 302   /* path_delta positiv: path off of peer above average (bad path for peer),
 303      path_delta negativ: path off of peer below average (good path for peer) */
 304   if (path_delta <= - 1.0)
 305     weight_alts = - num_alts / path_delta; /* discount alternative paths */
 306   else if (path_delta >= 1.0)
 307     weight_alts = num_alts * path_delta; /* overcount alternative paths */
 308   else
 309     weight_alts = num_alts; /* count alternative paths normally */
 310
 311
 312   /* off+1: long paths are generally harder to find and thus count
 313      a bit more as they get longer.  However, above-average paths
 314      still need to count less, hence the squaring of that factor. */
 315   return (off + 1.0) / (weight_alts * weight_alts);
 316 }
 317
 318
 319 /**
 320  * This peer is no longer be needed, clean it up now.
 321  *
 322  * @param cls peer to clean up
 323  */
 324 static void
 325 destroy_peer (void *cls)
 326 {
 327   struct CadetPeer *cp = cls;
 328
 329   LOG (GNUNET_ERROR_TYPE_DEBUG,
 330        "Destroying state about peer %s\n",
 331        GCP_2s (cp));
 332   cp->destroy_task = NULL;
 333   GNUNET_assert (NULL == cp->t);
 334   GNUNET_assert (NULL == cp->core_mq);
 335   GNUNET_assert (0 == cp->num_paths);
 336   for (unsigned int i=0;i<cp->path_dll_length;i++)
 337     GNUNET_assert (NULL == cp->path_heads[i]);
 338   GNUNET_assert (0 == GNUNET_CONTAINER_multishortmap_size (cp->connections));
 339   GNUNET_assert (GNUNET_YES ==
 340                  GNUNET_CONTAINER_multipeermap_remove (peers,
 341                                                        &cp->pid,
 342                                                        cp));
 343   GNUNET_free_non_null (cp->path_heads);
 344   GNUNET_free_non_null (cp->path_tails);
 345   cp->path_dll_length = 0;
 346   if (NULL != cp->search_h)
 347   {
 348     GCD_search_stop (cp->search_h);
 349     cp->search_h = NULL;
 350   }
 351   /* FIXME: clean up search_delayedXXX! */
 352
 353   if (NULL != cp->hello_offer)
 354   {
 355     GNUNET_TRANSPORT_offer_hello_cancel (cp->hello_offer);
 356     cp->hello_offer = NULL;
 357   }
 358   if (NULL != cp->connectivity_suggestion)
 359   {
 360     GNUNET_ATS_connectivity_suggest_cancel (cp->connectivity_suggestion);
 361     cp->connectivity_suggestion = NULL;
 362   }
 363   GNUNET_CONTAINER_multishortmap_destroy (cp->connections);
 364   if (NULL != cp->path_heap)
 365   {
 366     GNUNET_CONTAINER_heap_destroy (cp->path_heap);
 367     cp->path_heap = NULL;
 368   }
 369   if (NULL != cp->heap_cleanup_task)
 370   {
 371     GNUNET_SCHEDULER_cancel (cp->heap_cleanup_task);
 372     cp->heap_cleanup_task = NULL;
 373   }
 374   GNUNET_free_non_null (cp->hello);
 375   /* Peer should not be freed if paths exist; if there are no paths,
 376      there ought to be no connections, and without connections, no
 377      notifications. Thus we can assert that mqm_head is empty at this
 378      point. */
 379   GNUNET_assert (NULL == cp->mqm_head);
 380   GNUNET_assert (NULL == cp->mqm_ready_ptr);
 381   GNUNET_free (cp);
 382 }
 383
 384
 385 /**
 386  * This peer is now on more "active" duty, activate processes related to it.
 387  *
 388  * @param cp the more-active peer
 389  */
 390 static void
 391 consider_peer_activate (struct CadetPeer *cp)
 392 {
 393   uint32_t strength;
 394
 395   LOG (GNUNET_ERROR_TYPE_DEBUG,
 396        "Updating peer %s activation state (%u connections)%s%s\n",
 397        GCP_2s (cp),
 398        GNUNET_CONTAINER_multishortmap_size (cp->connections),
 399        (NULL == cp->t) ? "" : " with tunnel",
 400        (NULL == cp->core_mq) ? "" : " with CORE link");
 401   if (NULL != cp->destroy_task)
 402   {
 403     /* It's active, do not destory! */
 404     GNUNET_SCHEDULER_cancel (cp->destroy_task);
 405     cp->destroy_task = NULL;
 406   }
 407   if ( (0 == GNUNET_CONTAINER_multishortmap_size (cp->connections)) &&
 408        (NULL == cp->t) )
 409   {
 410     /* We're just on a path or directly connected; don't bother too much */
 411     if (NULL != cp->connectivity_suggestion)
 412     {
 413       GNUNET_ATS_connectivity_suggest_cancel (cp->connectivity_suggestion);
 414       cp->connectivity_suggestion = NULL;
 415     }
 416     if (NULL != cp->search_h)
 417     {
 418       GCD_search_stop (cp->search_h);
 419       cp->search_h = NULL;
 420     }
 421     return;
 422   }
 423   if (NULL == cp->core_mq)
 424   {
 425     /* Lacks direct connection, try to create one by querying the DHT */
 426     if ( (NULL == cp->search_h) &&
 427          (DESIRED_CONNECTIONS_PER_TUNNEL > cp->num_paths) )
 428       cp->search_h
 429         = GCD_search (&cp->pid);
 430   }
 431   else
 432   {
 433     /* Have direct connection, stop DHT search if active */
 434     if (NULL != cp->search_h)
 435     {
 436       GCD_search_stop (cp->search_h);
 437       cp->search_h = NULL;
 438     }
 439   }
 440
 441   /* If we have a tunnel, our urge for connections is much bigger */
 442   strength = (NULL != cp->t) ? 32 : 1;
 443   if (NULL != cp->connectivity_suggestion)
 444     GNUNET_ATS_connectivity_suggest_cancel (cp->connectivity_suggestion);
 445   cp->connectivity_suggestion
 446     = GNUNET_ATS_connectivity_suggest (ats_ch,
 447                                        &cp->pid,
 448                                        strength);
 449 }
 450
 451
 452 /**
 453  * This peer may no longer be needed, consider cleaning it up.
 454  *
 455  * @param cp peer to clean up
 456  */
 457 static void
 458 consider_peer_destroy (struct CadetPeer *cp);
 459
 460
 461 /**
 462  * We really no longere care about a peer, stop hogging memory with paths to it.
 463  * Afterwards, see if there is more to be cleaned up about this peer.
 464  *
 465  * @param cls a `struct CadetPeer`.
 466  */
 467 static void
 468 drop_paths (void *cls)
 469 {
 470   struct CadetPeer *cp = cls;
 471   struct CadetPeerPath *path;
 472
 473   cp->destroy_task = NULL;
 474   while (NULL != (path = GNUNET_CONTAINER_heap_remove_root (cp->path_heap)))
 475     GCPP_release (path);
 476   consider_peer_destroy (cp);
 477 }
 478
 479
 480 /**
 481  * This peer may no longer be needed, consider cleaning it up.
 482  *
 483  * @param cp peer to clean up
 484  */
 485 static void
 486 consider_peer_destroy (struct CadetPeer *cp)
 487 {
 488   struct GNUNET_TIME_Relative exp;
 489
 490   if (NULL != cp->destroy_task)
 491   {
 492     GNUNET_SCHEDULER_cancel (cp->destroy_task);
 493     cp->destroy_task = NULL;
 494   }
 495   if (NULL != cp->t)
 496     return; /* still relevant! */
 497   if (NULL != cp->core_mq)
 498     return; /* still relevant! */
 499   if (0 != GNUNET_CONTAINER_multishortmap_size (cp->connections))
 500     return; /* still relevant! */
 501   if ( (NULL != cp->path_heap) &&
 502        (0 < GNUNET_CONTAINER_heap_get_size (cp->path_heap)) )
 503   {
 504     cp->destroy_task = GNUNET_SCHEDULER_add_delayed (IDLE_PATH_TIMEOUT,
 505                                                      &drop_paths,
 506                                                      cp);
 507     return;
 508   }
 509   if (0 != cp->num_paths)
 510     return; /* still relevant! */
 511   if (NULL != cp->hello)
 512   {
 513     /* relevant only until HELLO expires */
 514     exp = GNUNET_TIME_absolute_get_remaining (GNUNET_HELLO_get_last_expiration (cp->hello));
 515     cp->destroy_task = GNUNET_SCHEDULER_add_delayed (exp,
 516                                                      &destroy_peer,
 517                                                      cp);
 518     return;
 519   }
 520   cp->destroy_task = GNUNET_SCHEDULER_add_delayed (IDLE_PEER_TIMEOUT,
 521                                                    &destroy_peer,
 522                                                    cp);
 523 }
 524
 525
 526 /**
 527  * Set the message queue to @a mq for peer @a cp and notify watchers.
 528  *
 529  * @param cp peer to modify
 530  * @param mq message queue to set (can be NULL)
 531  */
 532 void
 533 GCP_set_mq (struct CadetPeer *cp,
 534             struct GNUNET_MQ_Handle *mq)
 535 {
 536   LOG (GNUNET_ERROR_TYPE_DEBUG,
 537        "Message queue for peer %s is now %p\n",
 538        GCP_2s (cp),
 539        mq);
 540   cp->core_mq = mq;
 541   for (struct GCP_MessageQueueManager *mqm = cp->mqm_head, *next;
 542        NULL != mqm;
 543        mqm = next)
 544   {
 545     /* Save next pointer in case mqm gets freed by the callback */
 546     next = mqm->next;
 547     if (NULL == mq)
 548     {
 549       if (NULL != mqm->env)
 550       {
 551         GNUNET_MQ_discard (mqm->env);
 552         mqm->env = NULL;
 553         mqm->cb (mqm->cb_cls,
 554                  GNUNET_SYSERR);
 555       }
 556       else
 557       {
 558         mqm->cb (mqm->cb_cls,
 559                  GNUNET_NO);
 560       }
 561     }
 562     else
 563     {
 564       GNUNET_assert (NULL == mqm->env);
 565       mqm->cb (mqm->cb_cls,
 566                GNUNET_YES);
 567     }
 568   }
 569   if ( (NULL != mq) ||
 570        (NULL != cp->t) )
 571     consider_peer_activate (cp);
 572   else
 573     consider_peer_destroy (cp);
 574
 575   if ( (NULL != mq) &&
 576        (NULL != cp->t) )
 577   {
 578     /* have a new, direct path to the target, notify tunnel */
 579     struct CadetPeerPath *path;
 580
 581     path = GCPP_get_path_from_route (1,
 582                                      &cp->pid);
 583     GCT_consider_path (cp->t,
 584                        path,
 585                        0);
 586   }
 587 }
 588
 589
 590 /**
 591  * Debug function should NEVER return true in production code, useful to
 592  * simulate losses for testcases.
 593  *
 594  * @return #GNUNET_YES or #GNUNET_NO with the decision to drop.
 595  */
 596 static int
 597 should_I_drop (void)
 598 {
 599   if (0 == drop_percent)
 600     return GNUNET_NO;
 601   if (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 602                                 101) < drop_percent)
 603     return GNUNET_YES;
 604   return GNUNET_NO;
 605 }
 606
 607
 608 /**
 609  * Function called when CORE took one of the messages from
 610  * a message queue manager and transmitted it.
 611  *
 612  * @param cls the `struct CadetPeeer` where we made progress
 613  */
 614 static void
 615 mqm_send_done (void *cls);
 616
 617
 618 /**
 619  * Transmit current envelope from this @a mqm.
 620  *
 621  * @param mqm mqm to transmit message for now
 622  */
 623 static void
 624 mqm_execute (struct GCP_MessageQueueManager *mqm)
 625 {
 626   struct CadetPeer *cp = mqm->cp;
 627
 628   /* Move ready pointer to the next entry that might be ready. */
 629   if ( (mqm == cp->mqm_ready_ptr) &&
 630        (NULL != mqm->next) )
 631     cp->mqm_ready_ptr = mqm->next;
 632   /* Move entry to the end of the DLL, to be fair. */
 633   if (mqm != cp->mqm_tail)
 634   {
 635     GNUNET_CONTAINER_DLL_remove (cp->mqm_head,
 636                                  cp->mqm_tail,
 637                                  mqm);
 638     GNUNET_CONTAINER_DLL_insert_tail (cp->mqm_head,
 639                                       cp->mqm_tail,
 640                                       mqm);
 641   }
 642   cp->mqm_ready_counter--;
 643   if (GNUNET_YES == should_I_drop ())
 644   {
 645     LOG (GNUNET_ERROR_TYPE_DEBUG,
 646          "DROPPING message to peer %s from MQM %p\n",
 647          GCP_2s (cp),
 648          mqm);
 649     GNUNET_MQ_discard (mqm->env);
 650     mqm->env = NULL;
 651     mqm_send_done (cp);
 652   }
 653   else
 654   {
 655     {
 656       const struct GNUNET_MessageHeader *mh;
 657
 658       mh = GNUNET_MQ_env_get_msg (mqm->env);
 659       switch (ntohs (mh->type))
 660       {
 661       case GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX:
 662         {
 663           const struct GNUNET_CADET_TunnelKeyExchangeMessage *msg
 664             = (const struct GNUNET_CADET_TunnelKeyExchangeMessage *) mh;
 665           LOG (GNUNET_ERROR_TYPE_DEBUG,
 666                "P2P forwarding KX with ephemeral %s to %s on CID %s\n",
 667                GNUNET_e2s (&msg->ephemeral_key),
 668                GCP_2s (cp),
 669                GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 670         }
 671         break;
 672       default:
 673         break;
 674       }
 675     }
 676     LOG (GNUNET_ERROR_TYPE_DEBUG,
 677          "Sending to peer %s from MQM %p\n",
 678          GCP_2s (cp),
 679          mqm);
 680     GNUNET_MQ_send (cp->core_mq,
 681                     mqm->env);
 682     mqm->env = NULL;
 683   }
 684   mqm->cb (mqm->cb_cls,
 685            GNUNET_YES);
 686 }
 687
 688
 689 /**
 690  * Find the next ready message in the queue (starting
 691  * the search from the `cp->mqm_ready_ptr`) and if possible
 692  * execute the transmission.
 693  *
 694  * @param cp peer to try to send the next ready message to
 695  */
 696 static void
 697 send_next_ready (struct CadetPeer *cp)
 698 {
 699   struct GCP_MessageQueueManager *mqm;
 700
 701   if (0 == cp->mqm_ready_counter)
 702     return;
 703   while ( (NULL != (mqm = cp->mqm_ready_ptr)) &&
 704           (NULL == mqm->env) )
 705     cp->mqm_ready_ptr = mqm->next;
 706   if (NULL == mqm)
 707     return; /* nothing to do */
 708   mqm_execute (mqm);
 709 }
 710
 711
 712 /**
 713  * Function called when CORE took one of the messages from
 714  * a message queue manager and transmitted it.
 715  *
 716  * @param cls the `struct CadetPeeer` where we made progress
 717  */
 718 static void
 719 mqm_send_done (void *cls)
 720 {
 721   struct CadetPeer *cp = cls;
 722
 723   LOG (GNUNET_ERROR_TYPE_DEBUG,
 724        "Sending to peer %s completed\n",
 725        GCP_2s (cp));
 726   send_next_ready (cp);
 727 }
 728
 729
 730 /**
 731  * Send the message in @a env to @a cp.
 732  *
 733  * @param mqm the message queue manager to use for transmission
 734  * @param env envelope with the message to send; must NOT
 735  *            yet have a #GNUNET_MQ_notify_sent() callback attached to it
 736  */
 737 void
 738 GCP_send (struct GCP_MessageQueueManager *mqm,
 739           struct GNUNET_MQ_Envelope *env)
 740 {
 741   struct CadetPeer *cp = mqm->cp;
 742
 743   GNUNET_assert (NULL != env);
 744   LOG (GNUNET_ERROR_TYPE_DEBUG,
 745        "Queueing message to peer %s in MQM %p\n",
 746        GCP_2s (cp),
 747        mqm);
 748   GNUNET_assert (NULL != cp->core_mq);
 749   GNUNET_assert (NULL == mqm->env);
 750   GNUNET_MQ_notify_sent (env,
 751                          &mqm_send_done,
 752                          cp);
 753   mqm->env = env;
 754   cp->mqm_ready_counter++;
 755   if (mqm != cp->mqm_ready_ptr)
 756     cp->mqm_ready_ptr = cp->mqm_head;
 757   if (1 == cp->mqm_ready_counter)
 758     cp->mqm_ready_ptr = mqm;
 759   if (0 != GNUNET_MQ_get_length (cp->core_mq))
 760     return;
 761   send_next_ready (cp);
 762 }
 763
 764
 765 /**
 766  * Function called to destroy a peer now.
 767  *
 768  * @param cls NULL
 769  * @param pid identity of the peer (unused)
 770  * @param value the `struct CadetPeer` to clean up
 771  * @return #GNUNET_OK (continue to iterate)
 772  */
 773 static int
 774 destroy_iterator_cb (void *cls,
 775                      const struct GNUNET_PeerIdentity *pid,
 776                      void *value)
 777 {
 778   struct CadetPeer *cp = value;
 779
 780   if (NULL != cp->destroy_task)
 781   {
 782     GNUNET_SCHEDULER_cancel (cp->destroy_task);
 783     cp->destroy_task = NULL;
 784   }
 785   destroy_peer (cp);
 786   return GNUNET_OK;
 787 }
 788
 789
 790 /**
 791  * Clean up all entries about all peers.
 792  * Must only be called after all tunnels, CORE-connections and
 793  * connections are down.
 794  */
 795 void
 796 GCP_destroy_all_peers ()
 797 {
 798   LOG (GNUNET_ERROR_TYPE_DEBUG,
 799        "Destroying all peers now\n");
 800   GNUNET_CONTAINER_multipeermap_iterate (peers,
 801                                          &destroy_iterator_cb,
 802                                          NULL);
 803 }
 804
 805
 806 /**
 807  * Drop all paths owned by this peer, and do not
 808  * allow new ones to be added: We are shutting down.
 809  *
 810  * @param cp peer to drop paths to
 811  */
 812 void
 813 GCP_drop_owned_paths (struct CadetPeer *cp)
 814 {
 815   struct CadetPeerPath *path;
 816
 817   LOG (GNUNET_ERROR_TYPE_DEBUG,
 818        "Destroying all paths to %s\n",
 819        GCP_2s (cp));
 820   while (NULL != (path =
 821                   GNUNET_CONTAINER_heap_remove_root (cp->path_heap)))
 822     GCPP_release (path);
 823   GNUNET_CONTAINER_heap_destroy (cp->path_heap);
 824   cp->path_heap = NULL;
 825 }
 826
 827
 828 /**
 829  * Add an entry to the DLL of all of the paths that this peer is on.
 830  *
 831  * @param cp peer to modify
 832  * @param entry an entry on a path
 833  * @param off offset of this peer on the path
 834  */
 835 void
 836 GCP_path_entry_add (struct CadetPeer *cp,
 837                     struct CadetPeerPathEntry *entry,
 838                     unsigned int off)
 839 {
 840   GNUNET_assert (cp == GCPP_get_peer_at_offset (entry->path,
 841                                                 off));
 842   LOG (GNUNET_ERROR_TYPE_DEBUG,
 843        "Discovered that peer %s is on path %s at offset %u\n",
 844        GCP_2s (cp),
 845        GCPP_2s (entry->path),
 846        off);
 847   if (off >= cp->path_dll_length)
 848   {
 849     unsigned int len = cp->path_dll_length;
 850
 851     GNUNET_array_grow (cp->path_heads,
 852                        len,
 853                        off + 4);
 854     GNUNET_array_grow (cp->path_tails,
 855                        cp->path_dll_length,
 856                        off + 4);
 857   }
 858   GNUNET_CONTAINER_DLL_insert (cp->path_heads[off],
 859                                cp->path_tails[off],
 860                                entry);
 861   cp->off_sum += off;
 862   cp->num_paths++;
 863
 864   /* If we have a tunnel to this peer, tell the tunnel that there is a
 865      new path available. */
 866   if (NULL != cp->t)
 867     GCT_consider_path (cp->t,
 868                        entry->path,
 869                        off);
 870
 871   if ( (NULL != cp->search_h) &&
 872        (DESIRED_CONNECTIONS_PER_TUNNEL <= cp->num_paths) )
 873   {
 874     /* Now I have enough paths, stop search */
 875     GCD_search_stop (cp->search_h);
 876     cp->search_h = NULL;
 877   }
 878   if (NULL != cp->destroy_task)
 879   {
 880     /* paths changed, this resets the destroy timeout counter
 881        and aborts a destroy task that may no longer be valid
 882        to have (as we now have more paths via this peer). */
 883     consider_peer_destroy (cp);
 884   }
 885 }
 886
 887
 888 /**
 889  * Remove an entry from the DLL of all of the paths that this peer is on.
 890  *
 891  * @param cp peer to modify
 892  * @param entry an entry on a path
 893  * @param off offset of this peer on the path
 894  */
 895 void
 896 GCP_path_entry_remove (struct CadetPeer *cp,
 897                        struct CadetPeerPathEntry *entry,
 898                        unsigned int off)
 899 {
 900   LOG (GNUNET_ERROR_TYPE_DEBUG,
 901        "Removing knowledge about peer %s beging on path %s at offset %u\n",
 902        GCP_2s (cp),
 903        GCPP_2s (entry->path),
 904        off);
 905   GNUNET_CONTAINER_DLL_remove (cp->path_heads[off],
 906                                cp->path_tails[off],
 907                                entry);
 908   GNUNET_assert (0 < cp->num_paths);
 909   cp->off_sum -= off;
 910   cp->num_paths--;
 911   if ( (NULL == cp->core_mq) &&
 912        (NULL != cp->t) &&
 913        (NULL == cp->search_h) &&
 914        (DESIRED_CONNECTIONS_PER_TUNNEL > cp->num_paths) )
 915     cp->search_h
 916       = GCD_search (&cp->pid);
 917   if (NULL == cp->destroy_task)
 918   {
 919     /* paths changed, we might now be ready for destruction, check again */
 920     consider_peer_destroy (cp);
 921   }
 922 }
 923
 924
 925 /**
 926  * Prune down the number of paths to this peer, we seem to
 927  * have way too many.
 928  *
 929  * @param cls the `struct CadetPeer` to maintain the path heap for
 930  */
 931 static void
 932 path_heap_cleanup (void *cls)
 933 {
 934   struct CadetPeer *cp = cls;
 935   struct CadetPeerPath *root;
 936
 937   cp->heap_cleanup_task = NULL;
 938   while (GNUNET_CONTAINER_heap_get_size (cp->path_heap) >=
 939          2 * DESIRED_CONNECTIONS_PER_TUNNEL)
 940   {
 941     /* Now we have way too many, drop least desirable UNLESS it is in use!
 942        (Note that this intentionally keeps highly desireable, but currently
 943        unused paths around in the hope that we might be able to switch, even
 944        if the number of paths exceeds the threshold.) */
 945     root = GNUNET_CONTAINER_heap_peek (cp->path_heap);
 946     GNUNET_assert (NULL != root);
 947     if (NULL !=
 948         GCPP_get_connection (root,
 949                              cp,
 950                              GCPP_get_length (root) - 1))
 951       break; /* can't fix */
 952     /* Got plenty of paths to this destination, and this is a low-quality
 953        one that we don't care about. Allow it to die. */
 954     GNUNET_assert (root ==
 955                    GNUNET_CONTAINER_heap_remove_root (cp->path_heap));
 956     GCPP_release (root);
 957   }
 958 }
 959
 960
 961 /**
 962  * Try adding a @a path to this @a peer.  If the peer already
 963  * has plenty of paths, return NULL.
 964  *
 965  * @param cp peer to which the @a path leads to
 966  * @param path a path looking for an owner; may not be fully initialized yet!
 967  * @param off offset of @a cp in @a path
 968  * @param force force attaching the path
 969  * @return NULL if this peer does not care to become a new owner,
 970  *         otherwise the node in the peer's path heap for the @a path.
 971  */
 972 struct GNUNET_CONTAINER_HeapNode *
 973 GCP_attach_path (struct CadetPeer *cp,
 974                  struct CadetPeerPath *path,
 975                  unsigned int off,
 976                  int force)
 977 {
 978   GNUNET_CONTAINER_HeapCostType desirability;
 979   struct CadetPeerPath *root;
 980   GNUNET_CONTAINER_HeapCostType root_desirability;
 981   struct GNUNET_CONTAINER_HeapNode *hn;
 982
 983   GNUNET_assert (off == GCPP_get_length (path) - 1);
 984   GNUNET_assert (cp == GCPP_get_peer_at_offset (path,
 985                                                 off));
 986   if (NULL == cp->path_heap)
 987   {
 988     /* #GCP_drop_owned_paths() was already called, we cannot take new ones! */
 989     GNUNET_assert (GNUNET_NO == force);
 990     return NULL;
 991   }
 992   desirability = GCPP_get_desirability (path);
 993   if (GNUNET_NO == force)
 994   {
 995     /* FIXME: desirability is not yet initialized; tricky! */
 996     if (GNUNET_NO ==
 997         GNUNET_CONTAINER_heap_peek2 (cp->path_heap,
 998                                      (void **) &root,
 999                                      &root_desirability))
1000     {
1001       root = NULL;
1002       root_desirability = 0;
1003     }
1004
1005     if ( (DESIRED_CONNECTIONS_PER_TUNNEL > cp->num_paths) &&
1006          (desirability < root_desirability) )
1007     {
1008       LOG (GNUNET_ERROR_TYPE_DEBUG,
1009            "Decided to not attach path %s to peer %s due to undesirability\n",
1010            GCPP_2s (path),
1011            GCP_2s (cp));
1012       return NULL;
1013     }
1014   }
1015
1016   LOG (GNUNET_ERROR_TYPE_DEBUG,
1017        "Attaching path %s to peer %s (%s)\n",
1018        GCPP_2s (path),
1019        GCP_2s (cp),
1020        (GNUNET_NO == force) ? "desirable" : "forced");
1021
1022   /* Yes, we'd like to add this path, add to our heap */
1023   hn = GNUNET_CONTAINER_heap_insert (cp->path_heap,
1024                                      path,
1025                                      desirability);
1026
1027   /* Consider maybe dropping other paths because of the new one */
1028   if ( (GNUNET_CONTAINER_heap_get_size (cp->path_heap) >=
1029         2 * DESIRED_CONNECTIONS_PER_TUNNEL) &&
1030        (NULL != cp->heap_cleanup_task) )
1031     cp->heap_cleanup_task = GNUNET_SCHEDULER_add_now (&path_heap_cleanup,
1032                                                       cp);
1033   return hn;
1034 }
1035
1036
1037 /**
1038  * This peer can no longer own @a path as the path
1039  * has been extended and a peer further down the line
1040  * is now the new owner.
1041  *
1042  * @param cp old owner of the @a path
1043  * @param path path where the ownership is lost
1044  * @param hn note in @a cp's path heap that must be deleted
1045  */
1046 void
1047 GCP_detach_path (struct CadetPeer *cp,
1048                  struct CadetPeerPath *path,
1049                  struct GNUNET_CONTAINER_HeapNode *hn)
1050 {
1051   LOG (GNUNET_ERROR_TYPE_DEBUG,
1052        "Detatching path %s from peer %s\n",
1053        GCPP_2s (path),
1054        GCP_2s (cp));
1055   GNUNET_assert (path ==
1056                  GNUNET_CONTAINER_heap_remove_node (hn));
1057 }
1058
1059
1060 /**
1061  * Add a @a connection to this @a cp.
1062  *
1063  * @param cp peer via which the @a connection goes
1064  * @param cc the connection to add
1065  */
1066 void
1067 GCP_add_connection (struct CadetPeer *cp,
1068                     struct CadetConnection *cc)
1069 {
1070   LOG (GNUNET_ERROR_TYPE_DEBUG,
1071        "Adding %s to peer %s\n",
1072        GCC_2s (cc),
1073        GCP_2s (cp));
1074   GNUNET_assert (GNUNET_OK ==
1075                  GNUNET_CONTAINER_multishortmap_put (cp->connections,
1076                                                      &GCC_get_id (cc)->connection_of_tunnel,
1077                                                      cc,
1078                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1079   if (NULL != cp->destroy_task)
1080   {
1081     GNUNET_SCHEDULER_cancel (cp->destroy_task);
1082     cp->destroy_task = NULL;
1083   }
1084 }
1085
1086
1087 /**
1088  * Remove a @a connection that went via this @a cp.
1089  *
1090  * @param cp peer via which the @a connection went
1091  * @param cc the connection to remove
1092  */
1093 void
1094 GCP_remove_connection (struct CadetPeer *cp,
1095                        struct CadetConnection *cc)
1096 {
1097   LOG (GNUNET_ERROR_TYPE_DEBUG,
1098        "Removing connection %s from peer %s\n",
1099        GCC_2s (cc),
1100        GCP_2s (cp));
1101   GNUNET_assert (GNUNET_YES ==
1102                  GNUNET_CONTAINER_multishortmap_remove (cp->connections,
1103                                                         &GCC_get_id (cc)->connection_of_tunnel,
1104                                                         cc));
1105   consider_peer_destroy (cp);
1106 }
1107
1108
1109 /**
1110  * Retrieve the CadetPeer stucture associated with the
1111  * peer. Optionally create one and insert it in the appropriate
1112  * structures if the peer is not known yet.
1113  *
1114  * @param peer_id Full identity of the peer.
1115  * @param create #GNUNET_YES if a new peer should be created if unknown.
1116  *               #GNUNET_NO to return NULL if peer is unknown.
1117  * @return Existing or newly created peer structure.
1118  *         NULL if unknown and not requested @a create
1119  */
1120 struct CadetPeer *
1121 GCP_get (const struct GNUNET_PeerIdentity *peer_id,
1122          int create)
1123 {
1124   struct CadetPeer *cp;
1125
1126   cp = GNUNET_CONTAINER_multipeermap_get (peers,
1127                                           peer_id);
1128   if (NULL != cp)
1129     return cp;
1130   if (GNUNET_NO == create)
1131     return NULL;
1132   cp = GNUNET_new (struct CadetPeer);
1133   cp->pid = *peer_id;
1134   cp->connections = GNUNET_CONTAINER_multishortmap_create (32,
1135                                                            GNUNET_YES);
1136   cp->path_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1137   GNUNET_assert (GNUNET_YES ==
1138                  GNUNET_CONTAINER_multipeermap_put (peers,
1139                                                     &cp->pid,
1140                                                     cp,
1141                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1142   LOG (GNUNET_ERROR_TYPE_DEBUG,
1143        "Creating peer %s\n",
1144        GCP_2s (cp));
1145   return cp;
1146 }
1147
1148
1149 /**
1150  * Obtain the peer identity for a `struct CadetPeer`.
1151  *
1152  * @param cp our peer handle
1153  * @return the peer identity
1154  */
1155 const struct GNUNET_PeerIdentity *
1156 GCP_get_id (struct CadetPeer *cp)
1157 {
1158   return &cp->pid;
1159 }
1160
1161
1162 /**
1163  * Iterate over all known peers.
1164  *
1165  * @param iter Iterator.
1166  * @param cls Closure for @c iter.
1167  */
1168 void
1169 GCP_iterate_all (GNUNET_CONTAINER_PeerMapIterator iter,
1170                  void *cls)
1171 {
1172   GNUNET_CONTAINER_multipeermap_iterate (peers,
1173                                          iter,
1174                                          cls);
1175 }
1176
1177
1178 /**
1179  * Count the number of known paths toward the peer.
1180  *
1181  * @param cp Peer to get path info.
1182  * @return Number of known paths.
1183  */
1184 unsigned int
1185 GCP_count_paths (const struct CadetPeer *cp)
1186 {
1187   return cp->num_paths;
1188 }
1189
1190
1191 /**
1192  * Iterate over the paths to a peer.
1193  *
1194  * @param cp Peer to get path info.
1195  * @param callback Function to call for every path.
1196  * @param callback_cls Closure for @a callback.
1197  * @return Number of iterated paths.
1198  */
1199 unsigned int
1200 GCP_iterate_paths (struct CadetPeer *cp,
1201                    GCP_PathIterator callback,
1202                    void *callback_cls)
1203 {
1204   unsigned int ret = 0;
1205
1206   LOG (GNUNET_ERROR_TYPE_DEBUG,
1207        "Iterating over paths to peer %s%s\n",
1208        GCP_2s (cp),
1209        (NULL == cp->core_mq) ? "" : " including direct link");
1210   if (NULL != cp->core_mq)
1211   {
1212     struct CadetPeerPath *path;
1213
1214     path = GCPP_get_path_from_route (1,
1215                                      &cp->pid);
1216     ret++;
1217     if (GNUNET_NO ==
1218         callback (callback_cls,
1219                   path,
1220                   0))
1221       return ret;
1222   }
1223   for (unsigned int i=0;i<cp->path_dll_length;i++)
1224   {
1225     for (struct CadetPeerPathEntry *pe = cp->path_heads[i];
1226          NULL != pe;
1227          pe = pe->next)
1228     {
1229       ret++;
1230       if (GNUNET_NO ==
1231           callback (callback_cls,
1232                     pe->path,
1233                     i))
1234         return ret;
1235     }
1236   }
1237   return ret;
1238 }
1239
1240
1241 /**
1242  * Iterate over the paths to @a cp where
1243  * @a cp is at distance @a dist from us.
1244  *
1245  * @param cp Peer to get path info.
1246  * @param dist desired distance of @a cp to us on the path
1247  * @param callback Function to call for every path.
1248  * @param callback_cls Closure for @a callback.
1249  * @return Number of iterated paths.
1250  */
1251 unsigned int
1252 GCP_iterate_paths_at (struct CadetPeer *cp,
1253                       unsigned int dist,
1254                       GCP_PathIterator callback,
1255                       void *callback_cls)
1256 {
1257   unsigned int ret = 0;
1258
1259   if (dist >= cp->path_dll_length)
1260   {
1261     LOG (GNUNET_ERROR_TYPE_DEBUG,
1262          "Asked to look for paths at distance %u, but maximum for me is < %u\n",
1263          dist,
1264          cp->path_dll_length);
1265     return 0;
1266   }
1267   for (struct CadetPeerPathEntry *pe = cp->path_heads[dist];
1268        NULL != pe;
1269        pe = pe->next)
1270   {
1271     if (GNUNET_NO ==
1272         callback (callback_cls,
1273                   pe->path,
1274                   dist))
1275       return ret;
1276     ret++;
1277   }
1278   return ret;
1279 }
1280
1281
1282 /**
1283  * Get the tunnel towards a peer.
1284  *
1285  * @param cp Peer to get from.
1286  * @param create #GNUNET_YES to create a tunnel if we do not have one
1287  * @return Tunnel towards peer.
1288  */
1289 struct CadetTunnel *
1290 GCP_get_tunnel (struct CadetPeer *cp,
1291                 int create)
1292 {
1293   if (NULL == cp)
1294     return NULL;
1295   if ( (NULL != cp->t) ||
1296        (GNUNET_NO == create) )
1297     return cp->t;
1298   cp->t = GCT_create_tunnel (cp);
1299   consider_peer_activate (cp);
1300   return cp->t;
1301 }
1302
1303
1304 /**
1305  * Hello offer was passed to the transport service. Mark it
1306  * as done.
1307  *
1308  * @param cls the `struct CadetPeer` where the offer completed
1309  */
1310 static void
1311 hello_offer_done (void *cls)
1312 {
1313   struct CadetPeer *cp = cls;
1314
1315   cp->hello_offer = NULL;
1316 }
1317
1318
1319 /**
1320  * We got a HELLO for a @a peer, remember it, and possibly
1321  * trigger adequate actions (like trying to connect).
1322  *
1323  * @param cp the peer we got a HELLO for
1324  * @param hello the HELLO to remember
1325  */
1326 void
1327 GCP_set_hello (struct CadetPeer *cp,
1328                const struct GNUNET_HELLO_Message *hello)
1329 {
1330   struct GNUNET_HELLO_Message *mrg;
1331
1332   LOG (GNUNET_ERROR_TYPE_DEBUG,
1333        "Got %u byte HELLO for peer %s\n",
1334        (unsigned int) GNUNET_HELLO_size (hello),
1335        GCP_2s (cp));
1336   if (NULL != cp->hello_offer)
1337   {
1338     GNUNET_TRANSPORT_offer_hello_cancel (cp->hello_offer);
1339     cp->hello_offer = NULL;
1340   }
1341   if (NULL != cp->hello)
1342   {
1343     mrg = GNUNET_HELLO_merge (hello,
1344                               cp->hello);
1345     GNUNET_free (cp->hello);
1346     cp->hello = mrg;
1347   }
1348   else
1349   {
1350     cp->hello = GNUNET_memdup (hello,
1351                                GNUNET_HELLO_size (hello));
1352   }
1353   cp->hello_offer
1354     = GNUNET_TRANSPORT_offer_hello (cfg,
1355                                     GNUNET_HELLO_get_header (cp->hello) ,
1356                                     &hello_offer_done,
1357                                     cp);
1358   /* New HELLO means cp's destruction time may change... */
1359   consider_peer_destroy (cp);
1360 }
1361
1362
1363 /**
1364  * The tunnel to the given peer no longer exists, remove it from our
1365  * data structures, and possibly clean up the peer itself.
1366  *
1367  * @param cp the peer affected
1368  * @param t the dead tunnel
1369  */
1370 void
1371 GCP_drop_tunnel (struct CadetPeer *cp,
1372                  struct CadetTunnel *t)
1373 {
1374   LOG (GNUNET_ERROR_TYPE_DEBUG,
1375        "Dropping tunnel %s to peer %s\n",
1376        GCT_2s (t),
1377        GCP_2s (cp));
1378   GNUNET_assert (cp->t == t);
1379   cp->t = NULL;
1380   consider_peer_destroy (cp);
1381 }
1382
1383
1384 /**
1385  * Test if @a cp has a core-level connection
1386  *
1387  * @param cp peer to test
1388  * @return #GNUNET_YES if @a cp has a core-level connection
1389  */
1390 int
1391 GCP_has_core_connection (struct CadetPeer *cp)
1392 {
1393   return (NULL != cp->core_mq) ? GNUNET_YES : GNUNET_NO;
1394 }
1395
1396
1397 /**
1398  * Start message queue change notifications.
1399  *
1400  * @param cp peer to notify for
1401  * @param cb function to call if mq becomes available or unavailable
1402  * @param cb_cls closure for @a cb
1403  * @return handle to cancel request
1404  */
1405 struct GCP_MessageQueueManager *
1406 GCP_request_mq (struct CadetPeer *cp,
1407                 GCP_MessageQueueNotificationCallback cb,
1408                 void *cb_cls)
1409 {
1410   struct GCP_MessageQueueManager *mqm;
1411
1412   mqm = GNUNET_new (struct GCP_MessageQueueManager);
1413   mqm->cb = cb;
1414   mqm->cb_cls = cb_cls;
1415   mqm->cp = cp;
1416   GNUNET_CONTAINER_DLL_insert (cp->mqm_head,
1417                                cp->mqm_tail,
1418                                mqm);
1419   LOG (GNUNET_ERROR_TYPE_DEBUG,
1420        "Creating MQM %p for peer %s\n",
1421        mqm,
1422        GCP_2s (cp));
1423   if (NULL != cp->core_mq)
1424     cb (cb_cls,
1425         GNUNET_YES);
1426   return mqm;
1427 }
1428
1429
1430 /**
1431  * Stops message queue change notifications.
1432  *
1433  * @param mqm handle matching request to cancel
1434  * @param last_env final message to transmit, or NULL
1435  */
1436 void
1437 GCP_request_mq_cancel (struct GCP_MessageQueueManager *mqm,
1438                        struct GNUNET_MQ_Envelope *last_env)
1439 {
1440   struct CadetPeer *cp = mqm->cp;
1441
1442   LOG (GNUNET_ERROR_TYPE_DEBUG,
1443        "Destroying MQM %p for peer %s%s\n",
1444        mqm,
1445        GCP_2s (cp),
1446        (NULL == last_env) ? "" : " with last ditch transmission");
1447   if (NULL != mqm->env)
1448     GNUNET_MQ_discard (mqm->env);
1449   if (NULL != last_env)
1450   {
1451     if (NULL != cp->core_mq)
1452     {
1453       GNUNET_MQ_notify_sent (last_env,
1454                              &mqm_send_done,
1455                              cp);
1456       GNUNET_MQ_send (cp->core_mq,
1457                       last_env);
1458     }
1459     else
1460     {
1461       GNUNET_MQ_discard (last_env);
1462     }
1463   }
1464   if (cp->mqm_ready_ptr == mqm)
1465     cp->mqm_ready_ptr = mqm->next;
1466   GNUNET_CONTAINER_DLL_remove (cp->mqm_head,
1467                                cp->mqm_tail,
1468                                mqm);
1469   GNUNET_free (mqm);
1470 }
1471
1472
1473 /**
1474  * Send the message in @a env to @a cp, overriding queueing logic.
1475  * This function should only be used to send error messages outside
1476  * of flow and congestion control, similar to ICMP.  Note that
1477  * the envelope may be silently discarded as well.
1478  *
1479  * @param cp peer to send the message to
1480  * @param env envelope with the message to send
1481  */
1482 void
1483 GCP_send_ooo (struct CadetPeer *cp,
1484               struct GNUNET_MQ_Envelope *env)
1485 {
1486   LOG (GNUNET_ERROR_TYPE_DEBUG,
1487        "Sending message to %s out of management\n",
1488        GCP_2s (cp));
1489   if (NULL == cp->core_mq)
1490   {
1491     GNUNET_MQ_discard (env);
1492     return;
1493   }
1494   if (GNUNET_MQ_get_length (cp->core_mq) > MAX_OOO_QUEUE_SIZE)
1495   {
1496     GNUNET_MQ_discard (env);
1497     return;
1498   }
1499   GNUNET_MQ_notify_sent (env,
1500                          &mqm_send_done,
1501                          cp);
1502   GNUNET_MQ_send (cp->core_mq,
1503                   env);
1504 }
1505
1506
1507
1508
1509 /* end of gnunet-service-cadet-new_peer.c */