src/transport/gnunet-service-tng.c

   1 /*
   2  This file is part of GNUnet.
   3  Copyright (C) 2010-2016, 2018 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  * @file transport/gnunet-service-tng.c
  22  * @brief main for gnunet-service-tng
  23  * @author Christian Grothoff
  24  *
  25  * TODO:
  26  * - figure out how to transmit (selective) ACKs in case of uni-directional
  27  *   communicators (with/without core? DV-only?) When do we use ACKs?
  28  *   => communicators use selective ACKs for flow control
  29  *   => transport uses message-level ACKs for RTT, fragment confirmation
  30  *   => integrate DV into transport, use neither core nor communicators
  31  *      but rather give communicators transport-encapsulated messages
  32  *      (which could be core-data, background-channel traffic, or
  33  *       transport-to-transport traffic)
  34  *
  35  * Implement:
  36  * - manage fragmentation/defragmentation, retransmission, track RTT, loss, etc.
  37  *
  38  * Easy:
  39  * - use ATS bandwidth allocation callback and schedule transmissions!
  40  *
  41  * Plan:
  42  * - inform ATS about RTT, goodput/loss, overheads, etc.
  43  *
  44  * Later:
  45  * - change transport-core API to provide proper flow control in both
  46  *   directions, allow multiple messages per peer simultaneously (tag
  47  *   confirmations with unique message ID), and replace quota-out with
  48  *   proper flow control;
  49  *
  50  * Design realizations / discussion:
  51  * - communicators do flow control by calling MQ "notify sent"
  52  *   when 'ready'. They determine flow implicitly (i.e. TCP blocking)
  53  *   or explicitly via background channel FC ACKs.  As long as the
  54  *   channel is not full, they may 'notify sent' even if the other
  55  *   peer has not yet confirmed receipt. The other peer confirming
  56  *   is _only_ for FC, not for more reliable transmission; reliable
  57  *   transmission (i.e. of fragments) is left to _transport_.
  58  * - ACKs sent back in uni-directional communicators are done via
  59  *   the background channel API; here transport _may_ initially
  60  *   broadcast (with bounded # hops) if no path is known;
  61  * - transport should _integrate_ DV-routing and build a view of
  62  *   the network; then background channel traffic can be
  63  *   routed via DV as well as explicit "DV" traffic.
  64  * - background channel is also used for ACKs and NAT traversal support
  65  * - transport service is responsible for AEAD'ing the background
  66  *   channel, timestamps and monotonic time are used against replay
  67  *   of old messages -> peerstore needs to be supplied with
  68  *   "latest timestamps seen" data
  69  * - if transport implements DV, we likely need a 3rd peermap
  70  *   in addition to ephemerals and (direct) neighbours
  71  *   => in this data structure, we should track ATS metrics (distance, RTT, etc.)
  72  *   as well as latest timestamps seen, goodput, fragments for transmission, etc.
  73  *   ==> check if stuff needs to be moved out of "Neighbour"
  74  * - transport should encapsualte core-level messages and do its
  75  *   own ACKing for RTT/goodput/loss measurements _and_ fragment
  76  *   for retransmission
  77  */
  78 #include "platform.h"
  79 #include "gnunet_util_lib.h"
  80 #include "gnunet_statistics_service.h"
  81 #include "gnunet_transport_monitor_service.h"
  82 #include "gnunet_peerstore_service.h"
  83 #include "gnunet_hello_lib.h"
  84 #include "gnunet_ats_transport_service.h"
  85 #include "transport.h"
  86
  87
  88 /**
  89  * How many messages can we have pending for a given client process
  90  * before we start to drop incoming messages?  We typically should
  91  * have only one client and so this would be the primary buffer for
  92  * messages, so the number should be chosen rather generously.
  93  *
  94  * The expectation here is that most of the time the queue is large
  95  * enough so that a drop is virtually never required.  Note that
  96  * this value must be about as large as 'TOTAL_MSGS' in the
  97  * 'test_transport_api_reliability.c', otherwise that testcase may
  98  * fail.
  99  */
 100 #define MAX_PENDING (128 * 1024)
 101
 102
 103 GNUNET_NETWORK_STRUCT_BEGIN
 104
 105 /**
 106  * Outer layer of an encapsulated backchannel message.
 107  */
 108 struct TransportBackchannelEncapsulationMessage
 109 {
 110   /**
 111    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION.
 112    */
 113   struct GNUNET_MessageHeader header;
 114
 115   /**
 116    * Distance the backchannel message has traveled, to be updated at
 117    * each hop.  Used to bound the number of hops in case a backchannel
 118    * message is broadcast and thus travels without routing
 119    * information (during initial backchannel discovery).
 120    */
 121   uint32_t distance;
 122
 123   /**
 124    * Target's peer identity (as backchannels may be transmitted
 125    * indirectly, or even be broadcast).
 126    */
 127   struct GNUNET_PeerIdentity target;
 128
 129   /**
 130    * Ephemeral key setup by the sender for @e target, used
 131    * to encrypt the payload.
 132    */
 133   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
 134
 135   /**
 136    * HMAC over the ciphertext of the encrypted, variable-size
 137    * body that follows.  Verified via DH of @e target and
 138    * @e ephemeral_key
 139    */
 140   struct GNUNET_HashCode hmac;
 141
 142   /* Followed by encrypted, variable-size payload */
 143 };
 144
 145
 146 /**
 147  * Message by which a peer confirms that it is using an
 148  * ephemeral key.
 149  */
 150 struct EphemeralConfirmation
 151 {
 152
 153   /**
 154    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
 155    */
 156   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 157
 158   /**
 159    * How long is this signature over the ephemeral key
 160    * valid?
 161    */
 162   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
 163
 164   /**
 165    * Ephemeral key setup by the sender for @e target, used
 166    * to encrypt the payload.
 167    */
 168   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
 169
 170 };
 171
 172 /**
 173  * Plaintext of the variable-size payload that is encrypted
 174  * within a `struct TransportBackchannelEncapsulationMessage`
 175  */
 176 struct TransportBackchannelRequestPayload
 177 {
 178
 179   /**
 180    * Sender's peer identity.
 181    */
 182   struct GNUNET_PeerIdentity sender;
 183
 184   /**
 185    * Signature of the sender over an
 186    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL.
 187    */
 188   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 189
 190   /**
 191    * How long is this signature over the ephemeral key
 192    * valid?
 193    */
 194   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
 195
 196   /**
 197    * Current monotonic time of the sending transport service.  Used to
 198    * detect replayed messages.  Note that the receiver should remember
 199    * a list of the recently seen timestamps and only reject messages
 200    * if the timestamp is in the list, or the list is "full" and the
 201    * timestamp is smaller than the lowest in the list.  This list of
 202    * timestamps per peer should be persisted to guard against replays
 203    * after restarts.
 204    */
 205   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 206
 207   /* Followed by a `struct GNUNET_MessageHeader` with a message
 208      for a communicator */
 209
 210   /* Followed by a 0-termianted string specifying the name of
 211      the communicator which is to receive the message */
 212
 213 };
 214
 215 GNUNET_NETWORK_STRUCT_END
 216
 217
 218
 219 /**
 220  * What type of client is the `struct TransportClient` about?
 221  */
 222 enum ClientType
 223 {
 224   /**
 225    * We do not know yet (client is fresh).
 226    */
 227   CT_NONE = 0,
 228
 229   /**
 230    * Is the CORE service, we need to forward traffic to it.
 231    */
 232   CT_CORE = 1,
 233
 234   /**
 235    * It is a monitor, forward monitor data.
 236    */
 237   CT_MONITOR = 2,
 238
 239   /**
 240    * It is a communicator, use for communication.
 241    */
 242   CT_COMMUNICATOR = 3
 243 };
 244
 245
 246 /**
 247  * Entry in our cache of ephemeral keys we currently use.
 248  */
 249 struct EphemeralCacheEntry
 250 {
 251
 252   /**
 253    * Target's peer identity (we don't re-use ephemerals
 254    * to limit linkability of messages).
 255    */
 256   struct GNUNET_PeerIdentity target;
 257
 258   /**
 259    * Signature affirming @e ephemeral_key of type
 260    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
 261    */
 262   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 263
 264   /**
 265    * How long is @e sender_sig valid
 266    */
 267   struct GNUNET_TIME_Absolute ephemeral_validity;
 268
 269   /**
 270    * Our ephemeral key.
 271    */
 272   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
 273
 274   /**
 275    * Node in the ephemeral cache for this entry.
 276    * Used for expiration.
 277    */
 278   struct GNUNET_CONTAINER_HeapNode *hn;
 279 };
 280
 281
 282 /**
 283  * Client connected to the transport service.
 284  */
 285 struct TransportClient;
 286
 287
 288 /**
 289  * A neighbour that at least one communicator is connected to.
 290  */
 291 struct Neighbour;
 292
 293
 294 /**
 295  * An ATS session is a message queue provided by a communicator
 296  * via which we can reach a particular neighbour.
 297  */
 298 struct GNUNET_ATS_Session
 299 {
 300   /**
 301    * Kept in a MDLL.
 302    */
 303   struct GNUNET_ATS_Session *next_neighbour;
 304
 305   /**
 306    * Kept in a MDLL.
 307    */
 308   struct GNUNET_ATS_Session *prev_neighbour;
 309
 310   /**
 311    * Kept in a MDLL.
 312    */
 313   struct GNUNET_ATS_Session *prev_client;
 314
 315   /**
 316    * Kept in a MDLL.
 317    */
 318   struct GNUNET_ATS_Session *next_client;
 319
 320   /**
 321    * Which neighbour is this ATS session for?
 322    */
 323   struct Neighbour *neighbour;
 324
 325   /**
 326    * Which communicator offers this ATS session?
 327    */
 328   struct TransportClient *tc;
 329
 330   /**
 331    * Address served by the ATS session.
 332    */
 333   const char *address;
 334
 335   /**
 336    * Our current RTT estimate for this ATS session.
 337    */
 338   struct GNUNET_TIME_Relative rtt;
 339
 340   /**
 341    * Unique identifier of this ATS session with the communicator.
 342    */
 343   uint32_t qid;
 344
 345   /**
 346    * Maximum transmission unit supported by this ATS session.
 347    */
 348   uint32_t mtu;
 349
 350   /**
 351    * Distance to the target of this ATS session.
 352    */
 353   uint32_t distance;
 354
 355   /**
 356    * Network type offered by this ATS session.
 357    */
 358   enum GNUNET_NetworkType nt;
 359
 360   /**
 361    * Connection status for this ATS session.
 362    */
 363   enum GNUNET_TRANSPORT_ConnectionStatus cs;
 364
 365   /**
 366    * Messages pending.
 367    */
 368   uint32_t num_msg_pending;
 369
 370   /**
 371    * Bytes pending.
 372    */
 373   uint32_t num_bytes_pending;
 374
 375   /**
 376    * How much outbound bandwidth do we have available for this session?
 377    */
 378   struct GNUNET_BANDWIDTH_Tracker tracker_out;
 379
 380   /**
 381    * How much inbound bandwidth do we have available for this session?
 382    */
 383   struct GNUNET_BANDWIDTH_Tracker tracker_in;
 384 };
 385
 386
 387 /**
 388  * A neighbour that at least one communicator is connected to.
 389  */
 390 struct Neighbour
 391 {
 392
 393   /**
 394    * Which peer is this about?
 395    */
 396   struct GNUNET_PeerIdentity pid;
 397
 398   /**
 399    * Head of list of messages pending for this neighbour.
 400    */
 401   struct PendingMessage *pending_msg_head;
 402
 403   /**
 404    * Tail of list of messages pending for this neighbour.
 405    */
 406   struct PendingMessage *pending_msg_tail;
 407
 408   /**
 409    * Head of DLL of ATS sessions to this peer.
 410    */
 411   struct GNUNET_ATS_Session *session_head;
 412
 413   /**
 414    * Tail of DLL of ATS sessions to this peer.
 415    */
 416   struct GNUNET_ATS_Session *session_tail;
 417
 418   /**
 419    * Quota at which CORE is allowed to transmit to this peer
 420    * according to ATS.
 421    *
 422    * FIXME: not yet used, tricky to get right given multiple queues!
 423    *        (=> Idea: let ATS set a quota per queue and we add them up here?)
 424    * FIXME: how do we set this value initially when we tell CORE?
 425    *    Options: start at a minimum value or at literally zero (before ATS?)
 426    *         (=> Current thought: clean would be zero!)
 427    */
 428   struct GNUNET_BANDWIDTH_Value32NBO quota_out;
 429
 430 };
 431
 432
 433 /**
 434  * Transmission request from CORE that is awaiting delivery.
 435  */
 436 struct PendingMessage
 437 {
 438   /**
 439    * Kept in a MDLL of messages for this @a target.
 440    */
 441   struct PendingMessage *next_neighbour;
 442
 443   /**
 444    * Kept in a MDLL of messages for this @a target.
 445    */
 446   struct PendingMessage *prev_neighbour;
 447
 448   /**
 449    * Kept in a MDLL of messages from this @a client.
 450    */
 451   struct PendingMessage *next_client;
 452
 453   /**
 454    * Kept in a MDLL of messages from this @a client.
 455    */
 456   struct PendingMessage *prev_client;
 457
 458   /**
 459    * Target of the request.
 460    */
 461   struct Neighbour *target;
 462
 463   /**
 464    * Client that issued the transmission request.
 465    */
 466   struct TransportClient *client;
 467
 468   /**
 469    * Size of the original message.
 470    */
 471   uint32_t bytes_msg;
 472
 473 };
 474
 475
 476 /**
 477  * One of the addresses of this peer.
 478  */
 479 struct AddressListEntry
 480 {
 481
 482   /**
 483    * Kept in a DLL.
 484    */
 485   struct AddressListEntry *next;
 486
 487   /**
 488    * Kept in a DLL.
 489    */
 490   struct AddressListEntry *prev;
 491
 492   /**
 493    * Which communicator provides this address?
 494    */
 495   struct TransportClient *tc;
 496
 497   /**
 498    * The actual address.
 499    */
 500   const char *address;
 501
 502   /**
 503    * Current context for storing this address in the peerstore.
 504    */
 505   struct GNUNET_PEERSTORE_StoreContext *sc;
 506
 507   /**
 508    * Task to periodically do @e st operation.
 509    */
 510   struct GNUNET_SCHEDULER_Task *st;
 511
 512   /**
 513    * What is a typical lifetime the communicator expects this
 514    * address to have? (Always from now.)
 515    */
 516   struct GNUNET_TIME_Relative expiration;
 517
 518   /**
 519    * Address identifier used by the communicator.
 520    */
 521   uint32_t aid;
 522
 523   /**
 524    * Network type offered by this address.
 525    */
 526   enum GNUNET_NetworkType nt;
 527
 528 };
 529
 530
 531 /**
 532  * Client connected to the transport service.
 533  */
 534 struct TransportClient
 535 {
 536
 537   /**
 538    * Kept in a DLL.
 539    */
 540   struct TransportClient *next;
 541
 542   /**
 543    * Kept in a DLL.
 544    */
 545   struct TransportClient *prev;
 546
 547   /**
 548    * Handle to the client.
 549    */
 550   struct GNUNET_SERVICE_Client *client;
 551
 552   /**
 553    * Message queue to the client.
 554    */
 555   struct GNUNET_MQ_Handle *mq;
 556
 557   /**
 558    * What type of client is this?
 559    */
 560   enum ClientType type;
 561
 562   union
 563   {
 564
 565     /**
 566      * Information for @e type #CT_CORE.
 567      */
 568     struct {
 569
 570       /**
 571        * Head of list of messages pending for this client.
 572        */
 573       struct PendingMessage *pending_msg_head;
 574
 575       /**
 576        * Tail of list of messages pending for this client.
 577        */
 578       struct PendingMessage *pending_msg_tail;
 579
 580     } core;
 581
 582     /**
 583      * Information for @e type #CT_MONITOR.
 584      */
 585     struct {
 586
 587       /**
 588        * Peer identity to monitor the addresses of.
 589        * Zero to monitor all neighbours.  Valid if
 590        * @e type is #CT_MONITOR.
 591        */
 592       struct GNUNET_PeerIdentity peer;
 593
 594       /**
 595        * Is this a one-shot monitor?
 596        */
 597       int one_shot;
 598
 599     } monitor;
 600
 601
 602     /**
 603      * Information for @e type #CT_COMMUNICATOR.
 604      */
 605     struct {
 606       /**
 607        * If @e type is #CT_COMMUNICATOR, this communicator
 608        * supports communicating using these addresses.
 609        */
 610       char *address_prefix;
 611
 612       /**
 613        * Head of DLL of queues offered by this communicator.
 614        */
 615       struct GNUNET_ATS_Session *session_head;
 616
 617       /**
 618        * Tail of DLL of queues offered by this communicator.
 619        */
 620       struct GNUNET_ATS_Session *session_tail;
 621
 622       /**
 623        * Head of list of the addresses of this peer offered by this communicator.
 624        */
 625       struct AddressListEntry *addr_head;
 626
 627       /**
 628        * Tail of list of the addresses of this peer offered by this communicator.
 629        */
 630       struct AddressListEntry *addr_tail;
 631
 632       /**
 633        * Characteristics of this communicator.
 634        */
 635       enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
 636
 637     } communicator;
 638
 639   } details;
 640
 641 };
 642
 643
 644 /**
 645  * Head of linked list of all clients to this service.
 646  */
 647 static struct TransportClient *clients_head;
 648
 649 /**
 650  * Tail of linked list of all clients to this service.
 651  */
 652 static struct TransportClient *clients_tail;
 653
 654 /**
 655  * Statistics handle.
 656  */
 657 static struct GNUNET_STATISTICS_Handle *GST_stats;
 658
 659 /**
 660  * Configuration handle.
 661  */
 662 static const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
 663
 664 /**
 665  * Our public key.
 666  */
 667 static struct GNUNET_PeerIdentity GST_my_identity;
 668
 669 /**
 670  * Our private key.
 671  */
 672 static struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
 673
 674 /**
 675  * Map from PIDs to `struct Neighbour` entries.  A peer is
 676  * a neighbour if we have an MQ to it from some communicator.
 677  */
 678 static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 679
 680 /**
 681  * Database for peer's HELLOs.
 682  */
 683 static struct GNUNET_PEERSTORE_Handle *peerstore;
 684
 685 /**
 686  * Heap sorting `struct EphemeralCacheEntry` by their
 687  * key/signature validity.
 688  */
 689 static struct GNUNET_CONTAINER_Heap *ephemeral_heap;
 690
 691 /**
 692  * Hash map for looking up `struct EphemeralCacheEntry`s
 693  * by peer identity. (We may have ephemerals in our
 694  * cache for which we do not have a neighbour entry,
 695  * and similar many neighbours may not need ephemerals,
 696  * so we use a second map.)
 697  */
 698 static struct GNUNET_CONTAINER_MultiPeerMap *ephemeral_map;
 699
 700 /**
 701  * Our connection to ATS for allocation and bootstrapping.
 702  */
 703 static struct GNUNET_ATS_TransportHandle *ats;
 704
 705
 706 /**
 707  * Free cached ephemeral key.
 708  *
 709  * @param ece cached signature to free
 710  */
 711 static void
 712 free_ephemeral (struct EphemeralCacheEntry *ece)
 713 {
 714   GNUNET_CONTAINER_multipeermap_remove (ephemeral_map,
 715                                         &ece->target,
 716                                         ece);
 717   GNUNET_CONTAINER_heap_remove_node (ece->hn);
 718   GNUNET_free (ece);
 719 }
 720
 721
 722 /**
 723  * Lookup neighbour record for peer @a pid.
 724  *
 725  * @param pid neighbour to look for
 726  * @return NULL if we do not have this peer as a neighbour
 727  */
 728 static struct Neighbour *
 729 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
 730 {
 731   return GNUNET_CONTAINER_multipeermap_get (neighbours,
 732                                             pid);
 733 }
 734
 735
 736 /**
 737  * Details about what to notify monitors about.
 738  */
 739 struct MonitorEvent
 740 {
 741   /**
 742    * @deprecated To be discussed if we keep these...
 743    */
 744   struct GNUNET_TIME_Absolute last_validation;
 745   struct GNUNET_TIME_Absolute valid_until;
 746   struct GNUNET_TIME_Absolute next_validation;
 747
 748   /**
 749    * Current round-trip time estimate.
 750    */
 751   struct GNUNET_TIME_Relative rtt;
 752
 753   /**
 754    * Connection status.
 755    */
 756   enum GNUNET_TRANSPORT_ConnectionStatus cs;
 757
 758   /**
 759    * Messages pending.
 760    */
 761   uint32_t num_msg_pending;
 762
 763   /**
 764    * Bytes pending.
 765    */
 766   uint32_t num_bytes_pending;
 767
 768
 769 };
 770
 771
 772 /**
 773  * Notify monitor @a tc about an event.  That @a tc
 774  * cares about the event has already been checked.
 775  *
 776  * Send @a tc information in @a me about a @a peer's status with
 777  * respect to some @a address to all monitors that care.
 778  *
 779  * @param tc monitor to inform
 780  * @param peer peer the information is about
 781  * @param address address the information is about
 782  * @param nt network type associated with @a address
 783  * @param me detailed information to transmit
 784  */
 785 static void
 786 notify_monitor (struct TransportClient *tc,
 787                 const struct GNUNET_PeerIdentity *peer,
 788                 const char *address,
 789                 enum GNUNET_NetworkType nt,
 790                 const struct MonitorEvent *me)
 791 {
 792   struct GNUNET_MQ_Envelope *env;
 793   struct GNUNET_TRANSPORT_MonitorData *md;
 794   size_t addr_len = strlen (address) + 1;
 795
 796   env = GNUNET_MQ_msg_extra (md,
 797                              addr_len,
 798                              GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_DATA);
 799   md->nt = htonl ((uint32_t) nt);
 800   md->peer = *peer;
 801   md->last_validation = GNUNET_TIME_absolute_hton (me->last_validation);
 802   md->valid_until = GNUNET_TIME_absolute_hton (me->valid_until);
 803   md->next_validation = GNUNET_TIME_absolute_hton (me->next_validation);
 804   md->rtt = GNUNET_TIME_relative_hton (me->rtt);
 805   md->cs = htonl ((uint32_t) me->cs);
 806   md->num_msg_pending = htonl (me->num_msg_pending);
 807   md->num_bytes_pending = htonl (me->num_bytes_pending);
 808   memcpy (&md[1],
 809           address,
 810           addr_len);
 811   GNUNET_MQ_send (tc->mq,
 812                   env);
 813 }
 814
 815
 816 /**
 817  * Send information in @a me about a @a peer's status with respect
 818  * to some @a address to all monitors that care.
 819  *
 820  * @param peer peer the information is about
 821  * @param address address the information is about
 822  * @param nt network type associated with @a address
 823  * @param me detailed information to transmit
 824  */
 825 static void
 826 notify_monitors (const struct GNUNET_PeerIdentity *peer,
 827                  const char *address,
 828                  enum GNUNET_NetworkType nt,
 829                  const struct MonitorEvent *me)
 830 {
 831   static struct GNUNET_PeerIdentity zero;
 832
 833   for (struct TransportClient *tc = clients_head;
 834        NULL != tc;
 835        tc = tc->next)
 836   {
 837     if (CT_MONITOR != tc->type)
 838       continue;
 839     if (tc->details.monitor.one_shot)
 840       continue;
 841     if ( (0 != memcmp (&tc->details.monitor.peer,
 842                        &zero,
 843                        sizeof (zero))) &&
 844          (0 != memcmp (&tc->details.monitor.peer,
 845                        peer,
 846                        sizeof (*peer))) )
 847       continue;
 848     notify_monitor (tc,
 849                     peer,
 850                     address,
 851                     nt,
 852                     me);
 853   }
 854 }
 855
 856
 857 /**
 858  * Called whenever a client connects.  Allocates our
 859  * data structures associated with that client.
 860  *
 861  * @param cls closure, NULL
 862  * @param client identification of the client
 863  * @param mq message queue for the client
 864  * @return our `struct TransportClient`
 865  */
 866 static void *
 867 client_connect_cb (void *cls,
 868                    struct GNUNET_SERVICE_Client *client,
 869                    struct GNUNET_MQ_Handle *mq)
 870 {
 871   struct TransportClient *tc;
 872
 873   tc = GNUNET_new (struct TransportClient);
 874   tc->client = client;
 875   tc->mq = mq;
 876   GNUNET_CONTAINER_DLL_insert (clients_head,
 877                                clients_tail,
 878                                tc);
 879   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 880               "Client %p connected\n",
 881               tc);
 882   return tc;
 883 }
 884
 885
 886 /**
 887  * Release memory used by @a neighbour.
 888  *
 889  * @param neighbour neighbour entry to free
 890  */
 891 static void
 892 free_neighbour (struct Neighbour *neighbour)
 893 {
 894   GNUNET_assert (NULL == neighbour->session_head);
 895   GNUNET_assert (GNUNET_YES ==
 896                  GNUNET_CONTAINER_multipeermap_remove (neighbours,
 897                                                        &neighbour->pid,
 898                                                        neighbour));
 899   GNUNET_free (neighbour);
 900 }
 901
 902
 903 /**
 904  * Send message to CORE clients that we lost a connection.
 905  *
 906  * @param tc client to inform (must be CORE client)
 907  * @param pid peer the connection is for
 908  * @param quota_out current quota for the peer
 909  */
 910 static void
 911 core_send_connect_info (struct TransportClient *tc,
 912                         const struct GNUNET_PeerIdentity *pid,
 913                         struct GNUNET_BANDWIDTH_Value32NBO quota_out)
 914 {
 915   struct GNUNET_MQ_Envelope *env;
 916   struct ConnectInfoMessage *cim;
 917
 918   GNUNET_assert (CT_CORE == tc->type);
 919   env = GNUNET_MQ_msg (cim,
 920                        GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
 921   cim->quota_out = quota_out;
 922   cim->id = *pid;
 923   GNUNET_MQ_send (tc->mq,
 924                   env);
 925 }
 926
 927
 928 /**
 929  * Send message to CORE clients that we gained a connection
 930  *
 931  * @param pid peer the queue was for
 932  * @param quota_out current quota for the peer
 933  */
 934 static void
 935 cores_send_connect_info (const struct GNUNET_PeerIdentity *pid,
 936                          struct GNUNET_BANDWIDTH_Value32NBO quota_out)
 937 {
 938   for (struct TransportClient *tc = clients_head;
 939        NULL != tc;
 940        tc = tc->next)
 941   {
 942     if (CT_CORE != tc->type)
 943       continue;
 944     core_send_connect_info (tc,
 945                             pid,
 946                             quota_out);
 947   }
 948 }
 949
 950
 951 /**
 952  * Send message to CORE clients that we lost a connection.
 953  *
 954  * @param pid peer the connection was for
 955  */
 956 static void
 957 cores_send_disconnect_info (const struct GNUNET_PeerIdentity *pid)
 958 {
 959   for (struct TransportClient *tc = clients_head;
 960        NULL != tc;
 961        tc = tc->next)
 962   {
 963     struct GNUNET_MQ_Envelope *env;
 964     struct DisconnectInfoMessage *dim;
 965
 966     if (CT_CORE != tc->type)
 967       continue;
 968     env = GNUNET_MQ_msg (dim,
 969                          GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
 970     dim->peer = *pid;
 971     GNUNET_MQ_send (tc->mq,
 972                     env);
 973   }
 974 }
 975
 976
 977 /**
 978  * Free @a queue.
 979  *
 980  * @param queue the queue to free
 981  */
 982 static void
 983 free_queue (struct GNUNET_ATS_Session *queue)
 984 {
 985   struct Neighbour *neighbour = queue->neighbour;
 986   struct TransportClient *tc = queue->tc;
 987   struct MonitorEvent me = {
 988     .cs = GNUNET_TRANSPORT_CS_DOWN,
 989     .rtt = GNUNET_TIME_UNIT_FOREVER_REL
 990   };
 991
 992   GNUNET_CONTAINER_MDLL_remove (neighbour,
 993                                 neighbour->session_head,
 994                                 neighbour->session_tail,
 995                                 queue);
 996   GNUNET_CONTAINER_MDLL_remove (client,
 997                                 tc->details.communicator.session_head,
 998                                 tc->details.communicator.session_tail,
 999                                 queue);
1000
1001   notify_monitors (&neighbour->pid,
1002                    queue->address,
1003                    queue->nt,
1004                    &me);
1005   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_in);
1006   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_out);
1007   GNUNET_free (queue);
1008   if (NULL == neighbour->session_head)
1009     {
1010       cores_send_disconnect_info (&neighbour->pid);
1011       free_neighbour (neighbour);
1012     }
1013 }
1014
1015
1016 /**
1017  * Free @a ale
1018  *
1019  * @param ale address list entry to free
1020  */
1021 static void
1022 free_address_list_entry (struct AddressListEntry *ale)
1023 {
1024   struct TransportClient *tc = ale->tc;
1025
1026   GNUNET_CONTAINER_DLL_remove (tc->details.communicator.addr_head,
1027                                tc->details.communicator.addr_tail,
1028                                ale);
1029   if (NULL != ale->sc)
1030   {
1031     GNUNET_PEERSTORE_store_cancel (ale->sc);
1032     ale->sc = NULL;
1033   }
1034   if (NULL != ale->st)
1035   {
1036     GNUNET_SCHEDULER_cancel (ale->st);
1037     ale->st = NULL;
1038   }
1039   GNUNET_free (ale);
1040 }
1041
1042
1043 /**
1044  * Called whenever a client is disconnected.  Frees our
1045  * resources associated with that client.
1046  *
1047  * @param cls closure, NULL
1048  * @param client identification of the client
1049  * @param app_ctx our `struct TransportClient`
1050  */
1051 static void
1052 client_disconnect_cb (void *cls,
1053                       struct GNUNET_SERVICE_Client *client,
1054                       void *app_ctx)
1055 {
1056   struct TransportClient *tc = app_ctx;
1057
1058   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1059               "Client %p disconnected, cleaning up.\n",
1060               tc);
1061   GNUNET_CONTAINER_DLL_remove (clients_head,
1062                                clients_tail,
1063                                tc);
1064   switch (tc->type)
1065   {
1066   case CT_NONE:
1067     break;
1068   case CT_CORE:
1069     {
1070       struct PendingMessage *pm;
1071
1072       while (NULL != (pm = tc->details.core.pending_msg_head))
1073       {
1074         GNUNET_CONTAINER_MDLL_remove (client,
1075                                       tc->details.core.pending_msg_head,
1076                                       tc->details.core.pending_msg_tail,
1077                                       pm);
1078         pm->client = NULL;
1079       }
1080     }
1081     break;
1082   case CT_MONITOR:
1083     break;
1084   case CT_COMMUNICATOR:
1085     {
1086       struct GNUNET_ATS_Session *q;
1087       struct AddressListEntry *ale;
1088
1089       while (NULL != (q = tc->details.communicator.session_head))
1090         free_queue (q);
1091       while (NULL != (ale = tc->details.communicator.addr_head))
1092         free_address_list_entry (ale);
1093       GNUNET_free (tc->details.communicator.address_prefix);
1094     }
1095     break;
1096   }
1097   GNUNET_free (tc);
1098 }
1099
1100
1101 /**
1102  * Iterator telling new CORE client about all existing
1103  * connections to peers.
1104  *
1105  * @param cls the new `struct TransportClient`
1106  * @param pid a connected peer
1107  * @param value the `struct Neighbour` with more information
1108  * @return #GNUNET_OK (continue to iterate)
1109  */
1110 static int
1111 notify_client_connect_info (void *cls,
1112                             const struct GNUNET_PeerIdentity *pid,
1113                             void *value)
1114 {
1115   struct TransportClient *tc = cls;
1116   struct Neighbour *neighbour = value;
1117
1118   core_send_connect_info (tc,
1119                           pid,
1120                           neighbour->quota_out);
1121   return GNUNET_OK;
1122 }
1123
1124
1125 /**
1126  * Initialize a "CORE" client.  We got a start message from this
1127  * client, so add it to the list of clients for broadcasting of
1128  * inbound messages.
1129  *
1130  * @param cls the client
1131  * @param start the start message that was sent
1132  */
1133 static void
1134 handle_client_start (void *cls,
1135                      const struct StartMessage *start)
1136 {
1137   struct TransportClient *tc = cls;
1138   uint32_t options;
1139
1140   options = ntohl (start->options);
1141   if ( (0 != (1 & options)) &&
1142        (0 !=
1143         memcmp (&start->self,
1144                 &GST_my_identity,
1145                 sizeof (struct GNUNET_PeerIdentity)) ) )
1146   {
1147     /* client thinks this is a different peer, reject */
1148     GNUNET_break (0);
1149     GNUNET_SERVICE_client_drop (tc->client);
1150     return;
1151   }
1152   if (CT_NONE != tc->type)
1153   {
1154     GNUNET_break (0);
1155     GNUNET_SERVICE_client_drop (tc->client);
1156     return;
1157   }
1158   tc->type = CT_CORE;
1159   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
1160                                          &notify_client_connect_info,
1161                                          tc);
1162   GNUNET_SERVICE_client_continue (tc->client);
1163 }
1164
1165
1166 /**
1167  * Client asked for transmission to a peer.  Process the request.
1168  *
1169  * @param cls the client
1170  * @param obm the send message that was sent
1171  */
1172 static int
1173 check_client_send (void *cls,
1174                    const struct OutboundMessage *obm)
1175 {
1176   struct TransportClient *tc = cls;
1177   uint16_t size;
1178   const struct GNUNET_MessageHeader *obmm;
1179
1180   if (CT_CORE != tc->type)
1181   {
1182     GNUNET_break (0);
1183     return GNUNET_SYSERR;
1184   }
1185   size = ntohs (obm->header.size) - sizeof (struct OutboundMessage);
1186   if (size < sizeof (struct GNUNET_MessageHeader))
1187   {
1188     GNUNET_break (0);
1189     return GNUNET_SYSERR;
1190   }
1191   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
1192   if (size != ntohs (obmm->size))
1193   {
1194     GNUNET_break (0);
1195     return GNUNET_SYSERR;
1196   }
1197   return GNUNET_OK;
1198 }
1199
1200
1201 /**
1202  * Send a response to the @a pm that we have processed a
1203  * "send" request with status @a success. We
1204  * transmitted @a bytes_physical on the actual wire.
1205  * Sends a confirmation to the "core" client responsible
1206  * for the original request and free's @a pm.
1207  *
1208  * @param pm handle to the original pending message
1209  * @param success status code, #GNUNET_OK on success, #GNUNET_SYSERR
1210  *          for transmission failure
1211  * @param bytes_physical amount of bandwidth consumed
1212  */
1213 static void
1214 client_send_response (struct PendingMessage *pm,
1215                       int success,
1216                       uint32_t bytes_physical)
1217 {
1218   struct TransportClient *tc = pm->client;
1219   struct Neighbour *target = pm->target;
1220   struct GNUNET_MQ_Envelope *env;
1221   struct SendOkMessage *som;
1222
1223   if (NULL != tc)
1224   {
1225     env = GNUNET_MQ_msg (som,
1226                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
1227     som->success = htonl ((uint32_t) success);
1228     som->bytes_msg = htonl (pm->bytes_msg);
1229     som->bytes_physical = htonl (bytes_physical);
1230     som->peer = target->pid;
1231     GNUNET_MQ_send (tc->mq,
1232                     env);
1233     GNUNET_CONTAINER_MDLL_remove (client,
1234                                   tc->details.core.pending_msg_head,
1235                                   tc->details.core.pending_msg_tail,
1236                                   pm);
1237   }
1238   GNUNET_CONTAINER_MDLL_remove (neighbour,
1239                                 target->pending_msg_head,
1240                                 target->pending_msg_tail,
1241                                 pm);
1242   GNUNET_free (pm);
1243 }
1244
1245
1246 /**
1247  * Client asked for transmission to a peer.  Process the request.
1248  *
1249  * @param cls the client
1250  * @param obm the send message that was sent
1251  */
1252 static void
1253 handle_client_send (void *cls,
1254                     const struct OutboundMessage *obm)
1255 {
1256   struct TransportClient *tc = cls;
1257   struct PendingMessage *pm;
1258   const struct GNUNET_MessageHeader *obmm;
1259   struct Neighbour *target;
1260   uint32_t bytes_msg;
1261
1262   GNUNET_assert (CT_CORE == tc->type);
1263   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
1264   bytes_msg = ntohs (obmm->size);
1265   target = lookup_neighbour (&obm->peer);
1266   if (NULL == target)
1267   {
1268     /* Failure: don't have this peer as a neighbour (anymore).
1269        Might have gone down asynchronously, so this is NOT
1270        a protocol violation by CORE. Still count the event,
1271        as this should be rare. */
1272     struct GNUNET_MQ_Envelope *env;
1273     struct SendOkMessage *som;
1274
1275     env = GNUNET_MQ_msg (som,
1276                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
1277     som->success = htonl (GNUNET_SYSERR);
1278     som->bytes_msg = htonl (bytes_msg);
1279     som->bytes_physical = htonl (0);
1280     som->peer = obm->peer;
1281     GNUNET_MQ_send (tc->mq,
1282                     env);
1283     GNUNET_SERVICE_client_continue (tc->client);
1284     GNUNET_STATISTICS_update (GST_stats,
1285                               "# messages dropped (neighbour unknown)",
1286                               1,
1287                               GNUNET_NO);
1288     return;
1289   }
1290   pm = GNUNET_new (struct PendingMessage);
1291   pm->client = tc;
1292   pm->target = target;
1293   pm->bytes_msg = bytes_msg;
1294   GNUNET_CONTAINER_MDLL_insert (neighbour,
1295                                 target->pending_msg_head,
1296                                 target->pending_msg_tail,
1297                                 pm);
1298   GNUNET_CONTAINER_MDLL_insert (client,
1299                                 tc->details.core.pending_msg_head,
1300                                 tc->details.core.pending_msg_tail,
1301                                 pm);
1302   // FIXME: do the work, final continuation with call to:
1303   client_send_response (pm,
1304                         GNUNET_NO,
1305                         0);
1306 }
1307
1308
1309 /**
1310  * Communicator started.  Test message is well-formed.
1311  *
1312  * @param cls the client
1313  * @param cam the send message that was sent
1314  */
1315 static int
1316 check_communicator_available (void *cls,
1317                               const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
1318 {
1319   struct TransportClient *tc = cls;
1320   uint16_t size;
1321
1322   if (CT_NONE != tc->type)
1323   {
1324     GNUNET_break (0);
1325     return GNUNET_SYSERR;
1326   }
1327   tc->type = CT_COMMUNICATOR;
1328   size = ntohs (cam->header.size) - sizeof (*cam);
1329   if (0 == size)
1330     return GNUNET_OK; /* receive-only communicator */
1331   GNUNET_MQ_check_zero_termination (cam);
1332   return GNUNET_OK;
1333 }
1334
1335
1336 /**
1337  * Communicator started.  Process the request.
1338  *
1339  * @param cls the client
1340  * @param cam the send message that was sent
1341  */
1342 static void
1343 handle_communicator_available (void *cls,
1344                                const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
1345 {
1346   struct TransportClient *tc = cls;
1347   uint16_t size;
1348
1349   size = ntohs (cam->header.size) - sizeof (*cam);
1350   if (0 == size)
1351     return; /* receive-only communicator */
1352   tc->details.communicator.address_prefix
1353     = GNUNET_strdup ((const char *) &cam[1]);
1354   tc->details.communicator.cc
1355     = (enum GNUNET_TRANSPORT_CommunicatorCharacteristics) ntohl (cam->cc);
1356   GNUNET_SERVICE_client_continue (tc->client);
1357 }
1358
1359
1360 /**
1361  * Address of our peer added.  Test message is well-formed.
1362  *
1363  * @param cls the client
1364  * @param aam the send message that was sent
1365  */
1366 static int
1367 check_add_address (void *cls,
1368                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
1369 {
1370   struct TransportClient *tc = cls;
1371
1372   if (CT_COMMUNICATOR != tc->type)
1373   {
1374     GNUNET_break (0);
1375     return GNUNET_SYSERR;
1376   }
1377   GNUNET_MQ_check_zero_termination (aam);
1378   return GNUNET_OK;
1379 }
1380
1381
1382 /**
1383  * Ask peerstore to store our address.
1384  *
1385  * @param cls an `struct AddressListEntry *`
1386  */
1387 static void
1388 store_pi (void *cls);
1389
1390
1391 /**
1392  * Function called when peerstore is done storing our address.
1393  */
1394 static void
1395 peerstore_store_cb (void *cls,
1396                     int success)
1397 {
1398   struct AddressListEntry *ale = cls;
1399
1400   ale->sc = NULL;
1401   if (GNUNET_YES != success)
1402     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1403                 "Failed to store our own address `%s' in peerstore!\n",
1404                 ale->address);
1405   /* refresh period is 1/4 of expiration time, that should be plenty
1406      without being excessive. */
1407   ale->st = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_divide (ale->expiration,
1408                                                                        4ULL),
1409                                           &store_pi,
1410                                           ale);
1411 }
1412
1413
1414 /**
1415  * Ask peerstore to store our address.
1416  *
1417  * @param cls an `struct AddressListEntry *`
1418  */
1419 static void
1420 store_pi (void *cls)
1421 {
1422   struct AddressListEntry *ale = cls;
1423   void *addr;
1424   size_t addr_len;
1425   struct GNUNET_TIME_Absolute expiration;
1426
1427   ale->st = NULL;
1428   expiration = GNUNET_TIME_relative_to_absolute (ale->expiration);
1429   GNUNET_HELLO_sign_address (ale->address,
1430                              ale->nt,
1431                              expiration,
1432                              GST_my_private_key,
1433                              &addr,
1434                              &addr_len);
1435   ale->sc = GNUNET_PEERSTORE_store (peerstore,
1436                                     "transport",
1437                                     &GST_my_identity,
1438                                     GNUNET_HELLO_PEERSTORE_KEY,
1439                                     addr,
1440                                     addr_len,
1441                                     expiration,
1442                                     GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
1443                                     &peerstore_store_cb,
1444                                     ale);
1445   GNUNET_free (addr);
1446   if (NULL == ale->sc)
1447   {
1448     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1449                 "Failed to store our address `%s' with peerstore\n",
1450                 ale->address);
1451     ale->st = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
1452                                             &store_pi,
1453                                             ale);
1454   }
1455 }
1456
1457
1458 /**
1459  * Address of our peer added.  Process the request.
1460  *
1461  * @param cls the client
1462  * @param aam the send message that was sent
1463  */
1464 static void
1465 handle_add_address (void *cls,
1466                     const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
1467 {
1468   struct TransportClient *tc = cls;
1469   struct AddressListEntry *ale;
1470   size_t slen;
1471
1472   slen = ntohs (aam->header.size) - sizeof (*aam);
1473   ale = GNUNET_malloc (sizeof (struct AddressListEntry) + slen);
1474   ale->tc = tc;
1475   ale->address = (const char *) &ale[1];
1476   ale->expiration = GNUNET_TIME_relative_ntoh (aam->expiration);
1477   ale->aid = aam->aid;
1478   ale->nt = (enum GNUNET_NetworkType) ntohl (aam->nt);
1479   memcpy (&ale[1],
1480           &aam[1],
1481           slen);
1482   GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
1483                                tc->details.communicator.addr_tail,
1484                                ale);
1485   ale->st = GNUNET_SCHEDULER_add_now (&store_pi,
1486                                       ale);
1487   GNUNET_SERVICE_client_continue (tc->client);
1488 }
1489
1490
1491 /**
1492  * Address of our peer deleted.  Process the request.
1493  *
1494  * @param cls the client
1495  * @param dam the send message that was sent
1496  */
1497 static void
1498 handle_del_address (void *cls,
1499                     const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
1500 {
1501   struct TransportClient *tc = cls;
1502
1503   if (CT_COMMUNICATOR != tc->type)
1504   {
1505     GNUNET_break (0);
1506     GNUNET_SERVICE_client_drop (tc->client);
1507     return;
1508   }
1509   for (struct AddressListEntry *ale = tc->details.communicator.addr_head;
1510        NULL != ale;
1511        ale = ale->next)
1512   {
1513     if (dam->aid != ale->aid)
1514       continue;
1515     GNUNET_assert (ale->tc == tc);
1516     free_address_list_entry (ale);
1517     GNUNET_SERVICE_client_continue (tc->client);
1518   }
1519   GNUNET_break (0);
1520   GNUNET_SERVICE_client_drop (tc->client);
1521 }
1522
1523
1524 /**
1525  * Client notified us about transmission from a peer.  Process the request.
1526  *
1527  * @param cls the client
1528  * @param obm the send message that was sent
1529  */
1530 static int
1531 check_incoming_msg (void *cls,
1532                     const struct GNUNET_TRANSPORT_IncomingMessage *im)
1533 {
1534   struct TransportClient *tc = cls;
1535   uint16_t size;
1536   const struct GNUNET_MessageHeader *obmm;
1537
1538   if (CT_COMMUNICATOR != tc->type)
1539   {
1540     GNUNET_break (0);
1541     return GNUNET_SYSERR;
1542   }
1543   size = ntohs (im->header.size) - sizeof (*im);
1544   if (size < sizeof (struct GNUNET_MessageHeader))
1545   {
1546     GNUNET_break (0);
1547     return GNUNET_SYSERR;
1548   }
1549   obmm = (const struct GNUNET_MessageHeader *) &im[1];
1550   if (size != ntohs (obmm->size))
1551   {
1552     GNUNET_break (0);
1553     return GNUNET_SYSERR;
1554   }
1555   return GNUNET_OK;
1556 }
1557
1558
1559 /**
1560  * Incoming meessage.  Process the request.
1561  *
1562  * @param cls the client
1563  * @param im the send message that was received
1564  */
1565 static void
1566 handle_incoming_msg (void *cls,
1567                      const struct GNUNET_TRANSPORT_IncomingMessage *im)
1568 {
1569   struct TransportClient *tc = cls;
1570
1571   GNUNET_SERVICE_client_continue (tc->client);
1572 }
1573
1574
1575 /**
1576  * New queue became available.  Check message.
1577  *
1578  * @param cls the client
1579  * @param aqm the send message that was sent
1580  */
1581 static int
1582 check_add_queue_message (void *cls,
1583                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
1584 {
1585   struct TransportClient *tc = cls;
1586
1587   if (CT_COMMUNICATOR != tc->type)
1588   {
1589     GNUNET_break (0);
1590     return GNUNET_SYSERR;
1591   }
1592   GNUNET_MQ_check_zero_termination (aqm);
1593   return GNUNET_OK;
1594 }
1595
1596
1597 /**
1598  * Bandwidth tracker informs us that the delay until we
1599  * can transmit again changed.
1600  *
1601  * @param cls a `struct GNUNET_ATS_Session` for which the delay changed
1602  */
1603 static void
1604 tracker_update_cb (void *cls)
1605 {
1606   struct GNUNET_ATS_Session *queue = cls;
1607
1608   // FIXME: re-schedule transmission tasks if applicable!
1609 }
1610
1611
1612 /**
1613  * Bandwidth tracker informs us that excessive bandwidth was allocated
1614  * which is not being used.
1615  *
1616  * @param cls a `struct GNUNET_ATS_Session` for which the excess was noted
1617  */
1618 static void
1619 tracker_excess_cb (void *cls)
1620 {
1621   /* FIXME: what do we do? */
1622 }
1623
1624
1625 /**
1626  * New queue became available.  Process the request.
1627  *
1628  * @param cls the client
1629  * @param aqm the send message that was sent
1630  */
1631 static void
1632 handle_add_queue_message (void *cls,
1633                           const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
1634 {
1635   struct TransportClient *tc = cls;
1636   struct GNUNET_ATS_Session *queue;
1637   struct Neighbour *neighbour;
1638   const char *addr;
1639   uint16_t addr_len;
1640
1641   neighbour = lookup_neighbour (&aqm->receiver);
1642   if (NULL == neighbour)
1643   {
1644     neighbour = GNUNET_new (struct Neighbour);
1645     neighbour->pid = aqm->receiver;
1646     GNUNET_assert (GNUNET_OK ==
1647                    GNUNET_CONTAINER_multipeermap_put (neighbours,
1648                                                       &neighbour->pid,
1649                                                       neighbour,
1650                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1651     cores_send_connect_info (&neighbour->pid,
1652                              GNUNET_BANDWIDTH_ZERO);
1653     // FIXME: notify ATS!
1654   }
1655   addr_len = ntohs (aqm->header.size) - sizeof (*aqm);
1656   addr = (const char *) &aqm[1];
1657
1658   queue = GNUNET_malloc (sizeof (struct GNUNET_ATS_Session) + addr_len);
1659   queue->tc = tc;
1660   queue->address = (const char *) &queue[1];
1661   queue->rtt = GNUNET_TIME_UNIT_FOREVER_REL;
1662   queue->qid = aqm->qid;
1663   queue->mtu = ntohl (aqm->mtu);
1664   queue->distance = ntohl (aqm->distance);
1665   queue->nt = (enum GNUNET_NetworkType) ntohl (aqm->nt);
1666   queue->cs = (enum GNUNET_TRANSPORT_ConnectionStatus) ntohl (aqm->cs);
1667   queue->neighbour = neighbour;
1668   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_in,
1669                                   &tracker_update_cb,
1670                                   queue,
1671                                   GNUNET_BANDWIDTH_ZERO,
1672                                   0 /* FIXME: max carry in seconds! */,
1673                                   &tracker_excess_cb,
1674                                   queue);
1675   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_out,
1676                                   &tracker_update_cb,
1677                                   queue,
1678                                   GNUNET_BANDWIDTH_ZERO,
1679                                   0 /* FIXME: max carry in seconds! */,
1680                                   &tracker_excess_cb,
1681                                   queue);
1682   memcpy (&queue[1],
1683           addr,
1684           addr_len);
1685   /* notify monitors about new queue */
1686   {
1687     struct MonitorEvent me = {
1688       .rtt = queue->rtt,
1689       .cs = queue->cs
1690     };
1691
1692     notify_monitors (&neighbour->pid,
1693                      queue->address,
1694                      queue->nt,
1695                      &me);
1696   }
1697   GNUNET_CONTAINER_MDLL_insert (neighbour,
1698                                 neighbour->session_head,
1699                                 neighbour->session_tail,
1700                                 queue);
1701   GNUNET_CONTAINER_MDLL_insert (client,
1702                                 tc->details.communicator.session_head,
1703                                 tc->details.communicator.session_tail,
1704                                 queue);
1705   // FIXME: possibly transmit queued messages?
1706   GNUNET_SERVICE_client_continue (tc->client);
1707 }
1708
1709
1710 /**
1711  * Queue to a peer went down.  Process the request.
1712  *
1713  * @param cls the client
1714  * @param dqm the send message that was sent
1715  */
1716 static void
1717 handle_del_queue_message (void *cls,
1718                           const struct GNUNET_TRANSPORT_DelQueueMessage *dqm)
1719 {
1720   struct TransportClient *tc = cls;
1721
1722   if (CT_COMMUNICATOR != tc->type)
1723   {
1724     GNUNET_break (0);
1725     GNUNET_SERVICE_client_drop (tc->client);
1726     return;
1727   }
1728   for (struct GNUNET_ATS_Session *queue = tc->details.communicator.session_head;
1729        NULL != queue;
1730        queue = queue->next_client)
1731   {
1732     struct Neighbour *neighbour = queue->neighbour;
1733
1734     if ( (dqm->qid != queue->qid) ||
1735          (0 != memcmp (&dqm->receiver,
1736                        &neighbour->pid,
1737                        sizeof (struct GNUNET_PeerIdentity))) )
1738       continue;
1739     free_queue (queue);
1740     GNUNET_SERVICE_client_continue (tc->client);
1741     return;
1742   }
1743   GNUNET_break (0);
1744   GNUNET_SERVICE_client_drop (tc->client);
1745 }
1746
1747
1748 /**
1749  * Message was transmitted.  Process the request.
1750  *
1751  * @param cls the client
1752  * @param sma the send message that was sent
1753  */
1754 static void
1755 handle_send_message_ack (void *cls,
1756                          const struct GNUNET_TRANSPORT_SendMessageToAck *sma)
1757 {
1758   struct TransportClient *tc = cls;
1759
1760   if (CT_COMMUNICATOR != tc->type)
1761   {
1762     GNUNET_break (0);
1763     GNUNET_SERVICE_client_drop (tc->client);
1764     return;
1765   }
1766   GNUNET_SERVICE_client_continue (tc->client);
1767 }
1768
1769
1770 /**
1771  * Iterator telling new MONITOR client about all existing
1772  * queues to peers.
1773  *
1774  * @param cls the new `struct TransportClient`
1775  * @param pid a connected peer
1776  * @param value the `struct Neighbour` with more information
1777  * @return #GNUNET_OK (continue to iterate)
1778  */
1779 static int
1780 notify_client_queues (void *cls,
1781                       const struct GNUNET_PeerIdentity *pid,
1782                       void *value)
1783 {
1784   struct TransportClient *tc = cls;
1785   struct Neighbour *neighbour = value;
1786
1787   GNUNET_assert (CT_MONITOR == tc->type);
1788   for (struct GNUNET_ATS_Session *q = neighbour->session_head;
1789        NULL != q;
1790        q = q->next_neighbour)
1791   {
1792     struct MonitorEvent me = {
1793       .rtt = q->rtt,
1794       .cs = q->cs,
1795       .num_msg_pending = q->num_msg_pending,
1796       .num_bytes_pending = q->num_bytes_pending
1797     };
1798
1799     notify_monitor (tc,
1800                     pid,
1801                     q->address,
1802                     q->nt,
1803                     &me);
1804   }
1805   return GNUNET_OK;
1806 }
1807
1808
1809 /**
1810  * Initialize a monitor client.
1811  *
1812  * @param cls the client
1813  * @param start the start message that was sent
1814  */
1815 static void
1816 handle_monitor_start (void *cls,
1817                       const struct GNUNET_TRANSPORT_MonitorStart *start)
1818 {
1819   struct TransportClient *tc = cls;
1820
1821   if (CT_NONE != tc->type)
1822   {
1823     GNUNET_break (0);
1824     GNUNET_SERVICE_client_drop (tc->client);
1825     return;
1826   }
1827   tc->type = CT_MONITOR;
1828   tc->details.monitor.peer = start->peer;
1829   tc->details.monitor.one_shot = ntohl (start->one_shot);
1830   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
1831                                          &notify_client_queues,
1832                                          tc);
1833   GNUNET_SERVICE_client_mark_monitor (tc->client);
1834   GNUNET_SERVICE_client_continue (tc->client);
1835 }
1836
1837
1838 /**
1839  * Signature of a function called by ATS with the current bandwidth
1840  * allocation to be used as determined by ATS.
1841  *
1842  * @param cls closure, NULL
1843  * @param session session this is about
1844  * @param bandwidth_out assigned outbound bandwidth for the connection,
1845  *        0 to signal disconnect
1846  * @param bandwidth_in assigned inbound bandwidth for the connection,
1847  *        0 to signal disconnect
1848  */
1849 static void
1850 ats_allocation_cb (void *cls,
1851                    struct GNUNET_ATS_Session *session,
1852                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
1853                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in)
1854 {
1855   (void) cls;
1856   GNUNET_BANDWIDTH_tracker_update_quota (&session->tracker_out,
1857                                          bandwidth_out);
1858   GNUNET_BANDWIDTH_tracker_update_quota (&session->tracker_in,
1859                                          bandwidth_in);
1860 }
1861
1862
1863 /**
1864  * Find transport client providing communication service
1865  * for the protocol @a prefix.
1866  *
1867  * @param prefix communicator name
1868  * @return NULL if no such transport client is available
1869  */
1870 static struct TransportClient *
1871 lookup_communicator (const char *prefix)
1872 {
1873   GNUNET_break (0); // FIXME: implement
1874   return NULL;
1875 }
1876
1877
1878 /**
1879  * Signature of a function called by ATS suggesting transport to
1880  * try connecting with a particular address.
1881  *
1882  * @param cls closure, NULL
1883  * @param pid target peer
1884  * @param address the address to try
1885  */
1886 static void
1887 ats_suggestion_cb (void *cls,
1888                    const struct GNUNET_PeerIdentity *pid,
1889                    const char *address)
1890 {
1891   struct TransportClient *tc;
1892   char *prefix;
1893
1894   (void) cls;
1895   prefix = NULL; // FIXME
1896   tc = lookup_communicator (prefix);
1897   if (NULL == tc)
1898   {
1899     // STATS...
1900     return;
1901   }
1902   // FIXME: forward suggestion to tc
1903 }
1904
1905
1906 /**
1907  * Free neighbour entry.
1908  *
1909  * @param cls NULL
1910  * @param pid unused
1911  * @param value a `struct Neighbour`
1912  * @return #GNUNET_OK (always)
1913  */
1914 static int
1915 free_neighbour_cb (void *cls,
1916                    const struct GNUNET_PeerIdentity *pid,
1917                    void *value)
1918 {
1919   struct Neighbour *neighbour = value;
1920
1921   (void) cls;
1922   (void) pid;
1923   GNUNET_break (0); // should this ever happen?
1924   free_neighbour (neighbour);
1925
1926   return GNUNET_OK;
1927 }
1928
1929
1930 /**
1931  * Free ephemeral entry.
1932  *
1933  * @param cls NULL
1934  * @param pid unused
1935  * @param value a `struct Neighbour`
1936  * @return #GNUNET_OK (always)
1937  */
1938 static int
1939 free_ephemeral_cb (void *cls,
1940                    const struct GNUNET_PeerIdentity *pid,
1941                    void *value)
1942 {
1943   struct EphemeralCacheEntry *ece = value;
1944
1945   (void) cls;
1946   (void) pid;
1947   free_ephemeral (ece);
1948   return GNUNET_OK;
1949 }
1950
1951
1952 /**
1953  * Function called when the service shuts down.  Unloads our plugins
1954  * and cancels pending validations.
1955  *
1956  * @param cls closure, unused
1957  */
1958 static void
1959 do_shutdown (void *cls)
1960 {
1961   (void) cls;
1962
1963   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
1964                                          &free_neighbour_cb,
1965                                          NULL);
1966   if (NULL != ats)
1967   {
1968     GNUNET_ATS_transport_done (ats);
1969     ats = NULL;
1970   }
1971   if (NULL != peerstore)
1972   {
1973     GNUNET_PEERSTORE_disconnect (peerstore,
1974                                  GNUNET_NO);
1975     peerstore = NULL;
1976   }
1977   if (NULL != GST_stats)
1978   {
1979     GNUNET_STATISTICS_destroy (GST_stats,
1980                                GNUNET_NO);
1981     GST_stats = NULL;
1982   }
1983   if (NULL != GST_my_private_key)
1984   {
1985     GNUNET_free (GST_my_private_key);
1986     GST_my_private_key = NULL;
1987   }
1988   GNUNET_CONTAINER_multipeermap_destroy (neighbours);
1989   neighbours = NULL;
1990   GNUNET_CONTAINER_multipeermap_iterate (ephemeral_map,
1991                                          &free_ephemeral_cb,
1992                                          NULL);
1993   GNUNET_CONTAINER_multipeermap_destroy (ephemeral_map);
1994   ephemeral_map = NULL;
1995   GNUNET_CONTAINER_heap_destroy (ephemeral_heap);
1996   ephemeral_heap = NULL;
1997 }
1998
1999
2000 /**
2001  * Initiate transport service.
2002  *
2003  * @param cls closure
2004  * @param c configuration to use
2005  * @param service the initialized service
2006  */
2007 static void
2008 run (void *cls,
2009      const struct GNUNET_CONFIGURATION_Handle *c,
2010      struct GNUNET_SERVICE_Handle *service)
2011 {
2012   (void) cls;
2013   /* setup globals */
2014   GST_cfg = c;
2015   neighbours = GNUNET_CONTAINER_multipeermap_create (1024,
2016                                                      GNUNET_YES);
2017   ephemeral_map = GNUNET_CONTAINER_multipeermap_create (32,
2018                                                         GNUNET_YES);
2019   ephemeral_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
2020   GST_my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (GST_cfg);
2021   if (NULL == GST_my_private_key)
2022   {
2023     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2024                 _("Transport service is lacking key configuration settings. Exiting.\n"));
2025     GNUNET_SCHEDULER_shutdown ();
2026     return;
2027   }
2028   GNUNET_CRYPTO_eddsa_key_get_public (GST_my_private_key,
2029                                       &GST_my_identity.public_key);
2030   GNUNET_log(GNUNET_ERROR_TYPE_INFO,
2031              "My identity is `%s'\n",
2032              GNUNET_i2s_full (&GST_my_identity));
2033   GST_stats = GNUNET_STATISTICS_create ("transport",
2034                                         GST_cfg);
2035   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
2036                                  NULL);
2037   peerstore = GNUNET_PEERSTORE_connect (GST_cfg);
2038   if (NULL == peerstore)
2039   {
2040     GNUNET_break (0);
2041     GNUNET_SCHEDULER_shutdown ();
2042     return;
2043   }
2044   ats = GNUNET_ATS_transport_init (GST_cfg,
2045                                    &ats_allocation_cb,
2046                                    NULL,
2047                                    &ats_suggestion_cb,
2048                                    NULL);
2049   if (NULL == ats)
2050   {
2051     GNUNET_break (0);
2052     GNUNET_SCHEDULER_shutdown ();
2053     return;
2054   }
2055 }
2056
2057
2058 /**
2059  * Define "main" method using service macro.
2060  */
2061 GNUNET_SERVICE_MAIN
2062 ("transport",
2063  GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN,
2064  &run,
2065  &client_connect_cb,
2066  &client_disconnect_cb,
2067  NULL,
2068  /* communication with core */
2069  GNUNET_MQ_hd_fixed_size (client_start,
2070                           GNUNET_MESSAGE_TYPE_TRANSPORT_START,
2071                           struct StartMessage,
2072                           NULL),
2073  GNUNET_MQ_hd_var_size (client_send,
2074                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND,
2075                         struct OutboundMessage,
2076                         NULL),
2077  /* communication with communicators */
2078  GNUNET_MQ_hd_var_size (communicator_available,
2079                         GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
2080                         struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
2081                         NULL),
2082  GNUNET_MQ_hd_var_size (add_address,
2083                         GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
2084                         struct GNUNET_TRANSPORT_AddAddressMessage,
2085                         NULL),
2086  GNUNET_MQ_hd_fixed_size (del_address,
2087                           GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
2088                           struct GNUNET_TRANSPORT_DelAddressMessage,
2089                           NULL),
2090  GNUNET_MQ_hd_var_size (incoming_msg,
2091                         GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
2092                         struct GNUNET_TRANSPORT_IncomingMessage,
2093                         NULL),
2094  GNUNET_MQ_hd_var_size (add_queue_message,
2095                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
2096                         struct GNUNET_TRANSPORT_AddQueueMessage,
2097                         NULL),
2098  GNUNET_MQ_hd_fixed_size (del_queue_message,
2099                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
2100                           struct GNUNET_TRANSPORT_DelQueueMessage,
2101                           NULL),
2102  GNUNET_MQ_hd_fixed_size (send_message_ack,
2103                           GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
2104                           struct GNUNET_TRANSPORT_SendMessageToAck,
2105                           NULL),
2106  /* communication with monitors */
2107  GNUNET_MQ_hd_fixed_size (monitor_start,
2108                           GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_START,
2109                           struct GNUNET_TRANSPORT_MonitorStart,
2110                           NULL),
2111  GNUNET_MQ_handler_end ());
2112
2113
2114 /* end of file gnunet-service-transport.c */