gnunet-service-tng.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_hello_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
 
 
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_EcdhePublicKey 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_EcdhePublicKey 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 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;
};
 
 
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;
};
 
 
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;
 
  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;
 
  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;
};
 
 
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;
};
 
 
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_EcdhePublicKey ephemeral_key;
 
  struct GNUNET_CRYPTO_EcdhePrivateKey private_key;
};
 
 
struct QueueEntry
{
  struct QueueEntry *next;
 
  struct QueueEntry *prev;
 
  struct Queue *queue;
 
  struct PendingMessage *pm;
 
  uint64_t mid;
};
 
 
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;
 
  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;
};
 
 
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 IncomingRequest
{
  struct IncomingRequest *next;
 
  struct IncomingRequest *prev;
 
  struct GNUNET_PEERSTORE_WatchContext *wc;
 
  struct GNUNET_PeerIdentity pid;
};
 
 
struct PeerRequest
{
  struct GNUNET_PeerIdentity pid;
 
  struct TransportClient *tc;
 
  struct GNUNET_PEERSTORE_WatchContext *wc;
 
  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;
 
  unsigned long long logging_uuid;
 
  enum PendingMessageType pmt;
 
  enum GNUNET_MQ_PriorityPreferences prefs;
 
  struct DistanceVectorHop *used_dvh;
 
  uint16_t bytes_msg;
 
  uint16_t frag_off;
 
  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;
 
  const char *address;
 
  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;
 
      enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
    } 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;
 
  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_EcdhePublicKey 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;
 
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_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;
 
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 struct GNUNET_TIME_Absolute hello_mono_time;
 
