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