gnunet-service-transport.c

Go to the documentation of this file.

/*
   This file is part of GNUnet.
   Copyright (C) 2010-2016, 2018, 2019 GNUnet e.V.
 
   GNUnet is free software: you can redistribute it and/or modify it
   under the terms of the GNU Affero General Public License as published
   by the Free Software Foundation, either version 3 of the License,
   or (at your option) any later version.
 
   GNUnet is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_statistics_service.h"
#include "gnunet_transport_monitor_service.h"
#include "gnunet_peerstore_service.h"
#include "gnunet_nat_service.h"
#include "gnunet_hello_uri_lib.h"
#include "gnunet_signatures.h"
#include "transport.h"
 
#define RING_BUFFER_SIZE 16
 
#define MAX_FC_RETRANSMIT_COUNT 1000
 
#define MAX_CUMMULATIVE_ACKS 64
 
#define FC_NO_CHANGE_REPLY_PROBABILITY 8
 
#define IN_PACKET_SIZE_WITHOUT_MTU 128
 
#define GOODPUT_AGING_SLOTS 4
 
#define DEFAULT_WINDOW_SIZE (128 * 1024)
 
#define MAX_INCOMING_REQUEST 16
 
#define MAX_DV_DISCOVERY_SELECTION 16
 
#define RECV_WINDOW_SIZE 4
 
#define MIN_DV_PATH_LENGTH_FOR_INITIATOR 3
 
#define MAX_DV_HOPS_ALLOWED 16
 
#define MAX_DV_LEARN_PENDING 64
 
#define MAX_DV_PATHS_TO_TARGET 3
 
#define DELAY_WARN_THRESHOLD \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
 
#define DV_FORWARD_TIMEOUT \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 60)
 
#define DEFAULT_ACK_WAIT_DURATION \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
 
#define DV_QUALITY_RTT_THRESHOLD \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
 
#define DV_PATH_VALIDITY_TIMEOUT \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
 
#define BACKCHANNEL_INACTIVITY_TIMEOUT \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
 
#define DV_PATH_DISCOVERY_FREQUENCY \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
 
#define EPHEMERAL_VALIDITY \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
 
#define REASSEMBLY_EXPIRATION \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
 
#define FAST_VALIDATION_CHALLENGE_FREQ \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1)
 
#define MAX_VALIDATION_CHALLENGE_FREQ \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_DAYS, 1)
 
#define ACK_CUMMULATOR_TIMEOUT \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
 
#define DV_LEARN_BASE_FREQUENCY GNUNET_TIME_UNIT_MINUTES
 
#define DV_LEARN_QUALITY_THRESHOLD 100
 
#define MAX_ADDRESS_VALID_UNTIL \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MONTHS, 1)
 
#define ADDRESS_VALIDATION_LIFETIME \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
 
#define MIN_DELAY_ADDRESS_VALIDATION GNUNET_TIME_UNIT_MILLISECONDS
 
#define VALIDATION_RTT_BUFFER_FACTOR 3
 
#define COMMUNICATOR_TOTAL_QUEUE_LIMIT 512
 
#define QUEUE_LENGTH_LIMIT 32
 
#define QUEUE_ENTRY_TIMEOUT \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
 
#define RTT_DIFF  \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
 
GNUNET_NETWORK_STRUCT_BEGIN
 
struct MessageUUIDP
{
  uint64_t uuid GNUNET_PACKED;
};
 
 
struct AcknowledgementUUIDP
{
  struct GNUNET_Uuid value;
};
 
struct TransportBackchannelEncapsulationMessage
{
  struct GNUNET_MessageHeader header;
 
  /* Followed by *another* message header which is the message to
     the communicator */
 
  /* Followed by a 0-terminated name of the communicator */
};
 
 
struct EphemeralConfirmationPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 
  struct GNUNET_TIME_AbsoluteNBO sender_monotonic_time;
 
  struct GNUNET_PeerIdentity target;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral_key;
};
 
 
struct TransportDVBoxPayloadP
{
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  /* Followed by a `struct GNUNET_MessageHeader` with a message
     for the target peer */
};
 
 
struct TransportReliabilityBoxMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t ack_countdown GNUNET_PACKED;
 
  struct AcknowledgementUUIDP ack_uuid;
};
 
 
struct TransportCummulativeAckPayloadP
{
  struct GNUNET_TIME_RelativeNBO ack_delay;
 
  struct AcknowledgementUUIDP ack_uuid;
};
 
 
struct TransportReliabilityAckMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t ack_counter GNUNET_PACKED;
 
  /* followed by any number of `struct TransportCummulativeAckPayloadP`
     messages providing ACKs */
};
 
 
struct TransportFragmentBoxMessage
{
  struct GNUNET_MessageHeader header;
 
  uint16_t frag_off GNUNET_PACKED;
 
  uint16_t msg_size GNUNET_PACKED;
 
  struct AcknowledgementUUIDP ack_uuid;
 
  struct MessageUUIDP msg_uuid;
};
 
 
struct DvInitPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
};
 
 
struct DvHopPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 
  struct GNUNET_PeerIdentity pred;
 
  struct GNUNET_PeerIdentity succ;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
};
 
 
struct DVPathEntryP
{
  struct GNUNET_PeerIdentity hop;
 
  struct GNUNET_CRYPTO_EddsaSignature hop_sig;
};
 
 
struct TransportDVLearnMessage
{
  struct GNUNET_MessageHeader header;
 
  uint16_t num_hops GNUNET_PACKED;
 
  uint16_t bidirectional GNUNET_PACKED;
 
  struct GNUNET_TIME_RelativeNBO non_network_delay;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_EddsaSignature init_sig;
 
  struct GNUNET_PeerIdentity initiator;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  /* Followed by @e num_hops `struct DVPathEntryP` values,
     excluding the initiator of the DV trace; the last entry is the
     current sender; the current peer must not be included. */
};
 
 
struct TransportDVBoxMessage
{
  struct GNUNET_MessageHeader header;
 
  unsigned int without_fc;
 
  uint16_t total_hops GNUNET_PACKED;
 
  uint16_t num_hops GNUNET_PACKED;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral_key;
 
  struct GNUNET_ShortHashCode iv;
 
  struct GNUNET_HashCode hmac;
 
  uint16_t orig_size GNUNET_PACKED;
 
  /* Followed by @e num_hops `struct GNUNET_PeerIdentity` values;
     excluding the @e origin and the current peer, the last must be
     the ultimate target; if @e num_hops is zero, the receiver of this
     message is the ultimate target. */
 
  /* Followed by encrypted, variable-size payload, which
     must begin with a `struct TransportDVBoxPayloadP` */
 
  /* Followed by the actual message, which itself must not be a
     a DV_LEARN or DV_BOX message! */
};
 
 
struct TransportValidationChallengeMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t reserved GNUNET_PACKED;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  struct GNUNET_TIME_AbsoluteNBO sender_time;
};
 
 
struct TransportValidationPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 
  struct GNUNET_TIME_RelativeNBO validity_duration;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
};
 
 
struct TransportValidationResponseMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t reserved GNUNET_PACKED;
 
  struct GNUNET_CRYPTO_EddsaSignature signature;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  struct GNUNET_TIME_AbsoluteNBO origin_time;
 
  struct GNUNET_TIME_RelativeNBO validity_duration;
};
 
struct TransportGlobalNattedAddress
{
  unsigned int address_length;
 
  /* Followed by @e address_length bytes of the address. */
};
 
struct TransportFlowControlMessage
{
  struct GNUNET_MessageHeader header;
 
  uint32_t seq GNUNET_PACKED;
 
  uint64_t inbound_window_size GNUNET_PACKED;
 
  uint64_t outbound_sent GNUNET_PACKED;
 
  uint64_t outbound_window_size GNUNET_PACKED;
 
  struct GNUNET_TIME_AbsoluteNBO sender_time;
 
  struct GNUNET_TIME_RelativeNBO rtt;
 
  unsigned int sync_ready;
 
  unsigned int number_of_addresses;
 
  size_t size_of_addresses;
 
  /* Followed by @e number_of_addresses struct TransportGlobalNattedAddress. */
};
 
GNUNET_NETWORK_STRUCT_END
 
 
enum ClientType
{
  CT_NONE = 0,
 
  CT_CORE = 1,
 
  CT_MONITOR = 2,
 
  CT_COMMUNICATOR = 3,
 
  CT_APPLICATION = 4
};
 
 
enum RouteMessageOptions
{
  RMO_NONE = 0,
 
  RMO_DV_ALLOWED = 1,
 
  RMO_UNCONFIRMED_ALLOWED = 2,
 
  RMO_ANYTHING_GOES = (RMO_DV_ALLOWED | RMO_UNCONFIRMED_ALLOWED),
 
  RMO_REDUNDANT = 4
};
 
 
struct LearnLaunchEntry
{
  struct LearnLaunchEntry *prev;
 
  struct LearnLaunchEntry *next;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  struct GNUNET_TIME_Absolute launch_time;
};
 
 
struct TransmissionHistoryEntry
{
  uint64_t bytes_sent;
 
  uint64_t bytes_received;
};
 
 
struct PerformanceData
{
  struct GNUNET_TIME_Relative aged_rtt;
 
  struct TransmissionHistoryEntry the[GOODPUT_AGING_SLOTS];
 
  unsigned int last_age;
};
 
 
struct TransportClient;
 
struct Neighbour;
 
struct DistanceVector;
 
struct Queue;
 
struct PendingMessage;
 
struct DistanceVectorHop;
 
struct VirtualLink;
 
 
struct CommunicatorMessageContext
{
  struct CommunicatorMessageContext *next;
 
  struct CommunicatorMessageContext *prev;
 
  struct TransportClient *tc;
 
  struct GNUNET_TRANSPORT_IncomingMessage im;
 
  const struct GNUNET_MessageHeader *mh;
 
  uint16_t total_hops;
 
  unsigned int continue_send;
};
 
 
struct RingBufferEntry
{
  struct CommunicatorMessageContext *cmc;
 
  struct GNUNET_MessageHeader *mh;
};
 
 
struct CoreSentContext
{
  struct CoreSentContext *next;
 
  struct CoreSentContext *prev;
 
  struct VirtualLink *vl;
 
  uint16_t size;
 
  uint16_t isize;
};
 
 
struct ReassemblyContext
{
  struct MessageUUIDP msg_uuid;
 
  struct VirtualLink *virtual_link;
 
  struct GNUNET_CONTAINER_HeapNode *hn;
 
  uint8_t *bitfield;
 
  struct GNUNET_TIME_Absolute reassembly_timeout;
 
  struct GNUNET_TIME_Absolute last_frag;
 
  uint16_t msg_size;
 
  uint16_t msg_missing;
 
  /* Followed by @e msg_size bytes of the (partially) defragmented original
   * message */
 
  /* Followed by @e bitfield data */
};
 
 
struct VirtualLink
{
  struct GNUNET_PeerIdentity target;
 
  struct GNUNET_CONTAINER_MultiHashMap32 *reassembly_map;
 
  struct GNUNET_CONTAINER_Heap *reassembly_heap;
 
  struct GNUNET_SCHEDULER_Task *reassembly_timeout_task;
 
  struct CommunicatorMessageContext *cmc_head;
 
  struct CommunicatorMessageContext *cmc_tail;
 
  struct PendingMessage *pending_msg_head;
 
  struct PendingMessage *pending_msg_tail;
 
  struct CoreSentContext *csc_tail;
 
  struct CoreSentContext *csc_head;
 
  struct GNUNET_SCHEDULER_Task *visibility_task;
 
  struct GNUNET_SCHEDULER_Task *fc_retransmit_task;
 
  struct GNUNET_StartBurstCls *sb_cls;
 
  char *burst_addr;
 
  unsigned int fc_retransmit_count;
 
  unsigned int confirmed;
 
  struct Neighbour *n;
 
  struct DistanceVector *dv;
 
  struct GNUNET_TIME_Absolute n_challenge_time;
 
  struct GNUNET_TIME_Absolute last_fc_transmission;
 
  struct GNUNET_TIME_Absolute last_fc_timestamp;
 
  struct GNUNET_TIME_Relative last_fc_rtt;
 
  struct GNUNET_TIME_Relative other_rtt;
 
  struct GNUNET_PEERSTORE_IterateContext *ic;
 
  uint64_t message_uuid_ctr;
 
  uint64_t available_fc_window_size;
 
  uint64_t incoming_fc_window_size_ram;
 
  uint64_t incoming_fc_window_size;
 
  uint64_t incoming_fc_window_size_used;
 
  int64_t incoming_fc_window_size_loss;
 
  uint64_t outbound_fc_window_size;
 
  uint64_t outbound_fc_window_size_used;
 
  uint64_t last_outbound_window_size_received;
 
  uint32_t fc_seq_gen;
 
  uint32_t last_fc_seq;
 
  int core_recv_window;
 
  enum GNUNET_GenericReturnValue sync_ready;
};
 
 
struct PendingAcknowledgement
{
  struct PendingAcknowledgement *next_pm;
 
  struct PendingAcknowledgement *prev_pm;
 
  struct PendingAcknowledgement *next_queue;
 
  struct PendingAcknowledgement *prev_queue;
 
  struct PendingAcknowledgement *next_dvh;
 
  struct PendingAcknowledgement *prev_dvh;
 
  struct PendingAcknowledgement *next_pa;
 
  struct PendingAcknowledgement *prev_pa;
 
  struct AcknowledgementUUIDP ack_uuid;
 
  struct PendingMessage *pm;
 
  struct DistanceVectorHop *dvh;
 
  struct Queue *queue;
 
  struct GNUNET_TIME_Absolute transmission_time;
 
  uint16_t message_size;
 
  unsigned int num_send;
};
 
 
struct DistanceVectorHop
{
  struct DistanceVectorHop *next_dv;
 
  struct DistanceVectorHop *prev_dv;
 
  struct DistanceVectorHop *next_neighbour;
 
  struct DistanceVectorHop *prev_neighbour;
 
  struct PendingAcknowledgement *pa_head;
 
  struct PendingAcknowledgement *pa_tail;
 
  struct Neighbour *next_hop;
 
  struct DistanceVector *dv;
 
  const struct GNUNET_PeerIdentity *path;
 
  struct GNUNET_TIME_Absolute timeout;
 
  struct GNUNET_TIME_Absolute path_valid_until;
 
  struct PerformanceData pd;
 
  unsigned int distance;
};
 
 
struct DistanceVector
{
  struct GNUNET_PeerIdentity target;
 
  struct DistanceVectorHop *dv_head;
 
  struct DistanceVectorHop *dv_tail;
 
  struct GNUNET_SCHEDULER_Task *timeout_task;
 
  struct VirtualLink *vl;
 
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
  struct GNUNET_TIME_Absolute ephemeral_validity;
 
  struct GNUNET_TIME_Absolute monotime;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral_key;
 
  struct GNUNET_ShortHashCode *km;
};
 
 
struct QueueEntry
{
  struct QueueEntry *next;
 
  struct QueueEntry *prev;
 
  struct Queue *queue;
 
  struct PendingMessage *pm;
 
  uint64_t mid;
 
  struct GNUNET_TIME_Absolute creation_timestamp;
};
 
 
struct Queue
{
  struct Queue *next_neighbour;
 
  struct Queue *prev_neighbour;
 
  struct Queue *prev_client;
 
  struct Queue *next_client;
 
  struct PendingAcknowledgement *pa_head;
 
  struct PendingAcknowledgement *pa_tail;
 
  struct QueueEntry *queue_head;
 
  struct QueueEntry *queue_tail;
 
  struct Neighbour *neighbour;
 
  struct TransportClient *tc;
 
  const char *address;
 
  unsigned int unlimited_length;
 
  struct GNUNET_SCHEDULER_Task *transmit_task;
 
  struct GNUNET_TIME_Absolute validated_until;
 
  struct PerformanceData pd;
 