static int in_shutdown;
 
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--;
  GNUNET_PEERSTORE_watch_cancel (ir->wc);
  ir->wc = 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 %llu\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_CONTAINER_DLL_remove (frag->qe->queue->queue_head,
                                   frag->qe->queue->queue_tail,
                                   frag->qe);
      frag->qe->queue->queue_length--;
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Removing QueueEntry MID %lu from queue\n",
                  frag->qe->mid);
      GNUNET_free (frag->qe);
    }
    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 %llu\n",
                pm->logging_uuid);
    GNUNET_CONTAINER_MDLL_remove (vl,
                                  vl->pending_msg_head,
                                  vl->pending_msg_tail,
                                  pm);
  }
  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;
    GNUNET_CONTAINER_DLL_remove (pm->qe->queue->queue_head,
                                 pm->qe->queue->queue_tail,
                                 pm->qe);
    pm->qe->queue->queue_length--;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Removing QueueEntry MID %lu from queue\n",
                pm->qe->mid);
    GNUNET_free (pm->qe);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "QueueEntry MID freed\n");
  }
  if (NULL != pm->bpm)
  {
    free_fragment_tree (pm->bpm);
    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)
{
  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);
  }
}
 
 
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 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");
  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_iterate_cancel (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)
{
  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 throttled due to communicator queue limit\n");
    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 throttled due to communicator queue length limit\n");
    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 throttled due to communicator message queue qid %u has capacity %lu.\n",
                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->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_log (GNUNET_ERROR_TYPE_DEBUG,
              "Cleaning up queue with length %u\n",
              queue->queue_length);
  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;
  }
  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;
 
  GNUNET_PEERSTORE_watch_cancel (pr->wc);
  pr->wc = 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;
 
      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;
 
  (void) value;
  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
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 (neighbours,
                                         &notify_client_connect_info,
                                         tc);
  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 <%llu> 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 continue_client)
{
  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 %lu\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);
  }
 
  if (GNUNET_YES == continue_client)
  {
    GNUNET_SERVICE_client_continue (cmc->tc->client);
  }
  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 %u\n",
              vl->core_recv_window);
  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);
    finish_cmc_handling (cmc);
  }
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
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);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Communicator with prefix `%s' connected\n",
              tc->details.communicator.address_prefix);
  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
update_ephemeral (struct DistanceVector *dv)
{
  struct EphemeralConfirmationPS ec;
 
  if (0 !=
      GNUNET_TIME_absolute_get_remaining (dv->ephemeral_validity).rel_value_us)
    return;
  dv->monotime = GNUNET_TIME_absolute_get_monotonic (GST_cfg);
  dv->ephemeral_validity =
    GNUNET_TIME_absolute_add (dv->monotime, EPHEMERAL_VALIDITY);
  GNUNET_CRYPTO_ecdhe_key_create (&dv->private_key);
  GNUNET_CRYPTO_ecdhe_key_get_public (&dv->private_key, &dv->ephemeral_key);
  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
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;
 
  GNUNET_log (
    GNUNET_ERROR_TYPE_DEBUG,
    "Queueing %u bytes of payload for transmission <%llu> 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 = queue->qid;
  smt->mid = 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->mid = 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 <%llu> 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_CONTAINER_DLL_insert (queue->queue_head, queue->queue_tail, qe);
    GNUNET_assert (CT_COMMUNICATOR == queue->tc->type);
    queue->queue_length++;
    queue->tc->details.communicator.total_queue_length++;
    //FIXME Probably this if statement here is completely wrong in this method,
    // and only fixed a symptom, but not an actual bug.
    if (0 == queue->q_capacity)
    {
      GNUNET_free (env);
      return;
    }
    if (GNUNET_NO == queue->unlimited_length)
      queue->q_capacity--;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Queue %s with qid %u has capacity %lu\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;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending message MID %lu of type %u (%u) and size %lu with MQ %p\n",
                smt->mid,
                ntohs (((const struct GNUNET_MessageHeader *) payload)->type),
                ntohs (smt->header.size),
                payload_size,
                queue->tc->mq);
    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_HashCode *km,
                            const struct GNUNET_ShortHashCode *iv,
                            struct DVKeyState *key)
{
  /* must match #dh_key_derive_eph_pub */
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_kdf (&key->material,
                                    sizeof(key->material),
                                    "transport-backchannel-key",
                                    strlen ("transport-backchannel-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_h2s (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 enum GNUNET_GenericReturnValue
dh_key_derive_eph_pid (
  const struct GNUNET_CRYPTO_EcdhePrivateKey *priv_ephemeral,
  const struct GNUNET_PeerIdentity *target,
  const struct GNUNET_ShortHashCode *iv,
  struct DVKeyState *key)
{
  struct GNUNET_HashCode km;
 
  if (GNUNET_YES != GNUNET_CRYPTO_ecdh_eddsa (priv_ephemeral,
                                              &target->public_key,
                                              &km))
    return GNUNET_SYSERR;
  // FIXME: Possibly also add return values here. We are processing
  // Input from other peers...
  dv_setup_key_state_from_km (&km, iv, key);
  return GNUNET_OK;
}
 
 
static enum GNUNET_GenericReturnValue
dh_key_derive_eph_pub (const struct GNUNET_CRYPTO_EcdhePublicKey *pub_ephemeral,
                       const struct GNUNET_ShortHashCode *iv,
                       struct DVKeyState *key)
{
  struct GNUNET_HashCode km;
 
  if (GNUNET_YES != GNUNET_CRYPTO_eddsa_ecdh (GST_my_private_key,
                                              pub_ephemeral,
                                              &km))
    return GNUNET_SYSERR;
  dv_setup_key_state_from_km (&km, iv, key);
  return GNUNET_OK;
}
 
 
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;
 
  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);
  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.
  GNUNET_assert (GNUNET_OK ==
                 dh_key_derive_eph_pid (&dv->private_key,
                                        &dv->target,
                                        &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 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;
 
  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;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending FC seq %u to %s with new window %llu\n",
              (unsigned int) vl->fc_seq_gen,
              GNUNET_i2s (&vl->target),
              (unsigned long long) vl->incoming_fc_window_size);
  monotime = GNUNET_TIME_absolute_get_monotonic (GST_cfg);
  vl->last_fc_transmission = monotime;
  fc.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FLOW_CONTROL);
  fc.header.size = htons (sizeof(fc));
  fc.seq = htonl (vl->fc_seq_gen++);
  fc.inbound_window_size = GNUNET_htonll (vl->incoming_fc_window_size);
  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++;
}
 
 
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 %llu 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 = 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->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 %llu 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->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->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 %llu 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 %u bytes as <%llu> to %s\n",
              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
store_pi (void *cls)
{
  struct AddressListEntry *ale = cls;
  void *addr;
  size_t addr_len;
  struct GNUNET_TIME_Absolute expiration;
 
  ale->st = NULL;
  expiration = GNUNET_TIME_relative_to_absolute (ale->expiration);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Storing our address `%s' in peerstore until %s!\n",
              ale->address,
              GNUNET_STRINGS_absolute_time_to_string (expiration));
  GNUNET_HELLO_sign_address (ale->address,
                             ale->nt,
                             hello_mono_time,
                             GST_my_private_key,
                             &addr,
                             &addr_len);
  ale->sc = GNUNET_PEERSTORE_store (peerstore,
                                    "transport",
                                    &GST_my_identity,
                                    GNUNET_PEERSTORE_TRANSPORT_HELLO_KEY,
                                    addr,
                                    addr_len,
                                    expiration,
                                    GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
                                    &peerstore_store_own_cb,
                                    ale);
  GNUNET_free (addr);
  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
