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