  struct GNUNET_PEERSTORE_Monitor *mo;
 
  uint64_t mid_gen;
 
  uint32_t qid;
 
  uint32_t mtu;
 
  uint32_t num_msg_pending;
 
  uint32_t num_bytes_pending;
 
  unsigned int queue_length;
 
  uint64_t q_capacity;
 
  uint32_t priority;
 
  enum GNUNET_NetworkType nt;
 
  enum GNUNET_TRANSPORT_ConnectionStatus cs;
 
  int idle;
 
  enum GNUNET_GenericReturnValue is_global_natted;
};
 
 
struct Neighbour
{
  struct GNUNET_PeerIdentity pid;
 
  struct DistanceVectorHop *dv_head;
 
  struct DistanceVectorHop *dv_tail;
 
  struct Queue *queue_head;
 
  struct Queue *queue_tail;
 
  struct GNUNET_PEERSTORE_IterateContext *get;
 
  struct GNUNET_PEERSTORE_StoreContext *sc;
 
  struct VirtualLink *vl;
 
  struct GNUNET_TIME_Absolute last_dv_learn_monotime;
 
  int dv_monotime_available;
 
  struct GNUNET_CONTAINER_MultiPeerMap *natted_addresses;
 
  unsigned int number_of_addresses;
 
  size_t size_of_global_addresses;
 
  enum GNUNET_GenericReturnValue is_global_natted;
};
 
 
struct IncomingRequest
{
  struct IncomingRequest *next;
 
  struct IncomingRequest *prev;
 
  struct GNUNET_PEERSTORE_Monitor *nc;
 
  struct GNUNET_PeerIdentity pid;
};
 
 
struct PeerRequest
{
  struct GNUNET_PeerIdentity pid;
 
  struct TransportClient *tc;
 
  struct GNUNET_PEERSTORE_Monitor *nc;
 
  enum GNUNET_MQ_PriorityPreferences pk;
 
  struct GNUNET_BANDWIDTH_Value32NBO bw;
};
 
 
enum PendingMessageType
{
  PMT_CORE = 0,
 
  PMT_FRAGMENT_BOX = 1,
 
  PMT_RELIABILITY_BOX = 2,
 
  PMT_DV_BOX = 3
};
 
 
struct PendingMessage
{
  struct PendingMessage *next_vl;
 
  struct PendingMessage *prev_vl;
 
  struct PendingMessage *next_client;
 
  struct PendingMessage *prev_client;
 
  struct PendingMessage *next_frag;
 
  struct PendingMessage *prev_frag;
 
  struct PendingAcknowledgement *pa_head;
 
  struct PendingAcknowledgement *pa_tail;
 
  struct PendingMessage *bpm;
 
  struct VirtualLink *vl;
 
  struct GNUNET_PeerIdentity target;
 
  struct QueueEntry *qe;
 
  struct TransportClient *client;
 
  struct PendingMessage *head_frag;
 
  struct PendingMessage *tail_frag;
 
  struct PendingMessage *frag_parent;
 
  struct GNUNET_TIME_Absolute timeout;
 
  struct GNUNET_TIME_Absolute next_attempt;
 
  struct MessageUUIDP msg_uuid;
 
  uint64_t logging_uuid;
 
  enum PendingMessageType pmt;
 
  enum GNUNET_MQ_PriorityPreferences prefs;
 
  struct DistanceVectorHop *used_dvh;
 
  uint16_t bytes_msg;
 
  uint16_t frag_off;
 
  uint32_t frags_in_flight;
 
  uint32_t frags_in_flight_round;
 
  uint16_t frag_count;
 
  int16_t msg_uuid_set;
 
  /* Followed by @e bytes_msg to transmit */
};
 
 
struct TransportCummulativeAckPayload
{
  struct GNUNET_TIME_Absolute receive_time;
 
  struct AcknowledgementUUIDP ack_uuid;
};
 
 
struct AcknowledgementCummulator
{
  struct GNUNET_PeerIdentity target;
 
  struct TransportCummulativeAckPayload ack_uuids[MAX_CUMMULATIVE_ACKS];
 
  struct GNUNET_SCHEDULER_Task *task;
 
  struct GNUNET_TIME_Absolute min_transmission_time;
 
  uint32_t ack_counter;
 
  unsigned int num_acks;
};
 
 
struct AddressListEntry
{
  struct AddressListEntry *next;
 
  struct AddressListEntry *prev;
 
  struct TransportClient *tc;
 
  struct GNUNET_PEERSTORE_StoreHelloContext *shc;
 
  const char *address;
 
  void *signed_address;
 
  size_t signed_address_len;
 
  struct GNUNET_PEERSTORE_StoreContext *sc;
 
  struct GNUNET_SCHEDULER_Task *st;
 
  struct GNUNET_TIME_Relative expiration;
 
  uint32_t aid;
 
  enum GNUNET_NetworkType nt;
};
 
 
struct TransportClient
{
  struct TransportClient *next;
 
  struct TransportClient *prev;
 
  struct GNUNET_SERVICE_Client *client;
 
  struct GNUNET_MQ_Handle *mq;
 
  enum ClientType type;
 
  union
  {
    struct
    {
      struct PendingMessage *pending_msg_head;
 
      struct PendingMessage *pending_msg_tail;
    } core;
 
    struct
    {
      struct GNUNET_PeerIdentity peer;
 
      int one_shot;
    } monitor;
 
 
    struct
    {
      char *address_prefix;
 
      struct Queue *queue_head;
 
      struct Queue *queue_tail;
 
      struct AddressListEntry *addr_head;
 
      struct AddressListEntry *addr_tail;
 
      unsigned int total_queue_length;
 
      struct GNUNET_SCHEDULER_Task *free_queue_entry_task;
 
      enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
 
      enum GNUNET_GenericReturnValue can_burst;
 
    } communicator;
 
    struct
    {
      struct GNUNET_CONTAINER_MultiPeerMap *requests;
    } application;
  } details;
};
 
 
struct ValidationState
{
  struct GNUNET_PeerIdentity pid;
 
  struct GNUNET_TIME_Absolute valid_until;
 
  struct GNUNET_TIME_Absolute validated_until;
 
  struct GNUNET_TIME_Absolute first_challenge_use;
 
  struct GNUNET_TIME_Absolute last_challenge_use;
 
  struct GNUNET_TIME_Absolute next_challenge;
 
  struct GNUNET_TIME_Relative challenge_backoff;
 
  struct GNUNET_TIME_Relative validation_rtt;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  struct GNUNET_HashCode hc;
 
  struct GNUNET_SCHEDULER_Task *revalidation_task;
 
  char *address;
 
  struct GNUNET_CONTAINER_HeapNode *hn;
 
  struct GNUNET_PEERSTORE_StoreContext *sc;
 
  uint32_t last_window_consum_limit;
 
  int awaiting_queue;
};
 
 
struct Backtalker
{
  struct GNUNET_PeerIdentity pid;
 
  struct GNUNET_TIME_Absolute monotonic_time;
 
  struct GNUNET_TIME_Absolute timeout;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation last_ephemeral;
 
  struct GNUNET_SCHEDULER_Task *task;
 
  struct CommunicatorMessageContext *cmc;
 
  struct GNUNET_PEERSTORE_IterateContext *get;
 
  struct GNUNET_PEERSTORE_StoreContext *sc;
 
  size_t body_size;
};
 
static struct RingBufferEntry *ring_buffer[RING_BUFFER_SIZE];
 
static unsigned int ring_buffer_head;
 
static unsigned int is_ring_buffer_full;
 
static struct PendingMessage *ring_buffer_dv[RING_BUFFER_SIZE];
 
static unsigned int ring_buffer_dv_head;
 
static unsigned int is_ring_buffer_dv_full;
 
static struct TransportClient *clients_head;
 
static struct TransportClient *clients_tail;
 
static struct GNUNET_STATISTICS_Handle *GST_stats;
 
static const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
 
static struct GNUNET_PeerIdentity GST_my_identity;
 
struct GNUNET_HELLO_Builder *GST_my_hello;
 
static struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
 
static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 
static struct GNUNET_CONTAINER_MultiPeerMap *backtalkers;
 
static struct GNUNET_CONTAINER_MultiPeerMap *ack_cummulators;
 
static struct GNUNET_CONTAINER_MultiUuidmap *pending_acks;
 
static struct GNUNET_CONTAINER_MultiPeerMap *dv_routes;
 
static struct GNUNET_CONTAINER_MultiPeerMap *validation_map;
 
static struct GNUNET_CONTAINER_MultiHashMap *revalidation_map;
 
static struct GNUNET_CONTAINER_MultiPeerMap *links;
 
static struct GNUNET_CONTAINER_MultiShortmap *dvlearn_map;
 
static struct LearnLaunchEntry *lle_head = NULL;
 
static struct LearnLaunchEntry *lle_tail = NULL;
 
static struct GNUNET_CONTAINER_Heap *validation_heap;
 
struct GNUNET_NAT_Handle *nh;
 
static struct GNUNET_PEERSTORE_Handle *peerstore;
 
static struct GNUNET_SCHEDULER_Task *dvlearn_task;
 
static struct GNUNET_SCHEDULER_Task *validation_task;
 
static struct IncomingRequest *ir_head;
 
static struct IncomingRequest *ir_tail;
 
static unsigned int ir_total;
 
static unsigned long long logging_uuid_gen;
 
static enum GNUNET_GenericReturnValue burst_running;
 
static struct GNUNET_TIME_Absolute hello_mono_time;
 
static int in_shutdown;
 
static struct GNUNET_SCHEDULER_Task *burst_task;
 
struct GNUNET_SCHEDULER_Task *burst_timeout_task;
 
enum GNUNET_GenericReturnValue use_burst;
 
static unsigned int
get_age ()
{
  struct GNUNET_TIME_Absolute now;
 
  now = GNUNET_TIME_absolute_get ();
  return now.abs_value_us / GNUNET_TIME_UNIT_MINUTES.rel_value_us / 15;
}
 
 
static void
free_incoming_request (struct IncomingRequest *ir)
{
  GNUNET_CONTAINER_DLL_remove (ir_head, ir_tail, ir);
  GNUNET_assert (ir_total > 0);
  ir_total--;
  if (NULL != ir->nc)
    GNUNET_PEERSTORE_monitor_stop (ir->nc);
  ir->nc = NULL;
  GNUNET_free (ir);
}
 
 
static void
free_pending_acknowledgement (struct PendingAcknowledgement *pa)
{
  struct Queue *q = pa->queue;
  struct PendingMessage *pm = pa->pm;
  struct DistanceVectorHop *dvh = pa->dvh;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "free_pending_acknowledgement\n");
  if (NULL != q)
  {
    GNUNET_CONTAINER_MDLL_remove (queue, q->pa_head, q->pa_tail, pa);
    pa->queue = NULL;
  }
  if (NULL != pm)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "remove pa from message\n");
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "remove pa from message %" PRIu64 "\n",
                pm->logging_uuid);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "remove pa from message %u\n",
                pm->pmt);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "remove pa from message %s\n",
                GNUNET_uuid2s (&pa->ack_uuid.value));
    GNUNET_CONTAINER_MDLL_remove (pm, pm->pa_head, pm->pa_tail, pa);
    pa->pm = NULL;
  }
  if (NULL != dvh)
  {
    GNUNET_CONTAINER_MDLL_remove (dvh, dvh->pa_head, dvh->pa_tail, pa);
    pa->queue = NULL;
  }
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multiuuidmap_remove (pending_acks,
                                                       &pa->ack_uuid.value,
                                                       pa));
  GNUNET_free (pa);
}
 
 
static void
free_fragment_tree (struct PendingMessage *root)
{
  struct PendingMessage *frag;
 
  while (NULL != (frag = root->head_frag))
  {
    struct PendingAcknowledgement *pa;
 
    free_fragment_tree (frag);
    while (NULL != (pa = frag->pa_head))
    {
      GNUNET_CONTAINER_MDLL_remove (pm, frag->pa_head, frag->pa_tail, pa);
      pa->pm = NULL;
    }
    GNUNET_CONTAINER_MDLL_remove (frag, root->head_frag, root->tail_frag, frag);
    if (NULL != frag->qe)
    {
      GNUNET_assert (frag == frag->qe->pm);
      frag->qe->pm = NULL;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Free frag %p\n",
                frag);
    GNUNET_free (frag);
  }
}
 
 
static void
free_pending_message (struct PendingMessage *pm)
{
  struct TransportClient *tc = pm->client;
  struct VirtualLink *vl = pm->vl;
  struct PendingAcknowledgement *pa;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Freeing pm %p\n",
              pm);
  if (NULL != tc)
  {
    GNUNET_CONTAINER_MDLL_remove (client,
                                  tc->details.core.pending_msg_head,
                                  tc->details.core.pending_msg_tail,
                                  pm);
  }
  if ((NULL != vl) && (NULL == pm->frag_parent))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Removing pm %" PRIu64 "\n",
                pm->logging_uuid);
    GNUNET_CONTAINER_MDLL_remove (vl,
                                  vl->pending_msg_head,
                                  vl->pending_msg_tail,
                                  pm);
  }
  else if (NULL != pm->frag_parent && PMT_DV_BOX != pm->pmt)
  {
    struct PendingMessage *root = pm->frag_parent;
 
    while (NULL != root->frag_parent && PMT_DV_BOX != root->pmt)
      root = root->frag_parent;
 
    root->frag_count--;
  }
  while (NULL != (pa = pm->pa_head))
  {
    if (NULL == pa)
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "free pending pa  null\n");
    if (NULL == pm->pa_tail)
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "free pending pa_tail null\n");
    if (NULL == pa->prev_pa)
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "free pending pa prev null\n");
    if (NULL == pa->next_pa)
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "free pending pa next null\n");
    GNUNET_CONTAINER_MDLL_remove (pm, pm->pa_head, pm->pa_tail, pa);
    pa->pm = NULL;
  }
 
  free_fragment_tree (pm);
  if (NULL != pm->qe)
  {
    GNUNET_assert (pm == pm->qe->pm);
    pm->qe->pm = NULL;
  }
  if (NULL != pm->bpm)
  {
    free_fragment_tree (pm->bpm);
    if (NULL != pm->bpm->qe)
    {
      struct QueueEntry *qe = pm->bpm->qe;
 
      qe->pm = NULL;
    }
    GNUNET_free (pm->bpm);
  }
 
  GNUNET_free (pm);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Freeing pm done\n");
}
 
 
static void
free_reassembly_context (struct ReassemblyContext *rc)
{
  struct VirtualLink *vl = rc->virtual_link;
 
  GNUNET_assert (rc == GNUNET_CONTAINER_heap_remove_node (rc->hn));
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap32_remove (vl->reassembly_map,
                                                         rc->msg_uuid.uuid,
                                                         rc));
  GNUNET_free (rc);
}
 
 
static void
reassembly_cleanup_task (void *cls)
{
  struct VirtualLink *vl = cls;
  struct ReassemblyContext *rc;
 
  vl->reassembly_timeout_task = NULL;
  while (NULL != (rc = GNUNET_CONTAINER_heap_peek (vl->reassembly_heap)))
  {
    if (0 == GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout)
        .rel_value_us)
    {
      free_reassembly_context (rc);
      continue;
    }
    GNUNET_assert (NULL == vl->reassembly_timeout_task);
    vl->reassembly_timeout_task =
      GNUNET_SCHEDULER_add_at (rc->reassembly_timeout,
                               &reassembly_cleanup_task,
                               vl);
    return;
  }
}
 
 
static int
free_reassembly_cb (void *cls, uint32_t key, void *value)
{
  struct ReassemblyContext *rc = value;
 
  (void) cls;
  (void) key;
  free_reassembly_context (rc);
  return GNUNET_OK;
}
 
 
static void
free_virtual_link (struct VirtualLink *vl)
{
  struct PendingMessage *pm;
  struct CoreSentContext *csc;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "free virtual link %p\n",
              vl);
 
  if (NULL != vl->reassembly_map)
  {
    GNUNET_CONTAINER_multihashmap32_iterate (vl->reassembly_map,
                                             &free_reassembly_cb,
                                             NULL);
    GNUNET_CONTAINER_multihashmap32_destroy (vl->reassembly_map);
    vl->reassembly_map = NULL;
    GNUNET_CONTAINER_heap_destroy (vl->reassembly_heap);
    vl->reassembly_heap = NULL;
  }
  if (NULL != vl->reassembly_timeout_task)
  {
    GNUNET_SCHEDULER_cancel (vl->reassembly_timeout_task);
    vl->reassembly_timeout_task = NULL;
  }
  while (NULL != (pm = vl->pending_msg_head))
    free_pending_message (pm);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (links, &vl->target, vl));
  if (NULL != vl->visibility_task)
  {
    GNUNET_SCHEDULER_cancel (vl->visibility_task);
    vl->visibility_task = NULL;
  }
  if (NULL != vl->fc_retransmit_task)
  {
    GNUNET_SCHEDULER_cancel (vl->fc_retransmit_task);
    vl->fc_retransmit_task = NULL;
  }
  while (NULL != (csc = vl->csc_head))
  {
    GNUNET_CONTAINER_DLL_remove (vl->csc_head, vl->csc_tail, csc);
    GNUNET_assert (vl == csc->vl);
    csc->vl = NULL;
  }
  GNUNET_break (NULL == vl->n);
  GNUNET_break (NULL == vl->dv);
  GNUNET_free (vl);
}
 
 
static void
free_validation_state (struct ValidationState *vs)
{
  if (NULL != vs->revalidation_task)
  {
    GNUNET_SCHEDULER_cancel (vs->revalidation_task);
    vs->revalidation_task = NULL;
  }
  /*memcpy (&hkey,
          &hc,
          sizeof (hkey));*/
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Remove key %s for address %s map size %u contains %u during freeing state\n",
              GNUNET_h2s (&vs->hc),
              vs->address,
              GNUNET_CONTAINER_multihashmap_size (revalidation_map),
              GNUNET_CONTAINER_multihashmap_contains (revalidation_map,
                                                      &vs->hc));
  GNUNET_CONTAINER_multihashmap_remove (revalidation_map, &vs->hc, vs);
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (validation_map, &vs->pid, vs));
  GNUNET_CONTAINER_heap_remove_node (vs->hn);
  vs->hn = NULL;
  if (NULL != vs->sc)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "store cancel\n");
    GNUNET_PEERSTORE_store_cancel (vs->sc);
    vs->sc = NULL;
  }
  GNUNET_free (vs->address);
  GNUNET_free (vs);
}
 
 
static struct Neighbour *
lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
{
  return GNUNET_CONTAINER_multipeermap_get (neighbours, pid);
}
 
 
static struct VirtualLink *
lookup_virtual_link (const struct GNUNET_PeerIdentity *pid)
{
  return GNUNET_CONTAINER_multipeermap_get (links, pid);
}
 
 
struct MonitorEvent
{
  struct GNUNET_TIME_Absolute last_validation;
  struct GNUNET_TIME_Absolute valid_until;
  struct GNUNET_TIME_Absolute next_validation;
 