handle_add_address (void *cls,
                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
{
  struct TransportClient *tc = cls;
  struct AddressListEntry *ale;
  size_t slen;
 
  /* 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);
  ale = GNUNET_malloc (sizeof(struct AddressListEntry) + slen);
  ale->tc = tc;
  ale->address = (const char *) &ale[1];
  ale->expiration = GNUNET_TIME_relative_ntoh (aam->expiration);
  ale->aid = aam->aid;
  ale->nt = (enum GNUNET_NetworkType) ntohl (aam->nt);
  memcpy (&ale[1], &aam[1], slen);
  GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
                               tc->details.communicator.addr_tail,
                               ale);
  ale->st = GNUNET_SCHEDULER_add_now (&store_pi, ale);
  GNUNET_SERVICE_client_continue (tc->client);
}
 
 
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 continue_client)
{
  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));
    finish_cmc_handling_with_continue (cmc, continue_client);
    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));
    finish_cmc_handling_with_continue (cmc, continue_client);
    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));
    finish_cmc_handling_with_continue (cmc, continue_client);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Delivered message from %s of type %u to CORE recv window %u\n",
              GNUNET_i2s (&cmc->im.sender),
              ntohs (mh->type),
              vl->core_recv_window);
  if (vl->core_recv_window > 0)
  {
    finish_cmc_handling_with_continue (cmc, continue_client);
    return;
  }
  /* Wait with calling #finish_cmc_handling(cmc) until the message
     was processed by CORE MQs (for CORE flow control)! */
  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;
 
    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\n",
                GNUNET_i2s (&cmc->im.sender),
                (unsigned int) ntohs (mh->type),
                (unsigned int) ntohs (mh_copy->type));
    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",
                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);*/
    GNUNET_SERVICE_client_continue (cmc->tc->client);
    // 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 %u\n",
                fsize,
                ntohs (fb->frag_off),
                msize,
                rc->msg_missing,
                GNUNET_i2s (&cmc->im.sender),
                (unsigned int) 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 *inbox =  (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 (inbox->type),
              ntohs (inbox->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 %llu %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);
    /* check if subtree is done */
    while ((NULL == pos->head_frag) && (pos->frag_off == pos->bytes_msg) &&
           (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);
        return;
      }
      GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, pm);
      free_pending_message (pm);
    }
 
    /* Was this the last applicable fragment? */
    if ((NULL == pos->head_frag) && (NULL == pos->frag_parent) &&
        (pos->frag_off == pos->bytes_msg))
      client_send_response (pos);
    return;
 
  case PMT_DV_BOX:
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Completed transmission of message %llu (DV Box)\n",
                pm->logging_uuid);
    if (NULL != pm->frag_parent)
    {
      if (NULL != pm->bpm)
      {
        GNUNET_free (pm->bpm);
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Freed bpm\n");
      }
      pos = pm->frag_parent;
      free_pending_message (pm);
      pos->bpm = NULL;
      client_send_response (pos);
    }
    else
      client_send_response (pm);
    return;
  }
}
 
 
static void
handle_acknowledged (struct PendingAcknowledgement *pa,
                     struct GNUNET_TIME_Relative ack_delay)
{
  struct GNUNET_TIME_Relative delay;
 