  struct GNUNET_TIME_Relative rtt;
 
  enum GNUNET_TRANSPORT_ConnectionStatus cs;
 
  uint32_t num_msg_pending;
 
  uint32_t num_bytes_pending;
};
 
 
static void
free_distance_vector_hop (struct DistanceVectorHop *dvh)
{
  struct Neighbour *n = dvh->next_hop;
  struct DistanceVector *dv = dvh->dv;
  struct PendingAcknowledgement *pa;
 
  while (NULL != (pa = dvh->pa_head))
  {
    GNUNET_CONTAINER_MDLL_remove (dvh, dvh->pa_head, dvh->pa_tail, pa);
    pa->dvh = NULL;
  }
  GNUNET_CONTAINER_MDLL_remove (neighbour, n->dv_head, n->dv_tail, dvh);
  GNUNET_CONTAINER_MDLL_remove (dv, dv->dv_head, dv->dv_tail, dvh);
  GNUNET_free (dvh);
}
 
 
static void
check_link_down (void *cls);
 
 
static void
cores_send_disconnect_info (const struct GNUNET_PeerIdentity *pid)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Informing CORE clients about disconnect from %s\n",
              GNUNET_i2s (pid));
  for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
  {
    struct GNUNET_MQ_Envelope *env;
    struct DisconnectInfoMessage *dim;
 
    if (CT_CORE != tc->type)
      continue;
    env = GNUNET_MQ_msg (dim, GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
    dim->peer = *pid;
    GNUNET_MQ_send (tc->mq, env);
  }
}
 
 
static void
free_dv_route (struct DistanceVector *dv)
{
  struct DistanceVectorHop *dvh;
 
  while (NULL != (dvh = dv->dv_head))
    free_distance_vector_hop (dvh);
  if (NULL == dv->dv_head)
  {
    struct VirtualLink *vl;
 
    GNUNET_assert (
      GNUNET_YES ==
      GNUNET_CONTAINER_multipeermap_remove (dv_routes, &dv->target, dv));
    if (NULL != (vl = dv->vl))
    {
      GNUNET_assert (dv == vl->dv);
      vl->dv = NULL;
      if (NULL == vl->n)
      {
        cores_send_disconnect_info (&dv->target);
        free_virtual_link (vl);
      }
      else
      {
        GNUNET_SCHEDULER_cancel (vl->visibility_task);
        vl->visibility_task = GNUNET_SCHEDULER_add_now (&check_link_down, vl);
      }
      dv->vl = NULL;
    }
 
    if (NULL != dv->timeout_task)
    {
      GNUNET_SCHEDULER_cancel (dv->timeout_task);
      dv->timeout_task = NULL;
    }
    GNUNET_free (dv->km);
    GNUNET_free (dv);
  }
}
 
 
static void
notify_monitor (struct TransportClient *tc,
                const struct GNUNET_PeerIdentity *peer,
                const char *address,
                enum GNUNET_NetworkType nt,
                const struct MonitorEvent *me)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_MonitorData *md;
  size_t addr_len = strlen (address) + 1;
 
  env = GNUNET_MQ_msg_extra (md,
                             addr_len,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_DATA);
  md->nt = htonl ((uint32_t) nt);
  md->peer = *peer;
  md->last_validation = GNUNET_TIME_absolute_hton (me->last_validation);
  md->valid_until = GNUNET_TIME_absolute_hton (me->valid_until);
  md->next_validation = GNUNET_TIME_absolute_hton (me->next_validation);
  md->rtt = GNUNET_TIME_relative_hton (me->rtt);
  md->cs = htonl ((uint32_t) me->cs);
  md->num_msg_pending = htonl (me->num_msg_pending);
  md->num_bytes_pending = htonl (me->num_bytes_pending);
  memcpy (&md[1], address, addr_len);
  GNUNET_MQ_send (tc->mq, env);
}
 
 
static void
notify_monitors (const struct GNUNET_PeerIdentity *peer,
                 const char *address,
                 enum GNUNET_NetworkType nt,
                 const struct MonitorEvent *me)
{
  for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
  {
    if (CT_MONITOR != tc->type)
      continue;
    if (tc->details.monitor.one_shot)
      continue;
    if ((GNUNET_NO == GNUNET_is_zero (&tc->details.monitor.peer)) &&
        (0 != GNUNET_memcmp (&tc->details.monitor.peer, peer)))
      continue;
    notify_monitor (tc, peer, address, nt, me);
  }
}
 
 
static void *
client_connect_cb (void *cls,
                   struct GNUNET_SERVICE_Client *client,
                   struct GNUNET_MQ_Handle *mq)
{
  struct TransportClient *tc;
 
  (void) cls;
  tc = GNUNET_new (struct TransportClient);
  tc->client = client;
  tc->mq = mq;
  GNUNET_CONTAINER_DLL_insert (clients_head, clients_tail, tc);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Client %p of type %u connected\n",
              tc,
              tc->type);
  return tc;
}
 
 
static enum GNUNET_GenericReturnValue
remove_global_addresses (void *cls,
                         const struct GNUNET_PeerIdentity *pid,
                         void *value)
{
  (void) cls;
  struct TransportGlobalNattedAddress *tgna = value;
 
  GNUNET_free (tgna);
 
  return GNUNET_OK;
}
 
 
static void
free_neighbour (struct Neighbour *neighbour)
{
  struct DistanceVectorHop *dvh;
  struct VirtualLink *vl;
 
  GNUNET_assert (NULL == neighbour->queue_head);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (neighbours,
                                                       &neighbour->pid,
                                                       neighbour));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Freeing neighbour\n");
  GNUNET_CONTAINER_multipeermap_iterate (neighbour->natted_addresses,
                                         &remove_global_addresses,
                                         NULL);
  GNUNET_CONTAINER_multipeermap_destroy (neighbour->natted_addresses);
  while (NULL != (dvh = neighbour->dv_head))
  {
    struct DistanceVector *dv = dvh->dv;
 
    free_distance_vector_hop (dvh);
    if (NULL == dv->dv_head)
      free_dv_route (dv);
  }
  if (NULL != neighbour->get)
  {
    GNUNET_PEERSTORE_iteration_stop (neighbour->get);
    neighbour->get = NULL;
  }
  if (NULL != neighbour->sc)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "store cancel\n");
    GNUNET_PEERSTORE_store_cancel (neighbour->sc);
    neighbour->sc = NULL;
  }
  if (NULL != (vl = neighbour->vl))
  {
    GNUNET_assert (neighbour == vl->n);
    vl->n = NULL;
    if (NULL == vl->dv)
    {
      cores_send_disconnect_info (&vl->target);
      free_virtual_link (vl);
    }
    else
    {
      GNUNET_SCHEDULER_cancel (vl->visibility_task);
      vl->visibility_task = GNUNET_SCHEDULER_add_now (&check_link_down, vl);
    }
    neighbour->vl = NULL;
  }
  GNUNET_free (neighbour);
}
 
 
static void
core_send_connect_info (struct TransportClient *tc,
                        const struct GNUNET_PeerIdentity *pid)
{
  struct GNUNET_MQ_Envelope *env;
  struct ConnectInfoMessage *cim;
 
  GNUNET_assert (CT_CORE == tc->type);
  env = GNUNET_MQ_msg (cim, GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
  cim->id = *pid;
  GNUNET_MQ_send (tc->mq, env);
}
 
 
static void
cores_send_connect_info (const struct GNUNET_PeerIdentity *pid)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Informing CORE clients about connection to %s\n",
              GNUNET_i2s (pid));
  for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
  {
    if (CT_CORE != tc->type)
      continue;
    core_send_connect_info (tc, pid);
  }
}
 
 
static void
transmit_on_queue (void *cls);
 
 
static unsigned int
check_for_queue_with_higher_prio (struct Queue *queue, struct Queue *queue_head)
{
  for (struct Queue *s = queue_head; NULL != s;
       s = s->next_client)
  {
    if (s->tc->details.communicator.address_prefix !=
        queue->tc->details.communicator.address_prefix)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "queue address %s qid %u compare with queue: address %s qid %u\n",
                  queue->address,
                  queue->qid,
                  s->address,
                  s->qid);
      if ((s->priority > queue->priority) && (0 < s->q_capacity) &&
          (QUEUE_LENGTH_LIMIT > s->queue_length) )
        return GNUNET_YES;
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Lower prio\n");
    }
  }
  return GNUNET_NO;
}
 
 
static void
schedule_transmit_on_queue (struct GNUNET_TIME_Relative delay,
                            struct Queue *queue,
                            enum GNUNET_SCHEDULER_Priority p)
{
  struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
 
  if (queue->validated_until.abs_value_us < now.abs_value_us)
    return;
  if (check_for_queue_with_higher_prio (queue,
                                        queue->tc->details.communicator.
                                        queue_head))
    return;
 