  delay = GNUNET_TIME_absolute_get_duration (pa->transmission_time);
  if (delay.rel_value_us > ack_delay.rel_value_us)
    delay = GNUNET_TIME_UNIT_ZERO;
  else
    delay = GNUNET_TIME_relative_subtract (delay, ack_delay);
  if (NULL != pa->queue)
    update_queue_performance (pa->queue, delay, pa->message_size);
  if (NULL != pa->dvh)
    update_dvh_performance (pa->dvh, delay, pa->message_size);
  if (NULL != pa->pm)
    completed_pending_message (pa->pm);
  free_pending_acknowledgement (pa);
}
 
 
static int
check_reliability_ack (void *cls,
                       const struct TransportReliabilityAckMessage *ra)
{
  unsigned int n_acks;
 
  (void) cls;
  n_acks = (ntohs (ra->header.size) - sizeof(*ra))
           / sizeof(struct TransportCummulativeAckPayloadP);
  if (0 == n_acks)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  if ((ntohs (ra->header.size) - sizeof(*ra)) !=
      n_acks * sizeof(struct TransportCummulativeAckPayloadP))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_reliability_ack (void *cls,
                        const struct TransportReliabilityAckMessage *ra)
{
  struct CommunicatorMessageContext *cmc = cls;
  const struct TransportCummulativeAckPayloadP *ack;
  unsigned int n_acks;
  uint32_t ack_counter;
 
  n_acks = (ntohs (ra->header.size) - sizeof(*ra))
           / sizeof(struct TransportCummulativeAckPayloadP);
  ack = (const struct TransportCummulativeAckPayloadP *) &ra[1];
  for (unsigned int i = 0; i < n_acks; i++)
  {
    struct PendingAcknowledgement *pa =
      GNUNET_CONTAINER_multiuuidmap_get (pending_acks, &ack[i].ack_uuid.value);
    if (NULL == pa)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Received ACK from %s with UUID %s which is unknown to us!\n",
                  GNUNET_i2s (&cmc->im.sender),
                  GNUNET_uuid2s (&ack[i].ack_uuid.value));
      GNUNET_STATISTICS_update (
        GST_stats,
        "# FRAGMENT_ACKS dropped, no matching pending message",
        1,
        GNUNET_NO);
      continue;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received ACK from %s with UUID %s\n",
                GNUNET_i2s (&cmc->im.sender),
                GNUNET_uuid2s (&ack[i].ack_uuid.value));
    handle_acknowledged (pa, GNUNET_TIME_relative_ntoh (ack[i].ack_delay));
  }
 
  ack_counter = htonl (ra->ack_counter);
  (void) ack_counter;  /* silence compiler warning for now */
  // FIXME-OPTIMIZE: track ACK losses based on ack_counter somewhere!
  // (DV and/or Neighbour?)
  finish_cmc_handling (cmc);
}
 
 
static int
check_backchannel_encapsulation (
  void *cls,
  const struct TransportBackchannelEncapsulationMessage *be)
{
  uint16_t size = ntohs (be->header.size) - sizeof(*be);
  const struct GNUNET_MessageHeader *inbox =
    (const struct GNUNET_MessageHeader *) &be[1];
  const char *is;
  uint16_t isize;
 
  (void) cls;
  if (ntohs (inbox->size) >= size)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  isize = ntohs (inbox->size);
  is = ((const char *) inbox) + isize;
  size -= isize;
  if ('\0' != is[size - 1])
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_YES;
}
 
 
static void
handle_backchannel_encapsulation (
  void *cls,
  const struct TransportBackchannelEncapsulationMessage *be)
{
  struct CommunicatorMessageContext *cmc = cls;
  struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *cbi;
  struct GNUNET_MQ_Envelope *env;
  struct TransportClient *tc;
  const struct GNUNET_MessageHeader *inbox =
    (const struct GNUNET_MessageHeader *) &be[1];
  uint16_t isize = ntohs (inbox->size);
  const char *target_communicator = ((const char *) inbox) + isize;
  char *sender;
  char *self;
 
  GNUNET_asprintf (&sender,
                   "%s",
                   GNUNET_i2s (&cmc->im.sender));
  GNUNET_asprintf (&self,
                   "%s",
                   GNUNET_i2s (&GST_my_identity));
 
  /* Find client providing this communicator */
  for (tc = clients_head; NULL != tc; tc = tc->next)
    if ((CT_COMMUNICATOR == tc->type) &&
        (0 ==
         strcmp (tc->details.communicator.address_prefix, target_communicator)))
      break;
  if (NULL == tc)
  {
    char *stastr;
 
    GNUNET_asprintf (
      &stastr,
      "# Backchannel message dropped: target communicator `%s' unknown",
      target_communicator);
    GNUNET_STATISTICS_update (GST_stats, stastr, 1, GNUNET_NO);
    GNUNET_free (stastr);
    finish_cmc_handling (cmc);
    return;
  }
  /* Finally, deliver backchannel message to communicator */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Delivering backchannel message from %s to %s of type %u to %s\n",
              sender,
              self,
              ntohs (inbox->type),
              target_communicator);
  env = GNUNET_MQ_msg_extra (
    cbi,
    isize,
    GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL_INCOMING);
  cbi->pid = cmc->im.sender;
  memcpy (&cbi[1], inbox, isize);
  GNUNET_MQ_send (tc->mq, env);
  finish_cmc_handling (cmc);
}
 
 
static void
path_cleanup_cb (void *cls)
{
  struct DistanceVector *dv = cls;
  struct DistanceVectorHop *pos;
 
  dv->timeout_task = NULL;
  while (NULL != (pos = dv->dv_head))
  {
    GNUNET_assert (dv == pos->dv);
    if (GNUNET_TIME_absolute_get_remaining (pos->timeout).rel_value_us > 0)
      break;
    free_distance_vector_hop (pos);
  }
  if (NULL == pos)
  {
    free_dv_route (dv);
    return;
  }
  dv->timeout_task =
    GNUNET_SCHEDULER_add_at (pos->timeout, &path_cleanup_cb, dv);
}
 
 
static void send_msg_from_cache (struct VirtualLink *vl)
{
 
  const struct GNUNET_PeerIdentity target = vl->target;
 
 
  if ((GNUNET_YES == is_ring_buffer_full) || (0 < ring_buffer_head))
  {
    struct RingBufferEntry *ring_buffer_copy[RING_BUFFER_SIZE];
    unsigned int tail = GNUNET_YES == is_ring_buffer_full ? ring_buffer_head :
                        0;
    unsigned int head = GNUNET_YES == is_ring_buffer_full ? RING_BUFFER_SIZE :
                        ring_buffer_head;
    struct GNUNET_TRANSPORT_IncomingMessage im;
    struct CommunicatorMessageContext *cmc;
    struct RingBufferEntry *rbe;
    struct GNUNET_MessageHeader *mh;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending from ring buffer, which has %u items\n",
                ring_buffer_head);
 
    ring_buffer_head = 0;
    for (unsigned int i = 0; i < head; i++)
    {
      rbe = ring_buffer[(i + tail) % RING_BUFFER_SIZE];
      cmc = rbe->cmc;
      mh = rbe->mh;
 
      im = cmc->im;
      // mh = cmc->mh;
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Sending to ring buffer target %s using vl target %s\n",
                  GNUNET_i2s (&im.sender),
                  GNUNET_i2s2 (&target));
      if (0 == GNUNET_memcmp (&target, &im.sender))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Finish handling message of type %u and size %u\n",
                    (unsigned int) ntohs (mh->type),
                    (unsigned int) ntohs (mh->size));
        finish_handling_raw_message (vl, mh, cmc, GNUNET_NO);
        GNUNET_free (mh);
      }
      else
      {
        ring_buffer_copy[i] = rbe;
        ring_buffer_head++;
      }
    }
 
    if ((GNUNET_YES == is_ring_buffer_full) && (RING_BUFFER_SIZE - 1 >
                                                ring_buffer_head))
    {
      is_ring_buffer_full = GNUNET_NO;
    }
 
    for (unsigned int i = 0; i < ring_buffer_head; i++)
    {
      ring_buffer[i] = ring_buffer_copy[i];
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "ring_buffer_copy[i]->mh->type for i %u %u\n",
                  i,
                  ring_buffer_copy[i]->mh->type);
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "ring_buffer[i]->mh->type for i %u %u\n",
                  i,
                  ring_buffer[i]->mh->type);
    }
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "%u items still in ring buffer\n",
                ring_buffer_head);
  }
 
  if ((GNUNET_YES == is_ring_buffer_full) || (0 < ring_buffer_dv_head))
  {
    struct PendingMessage *ring_buffer_dv_copy[RING_BUFFER_SIZE];
    struct PendingMessage *pm;
    unsigned int tail = GNUNET_YES == is_ring_buffer_dv_full ?
                        ring_buffer_dv_head :
                        0;
    unsigned int head = GNUNET_YES == is_ring_buffer_dv_full ?
                        RING_BUFFER_SIZE :
                        ring_buffer_dv_head;
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending from ring buffer dv, which has %u items\n",
                ring_buffer_dv_head);
 
    ring_buffer_dv_head = 0;
    for (unsigned int i = 0; i < head; i++)
    {
      pm = ring_buffer_dv[(i + tail) % RING_BUFFER_SIZE];
 
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Sending to ring buffer target %s using vl target %s\n",
                  GNUNET_i2s (&pm->target),
                  GNUNET_i2s2 (&target));
      if (0 == GNUNET_memcmp (&target, &pm->target))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Adding PendingMessage to vl, checking transmission.\n");
        pm->vl = vl;
        GNUNET_CONTAINER_MDLL_insert (vl,
                                      vl->pending_msg_head,
                                      vl->pending_msg_tail,
                                      pm);
 
        check_vl_transmission (vl);
      }
      else
      {
        ring_buffer_dv_copy[i] = pm;
        ring_buffer_dv_head++;
      }
    }
 
    if (is_ring_buffer_dv_full && (RING_BUFFER_SIZE - 1 > ring_buffer_dv_head))
    {
      is_ring_buffer_dv_full = GNUNET_NO;
    }
 
    for (unsigned int i = 0; i < ring_buffer_dv_head; i++)
      ring_buffer_dv[i] = ring_buffer_dv_copy[i];
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "%u items still in ring buffer dv.\n",
                ring_buffer_dv_head);
 
  }
}
 
 
static void
activate_core_visible_dv_path (struct DistanceVectorHop *hop)
{
  struct DistanceVector *dv = hop->dv;
  struct VirtualLink *vl;
 
  vl = lookup_virtual_link (&dv->target);
  if (NULL == vl)
  {
 
    vl = GNUNET_new (struct VirtualLink);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Creating new virtual link %p to %s using DV!\n",
                vl,
                GNUNET_i2s (&dv->target));
    vl->confirmed = GNUNET_YES;
    vl->message_uuid_ctr =
      GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
    vl->target = dv->target;
    vl->core_recv_window = RECV_WINDOW_SIZE;
    vl->available_fc_window_size = DEFAULT_WINDOW_SIZE;
    vl->incoming_fc_window_size = DEFAULT_WINDOW_SIZE;
    GNUNET_break (GNUNET_YES ==
                  GNUNET_CONTAINER_multipeermap_put (
                    links,
                    &vl->target,
                    vl,
                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    vl->dv = dv;
    dv->vl = vl;
    vl->visibility_task =
      GNUNET_SCHEDULER_add_at (hop->path_valid_until, &check_link_down, vl);
    consider_sending_fc (vl);
    /* We lacked a confirmed connection to the target
       before, so tell CORE about it (finally!) */
    cores_send_connect_info (&dv->target);
    send_msg_from_cache (vl);
  }
  else
  {
    /* Link was already up, remember dv is also now available and we are done */
    vl->dv = dv;
    dv->vl = vl;
    if (GNUNET_NO == vl->confirmed)
    {
      vl->confirmed = GNUNET_YES;
      vl->visibility_task =
        GNUNET_SCHEDULER_add_at (hop->path_valid_until, &check_link_down, vl);
      consider_sending_fc (vl);
      /* We lacked a confirmed connection to the target
         before, so tell CORE about it (finally!) */
      cores_send_connect_info (&dv->target);
      send_msg_from_cache (vl);
    }
    else
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Virtual link to %s could now also use DV!\n",
                  GNUNET_i2s (&dv->target));
  }
}
 
 
static int
learn_dv_path (const struct GNUNET_PeerIdentity *path,
               unsigned int path_len,
               struct GNUNET_TIME_Relative network_latency,
               struct GNUNET_TIME_Absolute path_valid_until)
{
  struct DistanceVectorHop *hop;
  struct DistanceVector *dv;
  struct Neighbour *next_hop;
  unsigned int shorter_distance;
 
  if (path_len < 3)
  {
    /* what a boring path! not allowed! */
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  GNUNET_assert (0 == GNUNET_memcmp (&GST_my_identity, &path[0]));
  next_hop = lookup_neighbour (&path[1]);
  if (NULL == next_hop)
  {
    /* next hop must be a neighbour, otherwise this whole thing is useless! */
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  for (unsigned int i = 2; i < path_len; i++)
    if (NULL != lookup_neighbour (&path[i]))
    {
      /* Useless path: we have a direct connection to some hop
         in the middle of the path, so this one is not even
         terribly useful for redundancy */
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Path of %u hops useless: directly link to hop %u (%s)\n",
                  path_len,
                  i,
                  GNUNET_i2s (&path[i]));
      GNUNET_STATISTICS_update (GST_stats,
                                "# Useless DV path ignored: hop is neighbour",
                                1,
                                GNUNET_NO);
      return GNUNET_SYSERR;
    }
  dv = GNUNET_CONTAINER_multipeermap_get (dv_routes, &path[path_len - 1]);
  if (NULL == dv)
  {
    dv = GNUNET_new (struct DistanceVector);
    dv->target = path[path_len - 1];
    dv->timeout_task = GNUNET_SCHEDULER_add_delayed (DV_PATH_VALIDITY_TIMEOUT,
                                                     &path_cleanup_cb,
                                                     dv);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multipeermap_put (
                     dv_routes,
                     &dv->target,
                     dv,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  }
  /* Check if we have this path already! */
  shorter_distance = 0;
  for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
       pos = pos->next_dv)
  {
    if (pos->distance < path_len - 3)
      shorter_distance++;
    /* Note that the distances in 'pos' excludes us (path[0]),
       the next_hop (path[1]) and the target so we need to subtract three
       and check next_hop explicitly */
    if ((pos->distance == path_len - 3) && (pos->next_hop == next_hop))
    {
      int match = GNUNET_YES;
 
      for (unsigned int i = 0; i < pos->distance; i++)
      {
        if (0 != GNUNET_memcmp (&pos->path[i], &path[i + 2]))
        {
          match = GNUNET_NO;
          break;
        }
      }
      if (GNUNET_YES == match)
      {
        struct GNUNET_TIME_Relative last_timeout;
 
        /* Re-discovered known path, update timeout */
        GNUNET_STATISTICS_update (GST_stats,
                                  "# Known DV path refreshed",
                                  1,
                                  GNUNET_NO);
        last_timeout = GNUNET_TIME_absolute_get_remaining (pos->timeout);
        pos->timeout =
          GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
        pos->path_valid_until =
          GNUNET_TIME_absolute_max (pos->path_valid_until, path_valid_until);
        GNUNET_CONTAINER_MDLL_remove (dv, dv->dv_head, dv->dv_tail, pos);
        GNUNET_CONTAINER_MDLL_insert (dv, dv->dv_head, dv->dv_tail, pos);
        if (0 <
            GNUNET_TIME_absolute_get_remaining (path_valid_until).rel_value_us)
          activate_core_visible_dv_path (pos);
        if (last_timeout.rel_value_us <
            GNUNET_TIME_relative_subtract (DV_PATH_VALIDITY_TIMEOUT,
                                           DV_PATH_DISCOVERY_FREQUENCY)
            .rel_value_us)
        {
          /* Some peer send DV learn messages too often, we are learning
             the same path faster than it would be useful; do not forward! */
          GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                      "Rediscovered path too quickly, not forwarding further\n");
          return GNUNET_NO;
        }
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Refreshed known path to %s valid until %s, forwarding further\n",
                    GNUNET_i2s (&dv->target),
                    GNUNET_STRINGS_absolute_time_to_string (
                      pos->path_valid_until));
        return GNUNET_YES;
      }
    }
  }
  /* Count how many shorter paths we have (incl. direct
     neighbours) before simply giving up on this one! */
  if (shorter_distance >= MAX_DV_PATHS_TO_TARGET)
  {
    /* We have a shorter path already! */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Have many shorter DV paths %s, not forwarding further\n",
                GNUNET_i2s (&dv->target));
    return GNUNET_NO;
  }
  /* create new DV path entry */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Discovered new DV path to %s valid until %s\n",
              GNUNET_i2s (&dv->target),
              GNUNET_STRINGS_absolute_time_to_string (path_valid_until));
  hop = GNUNET_malloc (sizeof(struct DistanceVectorHop)
                       + sizeof(struct GNUNET_PeerIdentity) * (path_len - 3));
  hop->next_hop = next_hop;
  hop->dv = dv;
  hop->path = (const struct GNUNET_PeerIdentity *) &hop[1];
  memcpy (&hop[1],
          &path[2],
          sizeof(struct GNUNET_PeerIdentity) * (path_len - 3));
  hop->timeout = GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
  hop->path_valid_until = path_valid_until;
  hop->distance = path_len - 3;
  hop->pd.aged_rtt = network_latency;
  GNUNET_CONTAINER_MDLL_insert (dv, dv->dv_head, dv->dv_tail, hop);
  GNUNET_CONTAINER_MDLL_insert (neighbour,
                                next_hop->dv_head,
                                next_hop->dv_tail,
                                hop);
  if (0 < GNUNET_TIME_absolute_get_remaining (path_valid_until).rel_value_us)
    activate_core_visible_dv_path (hop);
  return GNUNET_YES;
}
 
 
static int
check_dv_learn (void *cls, const struct TransportDVLearnMessage *dvl)
{
  uint16_t size = ntohs (dvl->header.size);
  uint16_t num_hops = ntohs (dvl->num_hops);
  const struct DVPathEntryP *hops = (const struct DVPathEntryP *) &dvl[1];
 
  (void) cls;
  if (size != sizeof(*dvl) + num_hops * sizeof(struct DVPathEntryP))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  if (num_hops > MAX_DV_HOPS_ALLOWED)
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  for (unsigned int i = 0; i < num_hops; i++)
  {
    if (0 == GNUNET_memcmp (&dvl->initiator, &hops[i].hop))
    {
      GNUNET_break_op (0);
      return GNUNET_SYSERR;
    }
    if (0 == GNUNET_memcmp (&GST_my_identity, &hops[i].hop))
    {
      GNUNET_break_op (0);
      return GNUNET_SYSERR;
    }
  }
  return GNUNET_YES;
}
 
 
static void
forward_dv_learn (const struct GNUNET_PeerIdentity *next_hop,
                  const struct TransportDVLearnMessage *msg,
                  uint16_t bi_history,
                  uint16_t nhops,
                  const struct DVPathEntryP *hops,
                  struct GNUNET_TIME_Absolute in_time)
{
  struct Neighbour *n;
  struct VirtualLink *vl;
  struct DVPathEntryP *dhops;
  char buf[sizeof(struct TransportDVLearnMessage)
           + (nhops + 1) * sizeof(struct DVPathEntryP)] GNUNET_ALIGN;
  struct TransportDVLearnMessage *fwd = (struct TransportDVLearnMessage *) buf;
  struct GNUNET_TIME_Relative nnd;
 