  if (queue->tc->details.communicator.total_queue_length >=
      COMMUNICATOR_TOTAL_QUEUE_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Transmission on queue %s (QID %u) throttled due to communicator queue limit\n",
                queue->address,
                queue->qid);
    GNUNET_STATISTICS_update (
      GST_stats,
      "# Transmission throttled due to communicator queue limit",
      1,
      GNUNET_NO);
    queue->idle = GNUNET_NO;
    return;
  }
  if (queue->queue_length >= QUEUE_LENGTH_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Transmission on queue %s (QID %u) throttled due to communicator queue length limit\n",
                queue->address,
                queue->qid);
    GNUNET_STATISTICS_update (GST_stats,
                              "# Transmission throttled due to queue queue limit",
                              1,
                              GNUNET_NO);
    queue->idle = GNUNET_NO;
    return;
  }
  if (0 == queue->q_capacity)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Transmission on queue %s (QID %u) throttled due to communicator message  has capacity %"
                PRIu64 ".\n",
                queue->address,
                queue->qid,
                queue->q_capacity);
    GNUNET_STATISTICS_update (GST_stats,
                              "# Transmission throttled due to message queue capacity",
                              1,
                              GNUNET_NO);
    queue->idle = GNUNET_NO;
    return;
  }
  /* queue might indeed be ready, schedule it */
  if (NULL != queue->transmit_task)
    GNUNET_SCHEDULER_cancel (queue->transmit_task);
  queue->transmit_task =
    GNUNET_SCHEDULER_add_delayed_with_priority (delay, p, &transmit_on_queue,
                                                queue);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Considering transmission on queue `%s' QID %llu to %s\n",
              queue->address,
              (unsigned long long) queue->qid,
              GNUNET_i2s (&queue->neighbour->pid));
}
 
 
static void
check_link_down (void *cls)
{
  struct VirtualLink *vl = cls;
  struct DistanceVector *dv = vl->dv;
  struct Neighbour *n = vl->n;
  struct GNUNET_TIME_Absolute dvh_timeout;
  struct GNUNET_TIME_Absolute q_timeout;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Checking if link is down\n");
  vl->visibility_task = NULL;
  dvh_timeout = GNUNET_TIME_UNIT_ZERO_ABS;
  if (NULL != dv)
  {
    for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
         pos = pos->next_dv)
      dvh_timeout = GNUNET_TIME_absolute_max (dvh_timeout,
                                              pos->path_valid_until);
    if (0 == GNUNET_TIME_absolute_get_remaining (dvh_timeout).rel_value_us)
    {
      vl->dv->vl = NULL;
      vl->dv = NULL;
    }
  }
  q_timeout = GNUNET_TIME_UNIT_ZERO_ABS;
  for (struct Queue *q = n->queue_head; NULL != q; q = q->next_neighbour)
    q_timeout = GNUNET_TIME_absolute_max (q_timeout, q->validated_until);
  if (0 == GNUNET_TIME_absolute_get_remaining (q_timeout).rel_value_us)
  {
    vl->n->vl = NULL;
    vl->n = NULL;
  }
  if ((NULL == vl->n) && (NULL == vl->dv))
  {
    cores_send_disconnect_info (&vl->target);
    free_virtual_link (vl);
    return;
  }
  vl->visibility_task =
    GNUNET_SCHEDULER_add_at (GNUNET_TIME_absolute_max (q_timeout, dvh_timeout),
                             &check_link_down,
                             vl);
}
 
 
static void
free_queue (struct Queue *queue)
{
  struct Neighbour *neighbour = queue->neighbour;
  struct TransportClient *tc = queue->tc;
  struct MonitorEvent me = { .cs = GNUNET_TRANSPORT_CS_DOWN,
                             .rtt = GNUNET_TIME_UNIT_FOREVER_REL };
  struct QueueEntry *qe;
  int maxxed;
  struct PendingAcknowledgement *pa;
  struct VirtualLink *vl;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Cleaning up queue %u\n", queue->qid);
  if (NULL != queue->mo)
  {
    GNUNET_PEERSTORE_monitor_stop (queue->mo);
    queue->mo = NULL;
  }
  if (NULL != queue->transmit_task)
  {
    GNUNET_SCHEDULER_cancel (queue->transmit_task);
    queue->transmit_task = NULL;
  }
  while (NULL != (pa = queue->pa_head))
  {
    GNUNET_CONTAINER_MDLL_remove (queue, queue->pa_head, queue->pa_tail, pa);
    pa->queue = NULL;
  }
 
  GNUNET_CONTAINER_MDLL_remove (neighbour,
                                neighbour->queue_head,
                                neighbour->queue_tail,
                                queue);
  GNUNET_CONTAINER_MDLL_remove (client,
                                tc->details.communicator.queue_head,
                                tc->details.communicator.queue_tail,
                                queue);
  maxxed = (COMMUNICATOR_TOTAL_QUEUE_LIMIT <=
            tc->details.communicator.total_queue_length);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Cleaning up queue with length %u\n",
              queue->queue_length);
  while (NULL != (qe = queue->queue_head))
  {
    GNUNET_CONTAINER_DLL_remove (queue->queue_head, queue->queue_tail, qe);
    queue->queue_length--;
    tc->details.communicator.total_queue_length--;
    if (NULL != qe->pm)
    {
      GNUNET_assert (qe == qe->pm->qe);
      qe->pm->qe = NULL;
    }
    GNUNET_free (qe);
  }
  GNUNET_assert (0 == queue->queue_length);
  if ((maxxed) && (COMMUNICATOR_TOTAL_QUEUE_LIMIT >
                   tc->details.communicator.total_queue_length))
  {
    /* Communicator dropped below threshold, resume all _other_ queues */
    GNUNET_STATISTICS_update (
      GST_stats,
      "# Transmission throttled due to communicator queue limit",
      -1,
      GNUNET_NO);
    for (struct Queue *s = tc->details.communicator.queue_head; NULL != s;
         s = s->next_client)
      schedule_transmit_on_queue (GNUNET_TIME_UNIT_ZERO,
                                  s,
                                  GNUNET_SCHEDULER_PRIORITY_DEFAULT);
  }
  notify_monitors (&neighbour->pid, queue->address, queue->nt, &me);
  GNUNET_free (queue);
 
  vl = lookup_virtual_link (&neighbour->pid);
  if ((NULL != vl) && (neighbour == vl->n))
  {
    GNUNET_SCHEDULER_cancel (vl->visibility_task);
    check_link_down (vl);
  }
  if (NULL == neighbour->queue_head)
  {
    free_neighbour (neighbour);
  }
}
 
 
static void
free_address_list_entry (struct AddressListEntry *ale)
{
  struct TransportClient *tc = ale->tc;
 
  GNUNET_CONTAINER_DLL_remove (tc->details.communicator.addr_head,
                               tc->details.communicator.addr_tail,
                               ale);
  if (NULL != ale->sc)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "store cancel\n");
    GNUNET_PEERSTORE_store_cancel (ale->sc);
    ale->sc = NULL;
  }
  if (NULL != ale->st)
  {
    GNUNET_SCHEDULER_cancel (ale->st);
    ale->st = NULL;
  }
  if (NULL != ale->signed_address)
    GNUNET_free (ale->signed_address);
  GNUNET_free (ale);
}
 
 
static int
stop_peer_request (void *cls,
                   const struct GNUNET_PeerIdentity *pid,
                   void *value)
{
  struct TransportClient *tc = cls;
  struct PeerRequest *pr = value;
 
  if (NULL != pr->nc)
    GNUNET_PEERSTORE_monitor_stop (pr->nc);
  pr->nc = NULL;
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (tc->details.application.requests,
                                          pid,
                                          pr));
  GNUNET_free (pr);
 
  return GNUNET_OK;
}
 
 
static void
do_shutdown (void *cls);
 
static void
client_disconnect_cb (void *cls,
                      struct GNUNET_SERVICE_Client *client,
                      void *app_ctx)
{
  struct TransportClient *tc = app_ctx;
 
  (void) cls;
  (void) client;
  GNUNET_CONTAINER_DLL_remove (clients_head, clients_tail, tc);
  switch (tc->type)
  {
  case CT_NONE:
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Unknown Client %p disconnected, cleaning up.\n",
                tc);
    break;
 
  case CT_CORE: {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "CORE Client %p disconnected, cleaning up.\n",
                  tc);
 
      struct PendingMessage *pm;
 
      while (NULL != (pm = tc->details.core.pending_msg_head))
      {
        GNUNET_CONTAINER_MDLL_remove (client,
                                      tc->details.core.pending_msg_head,
                                      tc->details.core.pending_msg_tail,
                                      pm);
        pm->client = NULL;
      }
    }
    break;
 
  case CT_MONITOR:
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "MONITOR Client %p disconnected, cleaning up.\n",
                tc);
 
    break;
 
  case CT_COMMUNICATOR: {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "COMMUNICATOR Client %p disconnected, cleaning up.\n",
                  tc);
 
      struct Queue *q;
      struct AddressListEntry *ale;
 
      if (NULL != tc->details.communicator.free_queue_entry_task)
        GNUNET_SCHEDULER_cancel (
          tc->details.communicator.free_queue_entry_task);
      while (NULL != (q = tc->details.communicator.queue_head))
        free_queue (q);
      while (NULL != (ale = tc->details.communicator.addr_head))
        free_address_list_entry (ale);
      GNUNET_free (tc->details.communicator.address_prefix);
    }
    break;
 
  case CT_APPLICATION:
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "APPLICATION Client %p disconnected, cleaning up.\n",
                tc);
 
    GNUNET_CONTAINER_multipeermap_iterate (tc->details.application.requests,
                                           &stop_peer_request,
                                           tc);
    GNUNET_CONTAINER_multipeermap_destroy (tc->details.application.requests);
    break;
  }
  GNUNET_free (tc);
  if ((GNUNET_YES == in_shutdown) && (NULL == clients_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Our last client disconnected\n");
    do_shutdown (cls);
  }
}
 
 
static int
notify_client_connect_info (void *cls,
                            const struct GNUNET_PeerIdentity *pid,
                            void *value)
{
  struct TransportClient *tc = cls;
  struct VirtualLink *vl = value;
 
  if ((NULL == vl) || (GNUNET_NO == vl->confirmed))
    return GNUNET_OK;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Telling new CORE client about existing connection to %s\n",
              GNUNET_i2s (pid));
  core_send_connect_info (tc, pid);
  return GNUNET_OK;
}
 
 
static void
finish_cmc_handling_with_continue (struct CommunicatorMessageContext *cmc,
                                   unsigned
                                   int free_cmc);
 
static enum GNUNET_GenericReturnValue
resume_communicators (void *cls,
                      const struct GNUNET_PeerIdentity *pid,
                      void *value)
{
  struct VirtualLink *vl = value;
  struct CommunicatorMessageContext *cmc;
 
  /* resume communicators */
  while (NULL != (cmc = vl->cmc_tail))
  {
    GNUNET_CONTAINER_DLL_remove (vl->cmc_head, vl->cmc_tail, cmc);
    if (GNUNET_NO == cmc->continue_send)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
  }
  return GNUNET_OK;
}
 
 
static void
handle_client_start (void *cls, const struct StartMessage *start)
{
  struct TransportClient *tc = cls;
  uint32_t options;
 
  options = ntohl (start->options);
  if ((0 != (1 & options)) &&
      (0 != GNUNET_memcmp (&start->self, &GST_my_identity)))
  {
    /* client thinks this is a different peer, reject */
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (tc->client);
    return;
  }
  if (CT_NONE != tc->type)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (tc->client);
    return;
  }
  tc->type = CT_CORE;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "New CORE client with PID %s registered\n",
              GNUNET_i2s (&start->self));
  GNUNET_CONTAINER_multipeermap_iterate (links,
                                         &notify_client_connect_info,
                                         tc);
  GNUNET_CONTAINER_multipeermap_iterate (links,
                                         &resume_communicators,
                                         NULL);
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
static int
check_client_send (void *cls, const struct OutboundMessage *obm)
{
  struct TransportClient *tc = cls;
  uint16_t size;
  const struct GNUNET_MessageHeader *obmm;
 
  if (CT_CORE != tc->type)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  size = ntohs (obm->header.size) - sizeof(struct OutboundMessage);
  if (size < sizeof(struct GNUNET_MessageHeader))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  obmm = (const struct GNUNET_MessageHeader *) &obm[1];
  if (size != ntohs (obmm->size))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
client_send_response (struct PendingMessage *pm)
{
  struct TransportClient *tc = pm->client;
  struct VirtualLink *vl = pm->vl;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "client send response\n");
  if (NULL != tc)
  {
    struct GNUNET_MQ_Envelope *env;
    struct SendOkMessage *so_msg;
 
    env = GNUNET_MQ_msg (so_msg, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
    so_msg->peer = vl->target;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Confirming transmission of <%" PRIu64 "> to %s\n",
                pm->logging_uuid,
                GNUNET_i2s (&vl->target));
    GNUNET_MQ_send (tc->mq, env);
  }
  free_pending_message (pm);
}
 
 
static unsigned int
pick_random_dv_hops (const struct DistanceVector *dv,
                     enum RouteMessageOptions options,
                     struct DistanceVectorHop **hops_array,
                     unsigned int hops_array_length)
{
  uint64_t choices[hops_array_length];
  uint64_t num_dv;
  unsigned int dv_count;
 
  /* Pick random vectors, but weighted by distance, giving more weight
     to shorter vectors */
  num_dv = 0;
  dv_count = 0;
  for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
       pos = pos->next_dv)
  {
    if ((0 == (options & RMO_UNCONFIRMED_ALLOWED)) &&
        (GNUNET_TIME_absolute_get_remaining (pos->path_valid_until)
         .rel_value_us == 0))
      continue;   /* pos unconfirmed and confirmed required */
    num_dv += MAX_DV_HOPS_ALLOWED - pos->distance;
    dv_count++;
  }
  if (0 == dv_count)
    return 0;
  if (dv_count <= hops_array_length)
  {
    dv_count = 0;
    for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
         pos = pos->next_dv)
      hops_array[dv_count++] = pos;
    return dv_count;
  }
  for (unsigned int i = 0; i < hops_array_length; i++)
  {
    int ok = GNUNET_NO;
    while (GNUNET_NO == ok)
    {
      choices[i] =
        GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, num_dv);
      ok = GNUNET_YES;
      for (unsigned int j = 0; j < i; j++)
        if (choices[i] == choices[j])
        {
          ok = GNUNET_NO;
          break;
        }
    }
  }
  dv_count = 0;
  num_dv = 0;
  for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
       pos = pos->next_dv)
  {
    uint32_t delta = MAX_DV_HOPS_ALLOWED - pos->distance;
 
    if ((0 == (options & RMO_UNCONFIRMED_ALLOWED)) &&
        (GNUNET_TIME_absolute_get_remaining (pos->path_valid_until)
         .rel_value_us == 0))
      continue;   /* pos unconfirmed and confirmed required */
    for (unsigned int i = 0; i < hops_array_length; i++)
      if ((num_dv <= choices[i]) && (num_dv + delta > choices[i]))
        hops_array[dv_count++] = pos;
    num_dv += delta;
  }
  return dv_count;
}
 
 
static int
check_communicator_available (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
{
  struct TransportClient *tc = cls;
  uint16_t size;
 
  if (CT_NONE != tc->type)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  tc->type = CT_COMMUNICATOR;
  size = ntohs (cam->header.size) - sizeof(*cam);
  if (0 == size)
    return GNUNET_OK; /* receive-only communicator */
  GNUNET_MQ_check_zero_termination (cam);
  return GNUNET_OK;
}
 
 
static void
finish_cmc_handling_with_continue (struct CommunicatorMessageContext *cmc,
                                   unsigned
                                   int free_cmc)
{
  if (0 != ntohl (cmc->im.fc_on))
  {
    /* send ACK when done to communicator for flow control! */
    struct GNUNET_MQ_Envelope *env;
    struct GNUNET_TRANSPORT_IncomingMessageAck *ack;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Acknowledge message with flow control id %" PRIu64 "\n",
                cmc->im.fc_id);
    env = GNUNET_MQ_msg (ack, GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG_ACK);
    ack->reserved = htonl (0);
    ack->fc_id = cmc->im.fc_id;
    ack->sender = cmc->im.neighbour_sender;
    GNUNET_MQ_send (cmc->tc->mq, env);
  }
 
  GNUNET_SERVICE_client_continue (cmc->tc->client);
 
  if (GNUNET_YES == free_cmc)
  {
    GNUNET_free (cmc);
  }
}
 
 
static void
finish_cmc_handling (struct CommunicatorMessageContext *cmc)
{
  finish_cmc_handling_with_continue (cmc, GNUNET_YES);
}
 
 
static void
handle_client_recv_ok (void *cls, const struct RecvOkMessage *rom)
{
  struct TransportClient *tc = cls;
  struct VirtualLink *vl;
  uint32_t delta;
  struct CommunicatorMessageContext *cmc;
 
  if (CT_CORE != tc->type)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (tc->client);
    return;
  }
  vl = lookup_virtual_link (&rom->peer);
  if ((NULL == vl) || (GNUNET_NO == vl->confirmed))
  {
    GNUNET_STATISTICS_update (GST_stats,
                              "# RECV_OK dropped: virtual link unknown",
                              1,
                              GNUNET_NO);
    GNUNET_SERVICE_client_continue (tc->client);
    return;
  }
  delta = ntohl (rom->increase_window_delta);
  vl->core_recv_window += delta;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "CORE ack receiving message, increased CORE recv window to %d\n",
              vl->core_recv_window);
  GNUNET_SERVICE_client_continue (tc->client);
  if (vl->core_recv_window <= 0)
    return;
  /* resume communicators */
  while (NULL != (cmc = vl->cmc_tail))
  {
    GNUNET_CONTAINER_DLL_remove (vl->cmc_head, vl->cmc_tail, cmc);
    if (GNUNET_NO == cmc->continue_send)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
  }
}
 
 
static void
handle_communicator_available (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
{
  struct TransportClient *tc = cls;
  uint16_t size;
 
  size = ntohs (cam->header.size) - sizeof(*cam);
  if (0 == size)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Receive-only communicator connected\n");
    return;   /* receive-only communicator */
  }
  tc->details.communicator.address_prefix =
    GNUNET_strdup ((const char *) &cam[1]);
  tc->details.communicator.cc =
    (enum GNUNET_TRANSPORT_CommunicatorCharacteristics) ntohl (cam->cc);
  tc->details.communicator.can_burst
    = (enum GNUNET_GenericReturnValue) ntohl (cam->can_burst);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Communicator for peer %s with prefix '%s' connected %s\n",
              GNUNET_i2s (&GST_my_identity),
              tc->details.communicator.address_prefix,
              tc->details.communicator.can_burst ? "can burst" :
              "can not burst");
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
static int
check_communicator_backchannel (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
{
  const struct GNUNET_MessageHeader *inbox;
  const char *is;
  uint16_t msize;
  uint16_t isize;
 
  (void) cls;
  msize = ntohs (cb->header.size) - sizeof(*cb);
  inbox = (const struct GNUNET_MessageHeader *) &cb[1];
  isize = ntohs (inbox->size);
  if (isize >= msize)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  is = (const char *) inbox;
  is += isize;
  msize -= isize;
  GNUNET_assert (0 < msize);
  if ('\0' != is[msize - 1])
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
sign_ephemeral (struct DistanceVector *dv)
{
  struct EphemeralConfirmationPS ec;
 
  dv->monotime = GNUNET_TIME_absolute_get_monotonic (GST_cfg);
  dv->ephemeral_validity =
    GNUNET_TIME_absolute_add (dv->monotime, EPHEMERAL_VALIDITY);
  ec.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL);
  ec.target = dv->target;
  ec.ephemeral_key = dv->ephemeral_key;
  ec.sender_monotonic_time = GNUNET_TIME_absolute_hton (dv->monotime);
  ec.purpose.size = htonl (sizeof(ec));
  GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
                            &ec,
                            &dv->sender_sig);
}
 
 
static void
free_queue_entry (struct QueueEntry *qe,
                  struct TransportClient *tc);
 
 
static void
free_timedout_queue_entry (void *cls)
{
  struct TransportClient *tc = cls;
  struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "freeing timedout queue entries\n");
 
  tc->details.communicator.free_queue_entry_task = NULL;
  for (struct Queue *queue = tc->details.communicator.queue_head; NULL != queue;
       queue = queue->next_client)
  {
    struct QueueEntry *qep = queue->queue_head;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "checking QID %u for timedout queue entries\n",
                queue->qid);
    while (NULL != qep)
    {
      struct QueueEntry *pos = qep;
 
      qep = qep->next;
      struct GNUNET_TIME_Relative diff = GNUNET_TIME_absolute_get_difference (
        pos->creation_timestamp, now);
 
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "diff to now %s \n",
                  GNUNET_TIME_relative2s (diff, GNUNET_NO));
      if (GNUNET_TIME_relative_cmp (QUEUE_ENTRY_TIMEOUT, <, diff))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Freeing timed out QueueEntry with MID %" PRIu64
                    " and QID %u\n",
                    pos->mid,
                    queue->qid);
        free_queue_entry (pos, tc);
      }
    }
  }
}
 
 
static void
queue_send_msg (struct Queue *queue,
                struct PendingMessage *pm,
                const void *payload,
                size_t payload_size)
{
  struct Neighbour *n = queue->neighbour;
  struct GNUNET_TRANSPORT_SendMessageTo *smt;
  struct GNUNET_MQ_Envelope *env;
  struct PendingAcknowledgement *pa;
 
  GNUNET_log (
    GNUNET_ERROR_TYPE_DEBUG,
    "Queueing %u bytes of payload for transmission <%" PRIu64
    "> on queue %llu to %s\n",
    (unsigned int) payload_size,
    (NULL == pm) ? 0 : pm->logging_uuid,
    (unsigned long long) queue->qid,
    GNUNET_i2s (&queue->neighbour->pid));
  env = GNUNET_MQ_msg_extra (smt,
                             payload_size,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
  smt->qid = htonl (queue->qid);
  smt->mid = GNUNET_htonll (queue->mid_gen);
  smt->receiver = n->pid;
  memcpy (&smt[1], payload, payload_size);
  {
    /* Pass the env to the communicator of queue for transmission. */
    struct QueueEntry *qe;
 
    qe = GNUNET_new (struct QueueEntry);
    qe->creation_timestamp = GNUNET_TIME_absolute_get ();
    qe->mid = queue->mid_gen;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Create QueueEntry with MID %" PRIu64
                " and QID %u and prefix %s\n",
                qe->mid,
                queue->qid,
                queue->tc->details.communicator.address_prefix);
    queue->mid_gen++;
    qe->queue = queue;
    if (NULL != pm)
    {
      qe->pm = pm;
      // TODO Why do we have a retransmission. When we know, make decision if we still want this.
      // GNUNET_assert (NULL == pm->qe);
      if (NULL != pm->qe)
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Retransmitting message <%" PRIu64
                    "> remove pm from qe with MID: %llu \n",
                    pm->logging_uuid,
                    (unsigned long long) pm->qe->mid);
        pm->qe->pm = NULL;
      }
      pm->qe = qe;
    }
    GNUNET_assert (CT_COMMUNICATOR == queue->tc->type);
    if (0 == queue->q_capacity)
    {
      // Messages without FC or fragments can get here.
      if (NULL != pm)
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Message %" PRIu64
                    " (pm type %u) was not send because queue has no capacity.\n",
                    pm->logging_uuid,
                    pm->pmt);
        pm->qe = NULL;
      }
      GNUNET_free (env);
      GNUNET_free (qe);
      return;
    }
    GNUNET_CONTAINER_DLL_insert (queue->queue_head, queue->queue_tail, qe);
    queue->queue_length++;
    queue->tc->details.communicator.total_queue_length++;
    if (GNUNET_NO == queue->unlimited_length)
      queue->q_capacity--;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Queue %s with qid %u has capacity %" PRIu64 "\n",
                queue->address,
                queue->qid,
                queue->q_capacity);
    if (COMMUNICATOR_TOTAL_QUEUE_LIMIT ==
        queue->tc->details.communicator.total_queue_length)
      queue->idle = GNUNET_NO;
    if (QUEUE_LENGTH_LIMIT == queue->queue_length)
      queue->idle = GNUNET_NO;
    if (0 == queue->q_capacity)
      queue->idle = GNUNET_NO;
 
    if (GNUNET_NO == queue->idle)
    {
      struct TransportClient *tc = queue->tc;
 
      if (NULL == tc->details.communicator.free_queue_entry_task)
        tc->details.communicator.free_queue_entry_task =
          GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
                                        &
                                        free_timedout_queue_entry,
                                        tc);
    }
    if (NULL != pm && NULL != (pa = pm->pa_head))
    {
      while (pm != pa->pm)
        pa = pa->next_pa;
      pa->num_send++;
    }
    // GNUNET_CONTAINER_multiuuidmap_get (pending_acks, &ack[i].ack_uuid.value);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending message MID %" PRIu64
                " of type %u (%u) and size %lu with MQ %p queue %s (QID %u) pending %"
                PRIu64 "\n",
                GNUNET_ntohll (smt->mid),
                ntohs (((const struct GNUNET_MessageHeader *) payload)->type),
                ntohs (smt->header.size),
                (unsigned long) payload_size,
                queue->tc->mq,
                queue->address,
                queue->qid,
                (NULL == pm) ? 0 : pm->logging_uuid);
    GNUNET_MQ_send (queue->tc->mq, env);
  }
}
 
 
static struct GNUNET_TIME_Relative
route_via_neighbour (const struct Neighbour *n,
                     const struct GNUNET_MessageHeader *hdr,
                     enum RouteMessageOptions options)
{
  struct GNUNET_TIME_Absolute now;
  unsigned int candidates;
  unsigned int sel1;
  unsigned int sel2;
  struct GNUNET_TIME_Relative rtt;
 
  /* Pick one or two 'random' queues from n (under constraints of options) */
  now = GNUNET_TIME_absolute_get ();
  /* FIXME-OPTIMIZE: give queues 'weights' and pick proportional to
     weight in the future; weight could be assigned by observed
     bandwidth (note: not sure if we should do this for this type
     of control traffic though). */
  candidates = 0;
  for (struct Queue *pos = n->queue_head; NULL != pos;
       pos = pos->next_neighbour)
  {
    if ((0 != (options & RMO_UNCONFIRMED_ALLOWED)) ||
        (pos->validated_until.abs_value_us > now.abs_value_us))
      candidates++;
  }
  if (0 == candidates)
  {
    /* This can happen rarely if the last confirmed queue timed
       out just as we were beginning to process this message. */
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Could not route message of type %u to %s: no valid queue\n",
                ntohs (hdr->type),
                GNUNET_i2s (&n->pid));
    GNUNET_STATISTICS_update (GST_stats,
                              "# route selection failed (all no valid queue)",
                              1,
                              GNUNET_NO);
    return GNUNET_TIME_UNIT_FOREVER_REL;
  }
 
  rtt = GNUNET_TIME_UNIT_FOREVER_REL;
  sel1 = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, candidates);
  if (0 == (options & RMO_REDUNDANT))
    sel2 = candidates; /* picks none! */
  else
    sel2 = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, candidates);
  candidates = 0;
  for (struct Queue *pos = n->queue_head; NULL != pos;
       pos = pos->next_neighbour)
  {
    if ((0 != (options & RMO_UNCONFIRMED_ALLOWED)) ||
        (pos->validated_until.abs_value_us > now.abs_value_us))
    {
      if ((sel1 == candidates) || (sel2 == candidates))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Routing message of type %u to %s using %s (#%u)\n",
                    ntohs (hdr->type),
                    GNUNET_i2s (&n->pid),
                    pos->address,
                    (sel1 == candidates) ? 1 : 2);
        rtt = GNUNET_TIME_relative_min (rtt, pos->pd.aged_rtt);
        queue_send_msg (pos, NULL, hdr, ntohs (hdr->size));
      }
      candidates++;
    }
  }
  return rtt;
}
 
 
struct DVKeyState
{
  gcry_cipher_hd_t cipher;
 
  struct
  {
    struct GNUNET_CRYPTO_AuthKey hmac_key;
 
    char aes_key[256 / 8];
 
    char aes_ctr[128 / 8];
  } material;
};
 
 
static void
dv_setup_key_state_from_km (const struct GNUNET_ShortHashCode *km,
                            const struct GNUNET_ShortHashCode *iv,
                            struct DVKeyState *key)
{
  /* must match what we defive from decapsulated key */
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_hkdf_expand (&key->material,
                                            sizeof(key->material),
                                            km,
                                            "gnunet-transport-dv-key",
                                            strlen ("gnunet-transport-dv-key")
                                            ,
                                            km,
                                            sizeof(*km),
                                            iv,
                                            sizeof(*iv),
                                            NULL));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Deriving backchannel key based on KM %s and IV %s\n",
              GNUNET_sh2s (km),
              GNUNET_sh2s (iv));
  GNUNET_assert (0 == gcry_cipher_open (&key->cipher,
                                        GCRY_CIPHER_AES256 /* low level: go for speed */
                                        ,
                                        GCRY_CIPHER_MODE_CTR,
                                        0 /* flags */));
  GNUNET_assert (0 == gcry_cipher_setkey (key->cipher,
                                          &key->material.aes_key,
                                          sizeof(key->material.aes_key)));
  gcry_cipher_setctr (key->cipher,
                      &key->material.aes_ctr,
                      sizeof(key->material.aes_ctr));
}
 
 
static void
dv_hmac (const struct DVKeyState *key,
         struct GNUNET_HashCode *hmac,
         const void *data,
         size_t data_size)
{
  GNUNET_CRYPTO_hmac (&key->material.hmac_key, data, data_size, hmac);
}
 
 
static void
dv_encrypt (struct DVKeyState *key, const void *in, void *dst, size_t in_size)
{
  GNUNET_assert (0 ==
                 gcry_cipher_encrypt (key->cipher, dst, in_size, in, in_size));
}
 
 
static enum GNUNET_GenericReturnValue
dv_decrypt (struct DVKeyState *key,
            void *out,
            const void *ciph,
            size_t out_size)
{
  return (0 ==
          gcry_cipher_decrypt (key->cipher,
                               out, out_size,
                               ciph, out_size)) ? GNUNET_OK : GNUNET_SYSERR;
}
 
 
static void
dv_key_clean (struct DVKeyState *key)
{
  gcry_cipher_close (key->cipher);
  GNUNET_CRYPTO_zero_keys (&key->material, sizeof(key->material));
}
 
 
typedef void (*DVMessageHandler) (void *cls,
                                  struct Neighbour *next_hop,
                                  const struct GNUNET_MessageHeader *hdr,
                                  enum RouteMessageOptions options);
 
static struct GNUNET_TIME_Relative
encapsulate_for_dv (struct DistanceVector *dv,
                    unsigned int num_dvhs,
                    struct DistanceVectorHop **dvhs,
                    const struct GNUNET_MessageHeader *hdr,
                    DVMessageHandler use,
                    void *use_cls,
                    enum RouteMessageOptions options,
                    enum GNUNET_GenericReturnValue without_fc)
{
  struct TransportDVBoxMessage box_hdr;
  struct TransportDVBoxPayloadP payload_hdr;
  uint16_t enc_body_size = ntohs (hdr->size);
  char enc[sizeof(struct TransportDVBoxPayloadP) + enc_body_size] GNUNET_ALIGN;
  struct DVKeyState *key;
  struct GNUNET_TIME_Relative rtt;
  struct GNUNET_ShortHashCode km;
 
  key = GNUNET_new (struct DVKeyState);
  /* Encrypt payload */
  box_hdr.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX);
  box_hdr.total_hops = htons (0);
  box_hdr.without_fc = htons (without_fc);
  // update_ephemeral (dv);
  if (0 ==
      GNUNET_TIME_absolute_get_remaining (dv->ephemeral_validity).rel_value_us)
  {
    GNUNET_CRYPTO_eddsa_kem_encaps (&dv->target.public_key,
                                    &dv->ephemeral_key,
                                    &km);
    dv->km = GNUNET_new (struct GNUNET_ShortHashCode);
    GNUNET_memcpy (dv->km, &km, sizeof(struct GNUNET_ShortHashCode));
    sign_ephemeral (dv);
  }
  box_hdr.ephemeral_key = dv->ephemeral_key;
  payload_hdr.sender_sig = dv->sender_sig;
 
  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
                              &box_hdr.iv,
                              sizeof(box_hdr.iv));
  // We are creating this key, so this must work.
  // FIXME: Possibly also add return values here. We are processing
  // Input from other peers...
  dv_setup_key_state_from_km (dv->km, &box_hdr.iv, key);
  payload_hdr.sender = GST_my_identity;
  payload_hdr.monotonic_time = GNUNET_TIME_absolute_hton (dv->monotime);
  dv_encrypt (key, &payload_hdr, enc, sizeof(payload_hdr));
  dv_encrypt (key,
              hdr,
              &enc[sizeof(struct TransportDVBoxPayloadP)],
              enc_body_size);
  dv_hmac (key, &box_hdr.hmac, enc, sizeof(enc));
  dv_key_clean (key);
  rtt = GNUNET_TIME_UNIT_FOREVER_REL;
  /* For each selected path, take the pre-computed header and body
     and add the path in the middle of the message; then send it. */
  for (unsigned int i = 0; i < num_dvhs; i++)
  {
    struct DistanceVectorHop *dvh = dvhs[i];
    unsigned int num_hops = dvh->distance + 1;
    char buf[sizeof(struct TransportDVBoxMessage)
             + sizeof(struct GNUNET_PeerIdentity) * num_hops
             + sizeof(struct TransportDVBoxPayloadP)
             + enc_body_size] GNUNET_ALIGN;
    struct GNUNET_PeerIdentity *dhops;
 
    box_hdr.header.size = htons (sizeof(buf));
    box_hdr.orig_size = htons (sizeof(buf));
    box_hdr.num_hops = htons (num_hops);
    memcpy (buf, &box_hdr, sizeof(box_hdr));
    dhops = (struct GNUNET_PeerIdentity *) &buf[sizeof(box_hdr)];
    memcpy (dhops,
            dvh->path,
            dvh->distance * sizeof(struct GNUNET_PeerIdentity));
    dhops[dvh->distance] = dv->target;
    if (GNUNET_EXTRA_LOGGING > 0)
    {
      char *path;
 
      path = GNUNET_strdup (GNUNET_i2s (&GST_my_identity));
      for (unsigned int j = 0; j < num_hops; j++)
      {
        char *tmp;
 
        GNUNET_asprintf (&tmp, "%s-%s", path, GNUNET_i2s (&dhops[j]));
        GNUNET_free (path);
        path = tmp;
      }
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Routing message of type %u to %s using DV (#%u/%u) via %s\n",
                  ntohs (hdr->type),
                  GNUNET_i2s (&dv->target),
                  i + 1,
                  num_dvhs,
                  path);
      GNUNET_free (path);
    }
    rtt = GNUNET_TIME_relative_min (rtt, dvh->pd.aged_rtt);
    memcpy (&dhops[num_hops], enc, sizeof(enc));
    use (use_cls,
         dvh->next_hop,
         (const struct GNUNET_MessageHeader *) buf,
         options);
    GNUNET_free (key);
  }
  return rtt;
}
 
 
static void
send_dv_to_neighbour (void *cls,
                      struct Neighbour *next_hop,
                      const struct GNUNET_MessageHeader *hdr,
                      enum RouteMessageOptions options)
{
  (void) cls;
  (void) route_via_neighbour (next_hop, hdr, RMO_UNCONFIRMED_ALLOWED);
}
 
 
static struct GNUNET_TIME_Relative
route_control_message_without_fc (struct VirtualLink *vl,
// route_control_message_without_fc (const struct GNUNET_PeerIdentity *target,
                                  const struct GNUNET_MessageHeader *hdr,
                                  enum RouteMessageOptions options)
{
  // struct VirtualLink *vl;
  struct Neighbour *n;
  struct DistanceVector *dv;
  struct GNUNET_TIME_Relative rtt1;
  struct GNUNET_TIME_Relative rtt2;
  const struct GNUNET_PeerIdentity *target = &vl->target;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Trying to route message of type %u to %s without fc\n",
              ntohs (hdr->type),
              GNUNET_i2s (target));
 
  // TODO Do this elsewhere. vl should be given as parameter to method.
  // vl = lookup_virtual_link (target);
  GNUNET_assert (NULL != vl && GNUNET_YES == vl->confirmed);
  if (NULL == vl)
    return GNUNET_TIME_UNIT_FOREVER_REL;
  n = vl->n;
  dv = (0 != (options & RMO_DV_ALLOWED)) ? vl->dv : NULL;
  if (0 == (options & RMO_UNCONFIRMED_ALLOWED))
  {
    /* if confirmed is required, and we do not have anything
       confirmed, drop respective options */
    if (NULL == n)
      n = lookup_neighbour (target);
    if ((NULL == dv) && (0 != (options & RMO_DV_ALLOWED)))
      dv = GNUNET_CONTAINER_multipeermap_get (dv_routes, target);
  }
  if ((NULL == n) && (NULL == dv))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Cannot route message of type %u to %s: no route\n",
                ntohs (hdr->type),
                GNUNET_i2s (target));
    GNUNET_STATISTICS_update (GST_stats,
                              "# Messages dropped in routing: no acceptable method",
                              1,
                              GNUNET_NO);
    return GNUNET_TIME_UNIT_FOREVER_REL;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Routing message of type %u to %s with options %X\n",
              ntohs (hdr->type),
              GNUNET_i2s (target),
              (unsigned int) options);
  /* If both dv and n are possible and we must choose:
     flip a coin for the choice between the two; for now 50/50 */
  if ((NULL != n) && (NULL != dv) && (0 == (options & RMO_REDUNDANT)))
  {
    if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 2))
      n = NULL;
    else
      dv = NULL;
  }
  if ((NULL != n) && (NULL != dv))
    options &= ~RMO_REDUNDANT; /* We will do one DV and one direct, that's
                                  enough for redundancy, so clear the flag. */
  rtt1 = GNUNET_TIME_UNIT_FOREVER_REL;
  rtt2 = GNUNET_TIME_UNIT_FOREVER_REL;
  if (NULL != n)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Try to route message of type %u to %s without fc via neighbour\n",
                ntohs (hdr->type),
                GNUNET_i2s (target));
    rtt1 = route_via_neighbour (n, hdr, options);
  }
  if (NULL != dv)
  {
    struct DistanceVectorHop *hops[2];
    unsigned int res;
 
    res = pick_random_dv_hops (dv,
                               options,
                               hops,
                               (0 == (options & RMO_REDUNDANT)) ? 1 : 2);
    if (0 == res)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Failed to route message, could not determine DV path\n");
      return rtt1;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "encapsulate_for_dv 1\n");
    rtt2 = encapsulate_for_dv (dv,
                               res,
                               hops,
                               hdr,
                               &send_dv_to_neighbour,
                               NULL,
                               options & (~RMO_REDUNDANT),
                               GNUNET_YES);
  }
  return GNUNET_TIME_relative_min (rtt1, rtt2);
}
 
 
static void
consider_sending_fc (void *cls);
 
static void
task_consider_sending_fc (void *cls)
{
  struct VirtualLink *vl = cls;
  vl->fc_retransmit_task = NULL;
  consider_sending_fc (cls);
}
 
 
static char *
get_address_without_port (const char *address);
 
 
struct AddGlobalAddressesContext
{
  size_t off;
  char *tgnas;
};
 
 
static enum GNUNET_GenericReturnValue
add_global_addresses (void *cls,
                      const struct GNUNET_PeerIdentity *pid,
                      void *value)
{
  struct AddGlobalAddressesContext *ctx = cls;
  struct TransportGlobalNattedAddress *tgna = value;
  char *addr = (char *) &tgna[1];
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sending address %s length %u\n",
              addr,
              ntohl (tgna->address_length));
  GNUNET_memcpy (&(ctx->tgnas[ctx->off]), tgna, sizeof (struct
                                                        TransportGlobalNattedAddress)
                 + ntohl (tgna->address_length));
  ctx->off += sizeof(struct TransportGlobalNattedAddress) + ntohl (tgna->
                                                                   address_length);
 
  return GNUNET_OK;
}
 
 
static struct GNUNET_TIME_Relative
calculate_rtt (struct DistanceVector *dv);
 
 
static void
consider_sending_fc (void *cls)
{
  struct VirtualLink *vl = cls;
  struct GNUNET_TIME_Absolute monotime;
  struct TransportFlowControlMessage *fc;
  struct GNUNET_TIME_Relative duration;
  struct GNUNET_TIME_Relative rtt;
  struct GNUNET_TIME_Relative rtt_average;
  struct Neighbour *n = vl->n;
 
  if (NULL != n && 0 < n->number_of_addresses)
  {
    size_t addresses_size =
      n->number_of_addresses * sizeof (struct TransportGlobalNattedAddress) + n
      ->size_of_global_addresses;
    char *tgnas = GNUNET_malloc (addresses_size);
    struct AddGlobalAddressesContext ctx;
    ctx.off = 0;
    ctx.tgnas = tgnas;
 
    fc = GNUNET_malloc (sizeof (struct TransportFlowControlMessage)
                        + addresses_size);
    fc->header.size = htons (sizeof(struct TransportFlowControlMessage)
                             + addresses_size);
    fc->size_of_addresses = htonl (n->size_of_global_addresses);
    fc->number_of_addresses = htonl (n->number_of_addresses);
    GNUNET_CONTAINER_multipeermap_iterate (n->natted_addresses,
                                           &add_global_addresses,
                                           &ctx);
    GNUNET_memcpy (&fc[1], tgnas, addresses_size);
    GNUNET_free (tgnas);
  }
  else
  {
    fc = GNUNET_malloc (sizeof (struct TransportFlowControlMessage));
    fc->header.size = htons (sizeof(struct TransportFlowControlMessage));
  }
 
  duration = GNUNET_TIME_absolute_get_duration (vl->last_fc_transmission);
  /* OPTIMIZE-FC-BDP: decide sane criteria on when to do this, instead of doing
     it always! */
  /* For example, we should probably ONLY do this if a bit more than
     an RTT has passed, or if the window changed "significantly" since
     then. See vl->last_fc_rtt! NOTE: to do this properly, we also
     need an estimate for the bandwidth-delay-product for the entire
     VL, as that determines "significantly". We have the delay, but
     the bandwidth statistics need to be added for the VL!*/(void) duration;
 
  if (NULL != vl->dv)
    rtt_average = calculate_rtt (vl->dv);
  else
    rtt_average = GNUNET_TIME_UNIT_FOREVER_REL;
  fc->rtt = GNUNET_TIME_relative_hton (rtt_average);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending FC seq %u to %s with new window %llu %lu %u\n",
              (unsigned int) vl->fc_seq_gen,
              GNUNET_i2s (&vl->target),
              (unsigned long long) vl->incoming_fc_window_size,
              (unsigned long) rtt_average.rel_value_us,
              vl->sync_ready);
  monotime = GNUNET_TIME_absolute_get_monotonic (GST_cfg);
  vl->last_fc_transmission = monotime;
  fc->sync_ready = vl->sync_ready;
  fc->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FLOW_CONTROL);
  fc->seq = htonl (vl->fc_seq_gen++);
  fc->inbound_window_size = GNUNET_htonll (vl->incoming_fc_window_size
                                           + vl->incoming_fc_window_size_used
                                           + vl->incoming_fc_window_size_loss);
  fc->outbound_sent = GNUNET_htonll (vl->outbound_fc_window_size_used);
  fc->outbound_window_size = GNUNET_htonll (vl->outbound_fc_window_size);
  fc->sender_time = GNUNET_TIME_absolute_hton (monotime);
  rtt = route_control_message_without_fc (vl, &fc->header, RMO_DV_ALLOWED);
  if (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == rtt.rel_value_us)
  {
    rtt = GNUNET_TIME_UNIT_SECONDS;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "FC retransmission to %s failed, will retry in %s\n",
                GNUNET_i2s (&vl->target),
                GNUNET_STRINGS_relative_time_to_string (rtt, GNUNET_YES));
    vl->last_fc_rtt = GNUNET_TIME_UNIT_ZERO;
  }
  else
  {
    /* OPTIMIZE-FC-BDP: rtt is not ideal, we can do better! */
    vl->last_fc_rtt = rtt;
  }
  if (NULL != vl->fc_retransmit_task)
    GNUNET_SCHEDULER_cancel (vl->fc_retransmit_task);
  if (MAX_FC_RETRANSMIT_COUNT == vl->fc_retransmit_count)
  {
    rtt = GNUNET_TIME_UNIT_MINUTES;
    vl->fc_retransmit_count = 0;
  }
  vl->fc_retransmit_task =
    GNUNET_SCHEDULER_add_delayed (rtt, &task_consider_sending_fc, vl);
  vl->fc_retransmit_count++;
  GNUNET_free (fc);
}
 
 
static void
check_vl_transmission (struct VirtualLink *vl)
{
  struct Neighbour *n = vl->n;
  struct DistanceVector *dv = vl->dv;
  struct GNUNET_TIME_Absolute now;
  struct VirtualLink *vl_next_hop;
  int elig;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "check_vl_transmission to target %s\n",
              GNUNET_i2s (&vl->target));
  /* Check that we have an eligible pending message!
     (cheaper than having #transmit_on_queue() find out!) */
  elig = GNUNET_NO;
  for (struct PendingMessage *pm = vl->pending_msg_head; NULL != pm;
       pm = pm->next_vl)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "check_vl_transmission loop\n");
    if (NULL != pm->qe)
      continue;   /* not eligible, is in a queue! */
    if (pm->bytes_msg + vl->outbound_fc_window_size_used >
        vl->outbound_fc_window_size)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Stalled message %" PRIu64
                  " transmission on VL %s due to flow control: %llu < %llu\n",
                  pm->logging_uuid,
                  GNUNET_i2s (&vl->target),
                  (unsigned long long) vl->outbound_fc_window_size,
                  (unsigned long long) (pm->bytes_msg
                                        + vl->outbound_fc_window_size_used));
      consider_sending_fc (vl);
      return;     /* We have a message, but flow control says "nope" */
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Target window on VL %s not stalled. Scheduling transmission on queue\n",
                GNUNET_i2s (&vl->target));
    /* Notify queues at direct neighbours that we are interested */
    now = GNUNET_TIME_absolute_get ();
    if (NULL != n)
    {
      for (struct Queue *queue = n->queue_head; NULL != queue;
           queue = queue->next_neighbour)
      {
        if ((GNUNET_YES == queue->idle) &&
            (queue->validated_until.abs_value_us > now.abs_value_us))
        {
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Direct neighbour %s not stalled\n",
                      GNUNET_i2s (&n->pid));
          schedule_transmit_on_queue (GNUNET_TIME_UNIT_ZERO,
                                      queue,
                                      GNUNET_SCHEDULER_PRIORITY_DEFAULT);
          elig = GNUNET_YES;
        }
        else
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Neighbour Queue QID: %u (%u) busy or invalid\n",
                      queue->qid,
                      queue->idle);
      }
    }
    /* Notify queues via DV that we are interested */
    if (NULL != dv)
    {
      /* Do DV with lower scheduler priority, which effectively means that
         IF a neighbour exists and is available, we prefer it. */
      for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
           pos = pos->next_dv)
      {
        struct Neighbour *nh_iter = pos->next_hop;
 
 
        if (pos->path_valid_until.abs_value_us <= now.abs_value_us)
          continue;   /* skip this one: path not validated */
        else
        {
          vl_next_hop = lookup_virtual_link (&nh_iter->pid);
          GNUNET_assert (NULL != vl_next_hop);
          if (pm->bytes_msg + vl_next_hop->outbound_fc_window_size_used >
              vl_next_hop->outbound_fc_window_size)
          {
            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                        "Stalled message %" PRIu64
                        " transmission on next hop %s due to flow control: %llu < %llu\n",
                        pm->logging_uuid,
                        GNUNET_i2s (&vl_next_hop->target),
                        (unsigned long
                         long) vl_next_hop->outbound_fc_window_size,
                        (unsigned long long) (pm->bytes_msg
                                              + vl_next_hop->
                                              outbound_fc_window_size_used));
            consider_sending_fc (vl_next_hop);
            continue; /* We have a message, but flow control says "nope" for the first hop of this path */
          }
          for (struct Queue *queue = nh_iter->queue_head; NULL != queue;
               queue = queue->next_neighbour)
            if ((GNUNET_YES == queue->idle) &&
                (queue->validated_until.abs_value_us > now.abs_value_us))
            {
              GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                          "Next hop neighbour %s not stalled\n",
                          GNUNET_i2s (&nh_iter->pid));
              schedule_transmit_on_queue (GNUNET_TIME_UNIT_ZERO,
                                          queue,
                                          GNUNET_SCHEDULER_PRIORITY_BACKGROUND);
              elig = GNUNET_YES;
            }
            else
              GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                          "DV Queue QID: %u (%u) busy or invalid\n",
                          queue->qid,
                          queue->idle);
        }
      }
    }
    if (GNUNET_YES == elig)
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Eligible message %" PRIu64 " of size %u to %s: %llu/%llu\n",
                  pm->logging_uuid,
                  pm->bytes_msg,
                  GNUNET_i2s (&vl->target),
                  (unsigned long long) vl->outbound_fc_window_size,
                  (unsigned long long) (pm->bytes_msg
                                        + vl->outbound_fc_window_size_used));
    break;
  }
}
 
 
static void
handle_client_send (void *cls, const struct OutboundMessage *obm)
{
  struct TransportClient *tc = cls;
  struct PendingMessage *pm;
  const struct GNUNET_MessageHeader *obmm;
  uint32_t bytes_msg;
  struct VirtualLink *vl;
  enum GNUNET_MQ_PriorityPreferences pp;
 
  GNUNET_assert (CT_CORE == tc->type);
  obmm = (const struct GNUNET_MessageHeader *) &obm[1];
  bytes_msg = ntohs (obmm->size);
  pp = (enum GNUNET_MQ_PriorityPreferences) ntohl (obm->priority);
  vl = lookup_virtual_link (&obm->peer);
  if ((NULL == vl) || (GNUNET_NO == vl->confirmed))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Don't have %s as a neighbour (anymore).\n",
                GNUNET_i2s (&obm->peer));
    /* Failure: don't have this peer as a neighbour (anymore).
       Might have gone down asynchronously, so this is NOT
       a protocol violation by CORE. Still count the event,
       as this should be rare. */
    GNUNET_SERVICE_client_continue (tc->client);
    GNUNET_STATISTICS_update (GST_stats,
                              "# messages dropped (neighbour unknown)",
                              1,
                              GNUNET_NO);
    return;
  }
 
  pm = GNUNET_malloc (sizeof(struct PendingMessage) + bytes_msg);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "1 created pm %p storing vl %p\n",
              pm,
              vl);
  pm->logging_uuid = logging_uuid_gen++;
  pm->prefs = pp;
  pm->client = tc;
  pm->vl = vl;
  pm->bytes_msg = bytes_msg;
  memcpy (&pm[1], obmm, bytes_msg);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending message of type %u  with %u bytes as <%" PRIu64
              "> to %s\n",
              ntohs (obmm->type),
              bytes_msg,
              pm->logging_uuid,
              GNUNET_i2s (&obm->peer));
  GNUNET_CONTAINER_MDLL_insert (client,
                                tc->details.core.pending_msg_head,
                                tc->details.core.pending_msg_tail,
                                pm);
  GNUNET_CONTAINER_MDLL_insert (vl,
                                vl->pending_msg_head,
                                vl->pending_msg_tail,
                                pm);
  check_vl_transmission (vl);
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
static void
handle_communicator_backchannel (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
{
  struct Neighbour *n;
  struct VirtualLink *vl;
  struct TransportClient *tc = cls;
  const struct GNUNET_MessageHeader *inbox =
    (const struct GNUNET_MessageHeader *) &cb[1];
  uint16_t isize = ntohs (inbox->size);
  const char *is = ((const char *) &cb[1]) + isize;
  size_t slen = strlen (is) + 1;
  char
    mbuf[slen + isize
         + sizeof(struct
                  TransportBackchannelEncapsulationMessage)] GNUNET_ALIGN;
  struct TransportBackchannelEncapsulationMessage *be =
    (struct TransportBackchannelEncapsulationMessage *) mbuf;
 
  /* 0-termination of 'is' was checked already in
   #check_communicator_backchannel() */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Preparing backchannel transmission to %s:%s of type %u and size %u\n",
              GNUNET_i2s (&cb->pid),
              is,
              ntohs (inbox->type),
              ntohs (inbox->size));
  /* encapsulate and encrypt message */
  be->header.type =
    htons (GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION);
  be->header.size = htons (sizeof(mbuf));
  memcpy (&be[1], inbox, isize);
  memcpy (&mbuf[sizeof(struct TransportBackchannelEncapsulationMessage)
                + isize],
          is,
          strlen (is) + 1);
  // route_control_message_without_fc (&cb->pid, &be->header, RMO_DV_ALLOWED);
  vl = lookup_virtual_link (&cb->pid);
  if ((NULL != vl) && (GNUNET_YES == vl->confirmed))
  {
    route_control_message_without_fc (vl, &be->header, RMO_DV_ALLOWED);
  }
  else
  {
    /* Use route via neighbour */
    n = lookup_neighbour (&cb->pid);
    if (NULL != n)
      route_via_neighbour (
        n,
        &be->header,
        RMO_NONE);
  }
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
static int
check_add_address (void *cls,
                   const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
{
  struct TransportClient *tc = cls;
 
  if (CT_COMMUNICATOR != tc->type)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  GNUNET_MQ_check_zero_termination (aam);
  return GNUNET_OK;
}
 
 
static void
store_pi (void *cls);
 
 
static void
peerstore_store_own_cb (void *cls, int success)
{
  struct AddressListEntry *ale = cls;
 
  ale->sc = NULL;
  if (GNUNET_YES != success)
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store our own address `%s' in peerstore!\n",
                ale->address);
  else
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Successfully stored our own address `%s' in peerstore!\n",
                ale->address);
  /* refresh period is 1/4 of expiration time, that should be plenty
     without being excessive. */
  ale->st =
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_divide (ale->expiration,
                                                               4ULL),
                                  &store_pi,
                                  ale);
}
 
 
static void
shc_cont (void *cls, int success)
{
  struct AddressListEntry *ale = cls;
  struct GNUNET_TIME_Absolute expiration;
 
  expiration = GNUNET_TIME_relative_to_absolute (ale->expiration);
  ale->sc = GNUNET_PEERSTORE_store (peerstore,
                                    "transport",
                                    &GST_my_identity,
                                    GNUNET_PEERSTORE_TRANSPORT_HELLO_KEY,
                                    ale->signed_address,
                                    ale->signed_address_len,
                                    expiration,
                                    GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
                                    &peerstore_store_own_cb,
                                    ale);
  if (NULL == ale->sc)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Failed to store our address `%s' with peerstore\n",
                ale->address);
    ale->st = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
                                            &store_pi,
                                            ale);
  }
}
 
 
static void
store_pi (void *cls)
{
  struct AddressListEntry *ale = cls;
  struct GNUNET_MQ_Envelope *env;
  const struct GNUNET_MessageHeader *msg;
  const char *dash;
  char *address_uri;
  char *prefix = GNUNET_HELLO_address_to_prefix (ale->address);
  unsigned int add_success;
 
  dash = strchr (ale->address, '-');
  GNUNET_assert (NULL != dash);
  dash++;
  GNUNET_asprintf (&address_uri,
                   "%s://%s",
                   prefix,
                   dash);
  GNUNET_free (prefix);
  ale->st = NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Storing our address `%s' in peerstore until %s!\n",
              ale->address,
              GNUNET_STRINGS_absolute_time_to_string (hello_mono_time));
  add_success = GNUNET_HELLO_builder_add_address (GST_my_hello,
                                                  address_uri);
  if (GNUNET_OK != add_success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Storing our address `%s' %s\n",
                address_uri,
                GNUNET_NO == add_success ? "not done" : "failed");
    GNUNET_free (address_uri);
    return;
  }
  else
  {
 
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Storing our address `%s'\n",
                address_uri);
  }
  // FIXME hello_mono_time used here?? What about expiration in ale?
  GNUNET_HELLO_sign_address (ale->address,
                             ale->nt,
                             hello_mono_time,
                             GST_my_private_key,
                             &ale->signed_address,
                             &ale->signed_address_len);
  GNUNET_free (address_uri);
  env = GNUNET_HELLO_builder_to_env (GST_my_hello,
                                     GST_my_private_key,
                                     GNUNET_TIME_UNIT_ZERO);
  msg = GNUNET_MQ_env_get_msg (env);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "store_pi 1\n");
  ale->shc = GNUNET_PEERSTORE_hello_add (peerstore,
                                         msg,
                                         shc_cont,
                                         ale);
  GNUNET_free (env);
}
 
 
static struct AddressListEntry *
create_address_entry (struct TransportClient *tc,
                      struct GNUNET_TIME_Relative expiration,
                      enum GNUNET_NetworkType nt,
                      const char *address,
                      uint32_t aid,
                      size_t slen)
{
  struct AddressListEntry *ale;
  char *address_without_port;
 
  ale = GNUNET_malloc (sizeof(struct AddressListEntry) + slen);
  ale->tc = tc;
  ale->address = (const char *) &ale[1];
  ale->expiration = expiration;
  ale->aid = aid;
  ale->nt = nt;
  memcpy (&ale[1], address, slen);
  address_without_port = get_address_without_port (ale->address);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Is this %s a local address (%s)\n",
              address_without_port,
              ale->address);
  if (0 != strcmp ("127.0.0.1", address_without_port))
    ale->st = GNUNET_SCHEDULER_add_now (&store_pi, ale);
  GNUNET_free (address_without_port);
 
  return ale;
}
 
 
static void
handle_add_address (void *cls,
                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
{
  struct TransportClient *tc = cls;
  struct AddressListEntry *ale;
  size_t slen;
  char *address;
 
  /* 0-termination of &aam[1] was checked in #check_add_address */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Communicator added address `%s'!\n",
              (const char *) &aam[1]);
  slen = ntohs (aam->header.size) - sizeof(*aam);
  address = GNUNET_malloc (slen);
  memcpy (address, &aam[1], slen);
  ale = create_address_entry (tc,
                              GNUNET_TIME_relative_ntoh (aam->expiration),
                              (enum GNUNET_NetworkType) ntohl (aam->nt),
                              address,
                              aam->aid,
                              slen);
  GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
                               tc->details.communicator.addr_tail,
                               ale);
  GNUNET_SERVICE_client_continue (tc->client);
  GNUNET_free (address);
}
 
 
static void
handle_del_address (void *cls,
                    const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
{
  struct TransportClient *tc = cls;
  struct AddressListEntry *alen;
 
  if (CT_COMMUNICATOR != tc->type)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (tc->client);
    return;
  }
  for (struct AddressListEntry *ale = tc->details.communicator.addr_head;
       NULL != ale;
       ale = alen)
  {
    alen = ale->next;
    if (dam->aid != ale->aid)
      continue;
    GNUNET_assert (ale->tc == tc);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Communicator deleted address `%s'!\n",
                ale->address);
    free_address_list_entry (ale);
    GNUNET_SERVICE_client_continue (tc->client);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
              "Communicator removed address we did not even have.\n");
  GNUNET_SERVICE_client_continue (tc->client);
  // GNUNET_SERVICE_client_drop (tc->client);
}
 
 
static void
demultiplex_with_cmc (struct CommunicatorMessageContext *cmc);
 
 
static void
core_env_sent_cb (void *cls)
{
  struct CoreSentContext *ctx = cls;
  struct VirtualLink *vl = ctx->vl;
 
  if (NULL == vl)
  {
    /* lost the link in the meantime, ignore */
    GNUNET_free (ctx);
    return;
  }
  GNUNET_CONTAINER_DLL_remove (vl->csc_head, vl->csc_tail, ctx);
  GNUNET_assert (vl->incoming_fc_window_size_ram >= ctx->size);
  vl->incoming_fc_window_size_ram -= ctx->size;
  vl->incoming_fc_window_size_used += ctx->isize;
  consider_sending_fc (vl);
  GNUNET_free (ctx);
}
 
 
static void
finish_handling_raw_message (struct VirtualLink *vl,
                             const struct GNUNET_MessageHeader *mh,
                             struct CommunicatorMessageContext *cmc,
                             unsigned int free_cmc)
{
  uint16_t size = ntohs (mh->size);
  int have_core;
 
  if (vl->incoming_fc_window_size_ram > UINT_MAX - size)
  {
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages dropped (FC arithmetic overflow)",
                              1,
                              GNUNET_NO);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "CORE messages of type %u with %u bytes dropped (FC arithmetic overflow)\n",
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh->size));
    if (GNUNET_YES == free_cmc)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
    return;
  }
  if (vl->incoming_fc_window_size_ram + size > vl->available_fc_window_size)
  {
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages dropped (FC window overflow)",
                              1,
                              GNUNET_NO);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "CORE messages of type %u with %u bytes dropped (FC window overflow)\n",
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh->size));
    if (GNUNET_YES == free_cmc)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
    return;
  }
 
  /* Forward to all CORE clients */
  have_core = GNUNET_NO;
  for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
  {
    struct GNUNET_MQ_Envelope *env;
    struct InboundMessage *im;
    struct CoreSentContext *ctx;
 
    if (CT_CORE != tc->type)
      continue;
    vl->incoming_fc_window_size_ram += size;
    env = GNUNET_MQ_msg_extra (im, size, GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
    ctx = GNUNET_new (struct CoreSentContext);
    ctx->vl = vl;
    ctx->size = size;
    ctx->isize = (GNUNET_NO == have_core) ? size : 0;
    have_core = GNUNET_YES;
    GNUNET_CONTAINER_DLL_insert (vl->csc_head, vl->csc_tail, ctx);
    GNUNET_MQ_notify_sent (env, &core_env_sent_cb, ctx);
    im->peer = cmc->im.sender;
    memcpy (&im[1], mh, size);
    GNUNET_MQ_send (tc->mq, env);
    vl->core_recv_window--;
  }
  if (GNUNET_NO == have_core)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Dropped message to CORE: no CORE client connected!\n");
    /* Nevertheless, count window as used, as it is from the
       perspective of the other peer! */
    vl->incoming_fc_window_size_used += size;
    /* TODO-M1 */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropped message of type %u with %u bytes to CORE: no CORE client connected!\n",
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh->size));
    if (GNUNET_YES == free_cmc)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Delivered message from %s of type %u to CORE recv window %d\n",
              GNUNET_i2s (&cmc->im.sender),
              ntohs (mh->type),
              vl->core_recv_window);
  if (vl->core_recv_window > 0)
  {
    if (GNUNET_YES == free_cmc)
      finish_cmc_handling_with_continue (cmc, GNUNET_YES);
    return;
  }
  /* Wait with calling #finish_cmc_handling(cmc) until the message
     was processed by CORE MQs (for CORE flow control)! */
  if (GNUNET_YES == free_cmc)
    GNUNET_CONTAINER_DLL_insert (vl->cmc_head, vl->cmc_tail, cmc);
}
 
 
static void
handle_raw_message (void *cls, const struct GNUNET_MessageHeader *mh)
{
  struct CommunicatorMessageContext *cmc = cls;
  // struct CommunicatorMessageContext *cmc_copy =
  // GNUNET_new (struct CommunicatorMessageContext);
  struct GNUNET_MessageHeader *mh_copy;
  struct RingBufferEntry *rbe;
  struct VirtualLink *vl;
  uint16_t size = ntohs (mh->size);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Handling raw message of type %u with %u bytes\n",
              (unsigned int) ntohs (mh->type),
              (unsigned int) ntohs (mh->size));
 
  if ((size > UINT16_MAX - sizeof(struct InboundMessage)) ||
      (size < sizeof(struct GNUNET_MessageHeader)))
  {
    struct GNUNET_SERVICE_Client *client = cmc->tc->client;
 
    GNUNET_break (0);
    finish_cmc_handling (cmc);
    GNUNET_SERVICE_client_drop (client);
    return;
  }
  vl = lookup_virtual_link (&cmc->im.sender);
  if ((NULL == vl) || (GNUNET_NO == vl->confirmed))
  {
    /* FIXME: sender is giving us messages for CORE but we don't have
       the link up yet! I *suspect* this can happen right now (i.e.
       sender has verified us, but we didn't verify sender), but if
       we pass this on, CORE would be confused (link down, messages
       arrive).  We should investigate more if this happens often,
       or in a persistent manner, and possibly do "something" about
       it. Thus logging as error for now. */
 
    mh_copy = GNUNET_malloc (size);
    rbe = GNUNET_new (struct RingBufferEntry);
    rbe->cmc = cmc;
    /*cmc_copy->tc = cmc->tc;
       cmc_copy->im = cmc->im;*/
    GNUNET_memcpy (mh_copy, mh, size);
 
    rbe->mh = mh_copy;
 
    if (GNUNET_YES == is_ring_buffer_full)
    {
      struct RingBufferEntry *rbe_old = ring_buffer[ring_buffer_head];
      GNUNET_free (rbe_old->cmc);
      GNUNET_free (rbe_old->mh);
      GNUNET_free (rbe_old);
    }
    ring_buffer[ring_buffer_head] = rbe;// cmc_copy;
    // cmc_copy->mh = (const struct GNUNET_MessageHeader *) mh_copy;
    cmc->mh = (const struct GNUNET_MessageHeader *) mh_copy;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Storing message for %s and type %u (%u) in ring buffer head %u is full %u\n",
                GNUNET_i2s (&cmc->im.sender),
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh_copy->type),
                ring_buffer_head,
                is_ring_buffer_full);
    if (RING_BUFFER_SIZE - 1 == ring_buffer_head)
    {
      ring_buffer_head = 0;
      is_ring_buffer_full = GNUNET_YES;
    }
    else
      ring_buffer_head++;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "%u items stored in ring buffer\n",
                GNUNET_YES == is_ring_buffer_full ? RING_BUFFER_SIZE :
                ring_buffer_head);
 
    /*GNUNET_break_op (0);
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages dropped (virtual link still down)",
                              1,
                              GNUNET_NO);
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "CORE messages of type %u with %u bytes dropped (virtual link still down)\n",
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh->size));
    finish_cmc_handling (cmc);*/
    finish_cmc_handling_with_continue (cmc, GNUNET_NO);
    cmc->continue_send = GNUNET_YES;
    // GNUNET_free (cmc);
    return;
  }
  finish_handling_raw_message (vl, mh, cmc, GNUNET_YES);
}
 
 
static int
check_fragment_box (void *cls, const struct TransportFragmentBoxMessage *fb)
{
  uint16_t size = ntohs (fb->header.size);
  uint16_t bsize = size - sizeof(*fb);
 
  (void) cls;
  if (0 == bsize)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  if (bsize + ntohs (fb->frag_off) > ntohs (fb->msg_size))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  if (ntohs (fb->frag_off) >= ntohs (fb->msg_size))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_YES;
}
 
 
static void
destroy_ack_cummulator (void *cls)
{
  struct AcknowledgementCummulator *ac = cls;
 
  ac->task = NULL;
  GNUNET_assert (0 == ac->num_acks);
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (ack_cummulators, &ac->target, ac));
  GNUNET_free (ac);
}
 
 
static void
transmit_cummulative_ack_cb (void *cls)
{
  struct Neighbour *n;
  struct VirtualLink *vl;
  struct AcknowledgementCummulator *ac = cls;
  char buf[sizeof(struct TransportReliabilityAckMessage)
           + ac->num_acks
           * sizeof(struct TransportCummulativeAckPayloadP)] GNUNET_ALIGN;
  struct TransportReliabilityAckMessage *ack =
    (struct TransportReliabilityAckMessage *) buf;
  struct TransportCummulativeAckPayloadP *ap;
 
  ac->task = NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending ACK with %u components to %s\n",
              ac->num_acks,
              GNUNET_i2s (&ac->target));
  GNUNET_assert (0 < ac->num_acks);
  ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK);
  ack->header.size =
    htons (sizeof(*ack)
           + ac->num_acks * sizeof(struct TransportCummulativeAckPayloadP));
  ack->ack_counter = htonl (ac->ack_counter += ac->num_acks);
  ap = (struct TransportCummulativeAckPayloadP *) &ack[1];
  for (unsigned int i = 0; i < ac->num_acks; i++)
  {
    ap[i].ack_uuid = ac->ack_uuids[i].ack_uuid;
    ap[i].ack_delay = GNUNET_TIME_relative_hton (
      GNUNET_TIME_absolute_get_duration (ac->ack_uuids[i].receive_time));
  }
  /*route_control_message_without_fc (
    &ac->target,
    &ack->header,
    RMO_DV_ALLOWED);*/
  vl = lookup_virtual_link (&ac->target);
  if ((NULL != vl) && (GNUNET_YES == vl->confirmed))
  {
    route_control_message_without_fc (
      vl,
      &ack->header,
      RMO_DV_ALLOWED);
  }
  else
  {
    /* Use route via neighbour */
    n = lookup_neighbour (&ac->target);
    if (NULL != n)
      route_via_neighbour (
        n,
        &ack->header,
        RMO_NONE);
  }
  ac->num_acks = 0;
  ac->task = GNUNET_SCHEDULER_add_delayed (ACK_CUMMULATOR_TIMEOUT,
                                           &destroy_ack_cummulator,
                                           ac);
}
 
 
static void
cummulative_ack (const struct GNUNET_PeerIdentity *pid,
                 const struct AcknowledgementUUIDP *ack_uuid,
                 struct GNUNET_TIME_Absolute max_delay)
{
  struct AcknowledgementCummulator *ac;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Scheduling ACK %s for transmission to %s\n",
              GNUNET_uuid2s (&ack_uuid->value),
              GNUNET_i2s (pid));
  ac = GNUNET_CONTAINER_multipeermap_get (ack_cummulators, pid);
  if (NULL == ac)
  {
    ac = GNUNET_new (struct AcknowledgementCummulator);
    ac->target = *pid;
    ac->min_transmission_time = max_delay;
    GNUNET_assert (GNUNET_YES ==
                   GNUNET_CONTAINER_multipeermap_put (
                     ack_cummulators,
                     &ac->target,
                     ac,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  }
  else
  {
    if (MAX_CUMMULATIVE_ACKS == ac->num_acks)
    {
      /* must run immediately, ack buffer full! */
      transmit_cummulative_ack_cb (ac);
    }
    GNUNET_SCHEDULER_cancel (ac->task);
    ac->min_transmission_time =
      GNUNET_TIME_absolute_min (ac->min_transmission_time, max_delay);
  }
  GNUNET_assert (ac->num_acks < MAX_CUMMULATIVE_ACKS);
  ac->ack_uuids[ac->num_acks].receive_time = GNUNET_TIME_absolute_get ();
  ac->ack_uuids[ac->num_acks].ack_uuid = *ack_uuid;
  ac->num_acks++;
  ac->task = GNUNET_SCHEDULER_add_at (ac->min_transmission_time,
                                      &transmit_cummulative_ack_cb,
                                      ac);
}
 
 
struct FindByMessageUuidContext
{
  struct MessageUUIDP message_uuid;
 
  struct ReassemblyContext *rc;
};
 
 
static int
find_by_message_uuid (void *cls, uint32_t key, void *value)
{
  struct FindByMessageUuidContext *fc = cls;
  struct ReassemblyContext *rc = value;
 
  (void) key;
  if (0 == GNUNET_memcmp (&fc->message_uuid, &rc->msg_uuid))
  {
    fc->rc = rc;
    return GNUNET_NO;
  }
  return GNUNET_YES;
}
 
 
static void
handle_fragment_box (void *cls, const struct TransportFragmentBoxMessage *fb)
{
  struct CommunicatorMessageContext *cmc = cls;
  struct VirtualLink *vl;
  struct ReassemblyContext *rc;
  const struct GNUNET_MessageHeader *msg;
  uint16_t msize;
  uint16_t fsize;
  uint16_t frag_off;
  char *target;
  struct GNUNET_TIME_Relative cdelay;
  struct FindByMessageUuidContext fc;
 
  vl = lookup_virtual_link (&cmc->im.sender);
  if ((NULL == vl) || (GNUNET_NO == vl->confirmed))
  {
    struct GNUNET_SERVICE_Client *client = cmc->tc->client;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "No virtual link for %s to handle fragment\n",
                GNUNET_i2s (&cmc->im.sender));
    GNUNET_break (0);
    finish_cmc_handling (cmc);
    GNUNET_SERVICE_client_drop (client);
    return;
  }
  if (NULL == vl->reassembly_map)
  {
    vl->reassembly_map = GNUNET_CONTAINER_multihashmap32_create (8);
    vl->reassembly_heap =
      GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
    vl->reassembly_timeout_task =
      GNUNET_SCHEDULER_add_delayed (REASSEMBLY_EXPIRATION,
                                    &reassembly_cleanup_task,
                                    vl);
  }
  msize = ntohs (fb->msg_size);
  fc.message_uuid = fb->msg_uuid;
  fc.rc = NULL;
  (void) GNUNET_CONTAINER_multihashmap32_get_multiple (vl->reassembly_map,
                                                       fb->msg_uuid.uuid,
                                                       &find_by_message_uuid,
                                                       &fc);
  fsize = ntohs (fb->header.size) - sizeof(*fb);
  if (NULL == (rc = fc.rc))
  {
    rc = GNUNET_malloc (sizeof(*rc) + msize    /* reassembly payload buffer */
                        + (msize + 7) / 8 * sizeof(uint8_t) /* bitfield */);
    rc->msg_uuid = fb->msg_uuid;
    rc->virtual_link = vl;
    rc->msg_size = msize;
    rc->reassembly_timeout =
      GNUNET_TIME_relative_to_absolute (REASSEMBLY_EXPIRATION);
    rc->last_frag = GNUNET_TIME_absolute_get ();
    rc->hn = GNUNET_CONTAINER_heap_insert (vl->reassembly_heap,
                                           rc,
                                           rc->reassembly_timeout.abs_value_us);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multihashmap32_put (
                     vl->reassembly_map,
                     rc->msg_uuid.uuid,
                     rc,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
    target = (char *) &rc[1];
    rc->bitfield = (uint8_t *) (target + rc->msg_size);
    if (fsize != rc->msg_size)
      rc->msg_missing = rc->msg_size;
    else
      rc->msg_missing = 0;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received fragment with size %u at offset %u/%u %u bytes missing from %s for NEW message %"
                PRIu64 "\n",
                fsize,
                ntohs (fb->frag_off),
                msize,
                rc->msg_missing,
                GNUNET_i2s (&cmc->im.sender),
                fb->msg_uuid.uuid);
  }
  else
  {
    target = (char *) &rc[1];
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received fragment at offset %u/%u from %s for message %u\n",
                ntohs (fb->frag_off),
                msize,
                GNUNET_i2s (&cmc->im.sender),
                (unsigned int) fb->msg_uuid.uuid);
  }
  if (msize != rc->msg_size)
  {
    GNUNET_break (0);
    finish_cmc_handling (cmc);
    return;
  }
 
  /* reassemble */
  if (0 == fsize)
  {
    GNUNET_break (0);
    finish_cmc_handling (cmc);
    return;
  }
  frag_off = ntohs (fb->frag_off);
  if (frag_off + fsize > msize)
  {
    /* Fragment (plus fragment size) exceeds message size! */
    GNUNET_break_op (0);
    finish_cmc_handling (cmc);
    return;
  }
  memcpy (&target[frag_off], &fb[1], fsize);
  /* update bitfield and msg_missing */
  for (unsigned int i = frag_off; i < frag_off + fsize; i++)
  {
    if (0 == (rc->bitfield[i / 8] & (1 << (i % 8))))
    {
      rc->bitfield[i / 8] |= (1 << (i % 8));
      rc->msg_missing--;
    }
  }
 
  /* Compute cumulative ACK */
  cdelay = GNUNET_TIME_absolute_get_duration (rc->last_frag);
  cdelay = GNUNET_TIME_relative_multiply (cdelay, rc->msg_missing / fsize);
  if (0 == rc->msg_missing)
    cdelay = GNUNET_TIME_UNIT_ZERO;
  cummulative_ack (&cmc->im.sender,
                   &fb->ack_uuid,
                   GNUNET_TIME_relative_to_absolute (cdelay));
  rc->last_frag = GNUNET_TIME_absolute_get ();
  /* is reassembly complete? */
  if (0 != rc->msg_missing)
  {
    finish_cmc_handling (cmc);
    return;
  }
  /* reassembly is complete, verify result */
  msg = (const struct GNUNET_MessageHeader *) &rc[1];
  if (ntohs (msg->size) != rc->msg_size)
  {
    GNUNET_break (0);
    free_reassembly_context (rc);
    finish_cmc_handling (cmc);
    return;
  }
  /* successful reassembly */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Fragment reassembly complete for message %u\n",
              (unsigned int) fb->msg_uuid.uuid);
  /* FIXME: check that the resulting msg is NOT a
     DV Box or Reliability Box, as that is NOT allowed! */
  cmc->mh = msg;
  demultiplex_with_cmc (cmc);
  /* FIXME-OPTIMIZE: really free here? Might be bad if fragments are still
     en-route and we forget that we finished this reassembly immediately!
     -> keep around until timeout?
     -> shorten timeout based on ACK? */
  free_reassembly_context (rc);
}
 
 
static int
check_reliability_box (void *cls,
                       const struct TransportReliabilityBoxMessage *rb)
{
  (void) cls;
  const struct GNUNET_MessageHeader *box =  (const struct
                                               GNUNET_MessageHeader *) &rb[1];
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "check_send_msg with size %u: inner msg type %u and size %u (%lu %lu)\n",
              ntohs (rb->header.size),
              ntohs (box->type),
              ntohs (box->size),
              sizeof (struct TransportReliabilityBoxMessage),
              sizeof (struct GNUNET_MessageHeader));
  GNUNET_MQ_check_boxed_message (rb);
  return GNUNET_YES;
}
 
 
static void
handle_reliability_box (void *cls,
                        const struct TransportReliabilityBoxMessage *rb)
{
  struct CommunicatorMessageContext *cmc = cls;
  const struct GNUNET_MessageHeader *inbox =
    (const struct GNUNET_MessageHeader *) &rb[1];
  struct GNUNET_TIME_Relative rtt;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received reliability box from %s with UUID %s of type %u\n",
              GNUNET_i2s (&cmc->im.sender),
              GNUNET_uuid2s (&rb->ack_uuid.value),
              (unsigned int) ntohs (inbox->type));
  rtt = GNUNET_TIME_UNIT_SECONDS; /* FIXME: should base this on "RTT", but we
                                     do not really have an RTT for the
                                   * incoming* queue (should we have
                                     the sender add it to the rb message?) */
  cummulative_ack (
    &cmc->im.sender,
    &rb->ack_uuid,
    (0 == ntohl (rb->ack_countdown))
    ? GNUNET_TIME_UNIT_ZERO_ABS
    : GNUNET_TIME_relative_to_absolute (
      GNUNET_TIME_relative_divide (rtt, 8 /* FIXME: magic constant */)));
  /* continue with inner message */
  /* FIXME: check that inbox is NOT a DV Box, fragment or another
     reliability box (not allowed!) */
  cmc->mh = inbox;
  demultiplex_with_cmc (cmc);
}
 
 
static void
update_pd_age (struct PerformanceData *pd, unsigned int age)
{
  unsigned int sage;
 
  if (age == pd->last_age)
    return; /* nothing to do */
  sage = GNUNET_MAX (pd->last_age, age - 2 * GOODPUT_AGING_SLOTS);
  for (unsigned int i = sage; i <= age - GOODPUT_AGING_SLOTS; i++)
  {
    struct TransmissionHistoryEntry *the = &pd->the[i % GOODPUT_AGING_SLOTS];
 
    the->bytes_sent = 0;
    the->bytes_received = 0;
  }
  pd->last_age = age;
}
 
 
static void
update_performance_data (struct PerformanceData *pd,
                         struct GNUNET_TIME_Relative rtt,
                         uint16_t bytes_transmitted_ok)
{
  uint64_t nval = rtt.rel_value_us;
  uint64_t oval = pd->aged_rtt.rel_value_us;
  unsigned int age = get_age ();
  struct TransmissionHistoryEntry *the = &pd->the[age % GOODPUT_AGING_SLOTS];
 
  if (oval == GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us)
    pd->aged_rtt = rtt;
  else
    pd->aged_rtt.rel_value_us = (nval + 7 * oval) / 8;
  update_pd_age (pd, age);
  the->bytes_received += bytes_transmitted_ok;
}
 
 
static void
update_queue_performance (struct Queue *q,
                          struct GNUNET_TIME_Relative rtt,
                          uint16_t bytes_transmitted_ok)
{
  update_performance_data (&q->pd, rtt, bytes_transmitted_ok);
}
 
 
static void
update_dvh_performance (struct DistanceVectorHop *dvh,
                        struct GNUNET_TIME_Relative rtt,
                        uint16_t bytes_transmitted_ok)
{
  update_performance_data (&dvh->pd, rtt, bytes_transmitted_ok);
}
 
 
static void
completed_pending_message (struct PendingMessage *pm)
{
  struct PendingMessage *pos;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Complete transmission of message %" PRIu64 " %u\n",
              pm->logging_uuid,
              pm->pmt);
  switch (pm->pmt)
  {
  case PMT_CORE:
  case PMT_RELIABILITY_BOX:
    /* Full message sent, we are done */
    client_send_response (pm);
    return;
 
  case PMT_FRAGMENT_BOX:
    /* Fragment sent over reliable channel */
    pos = pm->frag_parent;
    GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, pm);
    free_pending_message (pm);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "pos frag_off %lu pos bytes_msg %lu pmt %u parent %u\n",
                (unsigned long) pos->frag_off,
                (unsigned long) pos->bytes_msg,
                pos->pmt,
                NULL == pos->frag_parent ? 1 : 0);
    /* check if subtree is done */
    while ((NULL == pos->head_frag) && (pos->frag_off == (pos->bytes_msg
                                                          - sizeof(struct
                                                                   TransportFragmentBoxMessage)))
           &&
           (NULL != pos->frag_parent))
    {
      pm = pos;
      pos = pm->frag_parent;
      if ((NULL == pos) && (PMT_DV_BOX == pm->pmt))
      {
        client_send_response (pm);
        return;
      }
      else if (PMT_DV_BOX == pm->pmt)
      {
        client_send_response (pos);