  /* compute message for forwarding */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Forwarding DV learn message originating from %s to %s\n",
              GNUNET_i2s (&msg->initiator),
              GNUNET_i2s2 (next_hop));
  GNUNET_assert (nhops < MAX_DV_HOPS_ALLOWED);
  fwd->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN);
  fwd->header.size = htons (sizeof(struct TransportDVLearnMessage)
                            + (nhops + 1) * sizeof(struct DVPathEntryP));
  fwd->num_hops = htons (nhops + 1);
  fwd->bidirectional = htons (bi_history);
  nnd = GNUNET_TIME_relative_add (GNUNET_TIME_absolute_get_duration (in_time),
                                  GNUNET_TIME_relative_ntoh (
                                    msg->non_network_delay));
  fwd->non_network_delay = GNUNET_TIME_relative_hton (nnd);
  fwd->init_sig = msg->init_sig;
  fwd->initiator = msg->initiator;
  fwd->challenge = msg->challenge;
  fwd->monotonic_time = msg->monotonic_time;
  dhops = (struct DVPathEntryP *) &fwd[1];
  GNUNET_memcpy (dhops, hops, sizeof(struct DVPathEntryP) * nhops);
  dhops[nhops].hop = GST_my_identity;
  {
    struct DvHopPS dhp = {
      .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP),
      .purpose.size = htonl (sizeof(dhp)),
      .pred = (0 == nhops) ? msg->initiator : dhops[nhops - 1].hop,
      .succ = *next_hop,
      .challenge = msg->challenge
    };
    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
                              &dhp,
                              &dhops[nhops].hop_sig);
  }
  /*route_control_message_without_fc (next_hop,
                                    &fwd->header,
                                    RMO_UNCONFIRMED_ALLOWED);*/
  vl = lookup_virtual_link (next_hop);
  if ((NULL != vl) && (GNUNET_YES == vl->confirmed))
  {
    route_control_message_without_fc (vl,
                                      &fwd->header,
                                      RMO_UNCONFIRMED_ALLOWED);
  }
  else
  {
    /* Use route via neighbour */
    n = lookup_neighbour (next_hop);
    if (NULL != n)
      route_via_neighbour (
        n,
        &fwd->header,
        RMO_UNCONFIRMED_ALLOWED);
  }
}
 
 
static int
validate_dv_initiator_signature (
  struct GNUNET_TIME_AbsoluteNBO sender_monotonic_time,
  const struct GNUNET_PeerIdentity *init,
  const struct GNUNET_CRYPTO_ChallengeNonceP *challenge,
  const struct GNUNET_CRYPTO_EddsaSignature *init_sig)
{
  struct DvInitPS ip = { .purpose.purpose = htonl (
                           GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR),
                         .purpose.size = htonl (sizeof(ip)),
                         .monotonic_time = sender_monotonic_time,
                         .challenge = *challenge };
 
  if (
    GNUNET_OK !=
    GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR,
                                &ip,
                                init_sig,
                                &init->public_key))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
struct NeighbourSelectionContext
{
  const struct TransportDVLearnMessage *dvl;
 
  const struct DVPathEntryP *hops;
 
  struct GNUNET_TIME_Absolute in_time;
 
  uint32_t selections[MAX_DV_DISCOVERY_SELECTION];
 
  unsigned int num_eligible;
 
  unsigned int num_selections;
 
  uint16_t nhops;
 
  uint16_t bi_history;
};
 
 
static int
dv_neighbour_selection (void *cls,
                        const struct GNUNET_PeerIdentity *pid,
                        void *value)
{
  struct NeighbourSelectionContext *nsc = cls;
 
  (void) value;
  if (0 == GNUNET_memcmp (pid, &nsc->dvl->initiator))
    return GNUNET_YES; /* skip initiator */
  for (unsigned int i = 0; i < nsc->nhops; i++)
    if (0 == GNUNET_memcmp (pid, &nsc->hops[i].hop))
      return GNUNET_YES;
  /* skip peers on path */
  nsc->num_eligible++;
  return GNUNET_YES;
}
 
 
static int
dv_neighbour_transmission (void *cls,
                           const struct GNUNET_PeerIdentity *pid,
                           void *value)
{
  struct NeighbourSelectionContext *nsc = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "transmission %s\n",
              GNUNET_i2s (pid));
  (void) value;
  if (0 == GNUNET_memcmp (pid, &nsc->dvl->initiator))
    return GNUNET_YES; /* skip initiator */
  for (unsigned int i = 0; i < nsc->nhops; i++)
    if (0 == GNUNET_memcmp (pid, &nsc->hops[i].hop))
      return GNUNET_YES;
  /* skip peers on path */
  for (unsigned int i = 0; i < nsc->num_selections; i++)
  {
    if (nsc->selections[i] == nsc->num_eligible)
    {
      forward_dv_learn (pid,
                        nsc->dvl,
                        nsc->bi_history,
                        nsc->nhops,
                        nsc->hops,
                        nsc->in_time);
      break;
    }
  }
  nsc->num_eligible++;
  return GNUNET_YES;
}
 
 
static unsigned int
calculate_fork_degree (unsigned int hops_taken,
                       unsigned int neighbour_count,
                       unsigned int eligible_count)
{
  double target_total = 50.0; /* FIXME: use LOG(NSE)? */
  double eligible_ratio =
    ((double) eligible_count) / ((double) neighbour_count);
  double boost_factor = eligible_ratio * eligible_ratio;
  unsigned int rnd;
  double left;
 
  if (hops_taken >= 64)
  {
    GNUNET_break (0);
    return 0;   /* precaution given bitshift below */
  }
  for (unsigned int i = 1; i < hops_taken; i++)
  {
    /* For each hop, subtract the expected number of targets
       reached at distance d (so what remains divided by 2^d) */
    target_total -= (target_total * boost_factor / (1LLU << i));
  }
  rnd =
    (unsigned int) floor (target_total * boost_factor / (1LLU << hops_taken));
  /* round up or down probabilistically depending on how close we were
     when floor()ing to rnd */
  left = target_total - (double) rnd;
  if (UINT32_MAX * left >
      GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX))
    rnd++; /* round up */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Forwarding DV learn message of %u hops %u(/%u/%u) times\n",
              hops_taken,
              rnd,
              eligible_count,
              neighbour_count);
  return rnd;
}
 
 
static void
neighbour_store_dvmono_cb (void *cls, int success)
{
  struct Neighbour *n = cls;
 
  n->sc = NULL;
  if (GNUNET_YES != success)
    GNUNET_log (