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 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 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 ChallengeNonceP
{
  struct GNUNET_ShortHashCode 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 ChallengeNonceP challenge;
};


struct DvHopPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;

  struct GNUNET_PeerIdentity pred;

  struct GNUNET_PeerIdentity succ;

  struct 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 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;

  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;

  /* 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 ChallengeNonceP challenge;

  struct GNUNET_TIME_AbsoluteNBO sender_time;
};


struct TransportValidationPS
{
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;

  struct GNUNET_TIME_RelativeNBO validity_duration;

  struct ChallengeNonceP challenge;
};


struct TransportValidationResponseMessage
{
  struct GNUNET_MessageHeader header;

  uint32_t reserved GNUNET_PACKED;

  struct GNUNET_CRYPTO_EddsaSignature signature;

  struct 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 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;

  uint16_t total_hops;
};


struct CoreSentContext
{
  struct CoreSentContext *next;

  struct CoreSentContext *prev;

  struct VirtualLink *vl;

  uint16_t size;

  uint16_t isize;
};


struct VirtualLink
{
  struct GNUNET_PeerIdentity target;

  struct CommunicatorMessageContext *cmc_head;

  struct CommunicatorMessageContext *cmc_tail;

  struct PendingMessage *pending_msg_head;

  struct PendingMessage *pending_msg_tail;

  struct CoreSentContext *csc_tail;

  struct CoreSentContext *csc_head;

  struct GNUNET_SCHEDULER_Task *visibility_task;

  struct GNUNET_SCHEDULER_Task *fc_retransmit_task;

  struct 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;

  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;

  enum GNUNET_NetworkType nt;

  enum GNUNET_TRANSPORT_ConnectionStatus cs;

  int idle;
};


struct ReassemblyContext
{
  struct MessageUUIDP msg_uuid;

  struct Neighbour *neighbour;

  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 Neighbour
{
  struct GNUNET_PeerIdentity pid;

  struct GNUNET_CONTAINER_MultiHashMap32 *reassembly_map;

  struct GNUNET_CONTAINER_Heap *reassembly_heap;

  struct GNUNET_SCHEDULER_Task *reassembly_timeout_task;

  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 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;

  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 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 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;

static struct LearnLaunchEntry *lle_tail;

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 PendingAcknowledgement *pa_head;

static struct PendingAcknowledgement *pa_tail;

static struct IncomingRequest *ir_head;

static struct IncomingRequest *ir_tail;

static unsigned int ir_total;

static unsigned long long logging_uuid_gen;

static unsigned int pa_count;

static struct GNUNET_TIME_Absolute hello_mono_time;


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);
  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_CONTAINER_MDLL_remove (pa, pa_head, pa_tail, pa);
  pa_count--;
  if (NULL != q)
  {
    GNUNET_CONTAINER_MDLL_remove (queue, q->pa_head, q->pa_tail, pa);
    pa->queue = NULL;
  }
  if (NULL != pm)
  {
    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);
    GNUNET_free (frag);
  }
}


static void
free_pending_message (struct PendingMessage *pm)
{
  struct TransportClient *tc = pm->client;
  struct VirtualLink *vl = pm->vl;
  struct PendingAcknowledgement *pa;

  if (NULL != tc)
  {
    GNUNET_CONTAINER_MDLL_remove (client,
                                  tc->details.core.pending_msg_head,
                                  tc->details.core.pending_msg_tail,
                                  pm);
  }
  if (NULL != vl)
  {
    GNUNET_CONTAINER_MDLL_remove (vl,
                                  vl->pending_msg_head,
                                  vl->pending_msg_tail,
                                  pm);
  }
  while (NULL != (pa = pm->pa_head))
  {
    GNUNET_CONTAINER_MDLL_remove (pm, pm->pa_head, pm->pa_tail, pa);
    pa->pm = NULL;
  }

  free_fragment_tree (pm);
  if (NULL != pm->qe)
  {
    GNUNET_assert (pm == pm->qe->pm);
    pm->qe->pm = NULL;
  }
  if (NULL != pm->bpm)
  {
    free_fragment_tree (pm->bpm);
    GNUNET_free (pm->bpm);
  }
  GNUNET_free (pm);
}


static void
free_virtual_link (struct VirtualLink *vl)
{
  struct PendingMessage *pm;
  struct CoreSentContext *csc;

  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_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 ((0 != 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 connected\n", tc);
  return tc;
}


static void
free_reassembly_context (struct ReassemblyContext *rc)
{
  struct Neighbour *n = rc->neighbour;

  GNUNET_assert (rc == GNUNET_CONTAINER_heap_remove_node (rc->hn));
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap32_remove (n->reassembly_map,
                                                         rc->msg_uuid.uuid,
                                                         rc));
  GNUNET_free (rc);
}


static void
reassembly_cleanup_task (void *cls)
{
  struct Neighbour *n = cls;
  struct ReassemblyContext *rc;

  n->reassembly_timeout_task = NULL;
  while (NULL != (rc = GNUNET_CONTAINER_heap_peek (n->reassembly_heap)))
  {
    if (0 == GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout)
        .rel_value_us)
    {
      free_reassembly_context (rc);
      continue;
    }
    GNUNET_assert (NULL == n->reassembly_timeout_task);
    n->reassembly_timeout_task =
      GNUNET_SCHEDULER_add_at (rc->reassembly_timeout,
                               &reassembly_cleanup_task,
                               n);
    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_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));
  if (NULL != neighbour->reassembly_map)
  {
    GNUNET_CONTAINER_multihashmap32_iterate (neighbour->reassembly_map,
                                             &free_reassembly_cb,
                                             NULL);
    GNUNET_CONTAINER_multihashmap32_destroy (neighbour->reassembly_map);
    neighbour->reassembly_map = NULL;
    GNUNET_CONTAINER_heap_destroy (neighbour->reassembly_heap);
    neighbour->reassembly_heap = NULL;
  }
  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->reassembly_timeout_task)
  {
    GNUNET_SCHEDULER_cancel (neighbour->reassembly_timeout_task);
    neighbour->reassembly_timeout_task = NULL;
  }
  if (NULL != neighbour->get)
  {
    GNUNET_PEERSTORE_iterate_cancel (neighbour->get);
    neighbour->get = NULL;
  }
  if (NULL != neighbour->sc)
  {
    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 void
schedule_transmit_on_queue (struct Queue *queue,
                            enum GNUNET_SCHEDULER_Priority p)
{
  if (queue->tc->details.communicator.total_queue_length >=
      COMMUNICATOR_TOTAL_QUEUE_LIMIT)
  {
    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_STATISTICS_update (GST_stats,
                              "# Transmission throttled due to queue queue limit",
                              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_with_priority (p, &transmit_on_queue, queue);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Considering transmission on queue `%s' to %s\n",
              queue->address,
              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;

  vl->visibility_task = NULL;
  dvh_timeout = GNUNET_TIME_UNIT_ZERO_ABS;
  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;

  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);
  while (NULL != (qe = queue->queue_head))
  {
    GNUNET_CONTAINER_DLL_remove (queue->queue_head, queue->queue_tail, qe);
    queue->queue_length--;
    tc->details.communicator.total_queue_length--;
    if (NULL != qe->pm)
    {
      GNUNET_assert (qe == qe->pm->qe);
      qe->pm->qe = NULL;
    }
    GNUNET_free (qe);
  }
  GNUNET_assert (0 == queue->queue_length);
  if ((maxxed) && (COMMUNICATOR_TOTAL_QUEUE_LIMIT <
                   tc->details.communicator.total_queue_length))
  {
    /* Communicator dropped below threshold, resume all _other_ queues */
    GNUNET_STATISTICS_update (
      GST_stats,
      "# Transmission throttled due to communicator queue limit",
      -1,
      GNUNET_NO);
    for (struct Queue *s = tc->details.communicator.queue_head; NULL != s;
         s = s->next_client)
      schedule_transmit_on_queue (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_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);
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (tc->details.application.requests,
                                          pid,
                                          pr));
  GNUNET_free (pr);

  return GNUNET_OK;
}


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_log (GNUNET_ERROR_TYPE_DEBUG,
              "Client %p disconnected, cleaning up.\n",
              tc);
  GNUNET_CONTAINER_DLL_remove (clients_head, clients_tail, tc);
  switch (tc->type)
  {
  case CT_NONE:
    break;

  case CT_CORE: {
      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:
    break;

  case CT_COMMUNICATOR: {
      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_CONTAINER_multipeermap_iterate (tc->details.application.requests,
                                           &stop_peer_request,
                                           tc);
    GNUNET_CONTAINER_multipeermap_destroy (tc->details.application.requests);
    break;
  }
  GNUNET_free (tc);
}


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;

  if (NULL != tc)
  {
    struct GNUNET_MQ_Envelope *env;
    struct SendOkMessage *som;

    env = GNUNET_MQ_msg (som, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
    som->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 (struct CommunicatorMessageContext *cmc)
{
  if (0 != ntohl (cmc->im.fc_on))
  {
    /* send ACK when done to communicator for flow control! */
    struct GNUNET_MQ_Envelope *env;
    struct GNUNET_TRANSPORT_IncomingMessageAck *ack;

    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.sender;
    GNUNET_MQ_send (cmc->tc->mq, env);
  }
  GNUNET_SERVICE_client_continue (cmc->tc->client);
  GNUNET_free (cmc);
}


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_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;
  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);
  }
}


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_assert (GNUNET_OK ==
                 GNUNET_CRYPTO_ecdhe_key_create2 (&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.purpose.size = htonl (sizeof(ec));
  ec.target = dv->target;
  ec.ephemeral_key = dv->ephemeral_key;
  GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
                                                        &ec.purpose,
                                                        &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;

  queue->idle = GNUNET_NO;
  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;
      GNUNET_assert (NULL == pm->qe);
      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++;
    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;
    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)));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Deriving backchannel key based on KM %s and IV %s\n",
              GNUNET_h2s (km),
              GNUNET_sh2s (iv));
  gcry_cipher_open (&key->cipher,
                    GCRY_CIPHER_AES256 /* low level: go for speed */,
                    GCRY_CIPHER_MODE_CTR,
                    0 /* flags */);
  gcry_cipher_setkey (key->cipher,
                      &key->material.aes_key,
                      sizeof(key->material.aes_key));
  gcry_cipher_setctr (key->cipher,
                      &key->material.aes_ctr,
                      sizeof(key->material.aes_ctr));
}


static void
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;

  GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_ecdh_eddsa (priv_ephemeral,
                                                         &target->public_key,
                                                         &km));
  dv_setup_key_state_from_km (&km, iv, key);
}


static void
dh_key_derive_eph_pub (const struct GNUNET_CRYPTO_EcdhePublicKey *pub_ephemeral,
                       const struct GNUNET_ShortHashCode *iv,
                       struct DVKeyState *key)
{
  struct GNUNET_HashCode km;

  GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_eddsa_ecdh (GST_my_private_key,
                                                         pub_ephemeral,
                                                         &km));
  dv_setup_key_state_from_km (&km, iv, key);
}


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 void
dv_decrypt (struct DVKeyState *key,
            void *out,
            const void *ciph,
            size_t out_size)
{
  GNUNET_assert (
    0 == gcry_cipher_decrypt (key->cipher, out, out_size, ciph, out_size));
}


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)
{
  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 TransportDVBoxPayloadP *enc_payload_hdr =
    (struct TransportDVBoxPayloadP *) enc;
  struct DVKeyState key;
  struct GNUNET_TIME_Relative rtt;

  /* Encrypt payload */
  box_hdr.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX);
  box_hdr.total_hops = htons (0);
  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));
  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_payload_hdr, 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.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 + 1,
                  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);
  }
  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, options);
}


static struct GNUNET_TIME_Relative
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;

  vl = lookup_virtual_link (target);
  GNUNET_assert (NULL != vl);
  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 redunancy, so clear the flag. */
  rtt1 = GNUNET_TIME_UNIT_FOREVER_REL;
  rtt2 = GNUNET_TIME_UNIT_FOREVER_REL;
  if (NULL != n)
  {
    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;
    }
    rtt2 = encapsulate_for_dv (dv,
                               res,
                               hops,
                               hdr,
                               &send_dv_to_neighbour,
                               NULL,
                               options & (~RMO_REDUNDANT));
  }
  return GNUNET_TIME_relative_min (rtt1, rtt2);
}


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->target, &fc.header, RMO_NONE);
  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);
  vl->fc_retransmit_task =
    GNUNET_SCHEDULER_add_delayed (rtt, &consider_sending_fc, vl);
}


static void
check_vl_transmission (struct VirtualLink *vl)
{
  struct Neighbour *n = vl->n;
  struct DistanceVector *dv = vl->dv;
  struct GNUNET_TIME_Absolute now;
  int elig;

  /* 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)
  {
    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 transmision on VL %s due to flow control: %llu < %llu\n",
                  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" */
    }
    elig = GNUNET_YES;
    break;
  }
  if (GNUNET_NO == elig)
    return;

  /* 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))
        schedule_transmit_on_queue (queue, GNUNET_SCHEDULER_PRIORITY_DEFAULT);
  }
  /* 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 */
      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))
          schedule_transmit_on_queue (queue,
                                      GNUNET_SCHEDULER_PRIORITY_BACKGROUND);
    }
  }
}


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)
  {
    /* 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);
  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);
}


static void
handle_communicator_backchannel (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
{
  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;
  char
    mbuf[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\n",
              GNUNET_i2s (&cb->pid),
              is,
              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);
  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);
  }
  GNUNET_break (0);
  GNUNET_SERVICE_client_drop (tc->client);
}


static void
demultiplex_with_cmc (struct CommunicatorMessageContext *cmc,
                      const struct GNUNET_MessageHeader *msg);


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
handle_raw_message (void *cls, const struct GNUNET_MessageHeader *mh)
{
  struct CommunicatorMessageContext *cmc = cls;
  struct VirtualLink *vl;
  uint16_t size = ntohs (mh->size);
  int have_core;

  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)
  {
    /* 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. */GNUNET_break_op (0);
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages droped (virtual link still down)",
                              1,
                              GNUNET_NO);

    finish_cmc_handling (cmc);
    return;
  }
  if (vl->incoming_fc_window_size_ram > UINT_MAX - size)
  {
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages droped (FC arithmetic overflow)",
                              1,
                              GNUNET_NO);

    finish_cmc_handling (cmc);
    return;
  }
  if (vl->incoming_fc_window_size_ram + size > vl->available_fc_window_size)
  {
    GNUNET_STATISTICS_update (GST_stats,
                              "# CORE messages droped (FC window overflow)",
                              1,
                              GNUNET_NO);
    finish_cmc_handling (cmc);
    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 */
    finish_cmc_handling (cmc);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Delivered message from %s of type %u to CORE\n",
              GNUNET_i2s (&cmc->im.sender),
              ntohs (mh->type));
  if (vl->core_recv_window > 0)
  {
    finish_cmc_handling (cmc);
    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 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 AcknowledgementCummulator *ac = cls;
  char buf[sizeof(struct TransportReliabilityAckMessage)
           + ac->ack_counter
           * 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->ack_counter,
              GNUNET_i2s (&ac->target));
  GNUNET_assert (0 < ac->ack_counter);
  ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK);
  ack->header.size =
    htons (sizeof(*ack)
           + ac->ack_counter * sizeof(struct TransportCummulativeAckPayloadP));
  ack->ack_counter = htonl (ac->ack_counter++);
  ap = (struct TransportCummulativeAckPayloadP *) &ack[1];
  for (unsigned int i = 0; i < ac->ack_counter; 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);
  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! */
      GNUNET_SCHEDULER_cancel (ac->task);
      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 Neighbour *n;
  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;

  n = lookup_neighbour (&cmc->im.sender);
  if (NULL == n)
  {
    struct GNUNET_SERVICE_Client *client = cmc->tc->client;

    GNUNET_break (0);
    finish_cmc_handling (cmc);
    GNUNET_SERVICE_client_drop (client);
    return;
  }
  if (NULL == n->reassembly_map)
  {
    n->reassembly_map = GNUNET_CONTAINER_multihashmap32_create (8);
    n->reassembly_heap =
      GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
    n->reassembly_timeout_task =
      GNUNET_SCHEDULER_add_delayed (REASSEMBLY_EXPIRATION,
                                    &reassembly_cleanup_task,
                                    n);
  }
  msize = ntohs (fb->msg_size);
  fc.message_uuid = fb->msg_uuid;
  fc.rc = NULL;
  (void) GNUNET_CONTAINER_multihashmap32_get_multiple (n->reassembly_map,
                                                       fb->msg_uuid.uuid,
                                                       &find_by_message_uuid,
                                                       &fc);
  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->neighbour = n;
    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 (n->reassembly_heap,
                                           rc,
                                           rc->reassembly_timeout.abs_value_us);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multihashmap32_put (
                     n->reassembly_map,
                     rc->msg_uuid.uuid,
                     rc,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
    target = (char *) &rc[1];
    rc->bitfield = (uint8_t *) (target + rc->msg_size);
    rc->msg_missing = rc->msg_size;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received fragment at offset %u/%u from %s for NEW message %u\n",
                ntohs (fb->frag_off),
                msize,
                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 */
  fsize = ntohs (fb->header.size) - sizeof(*fb);
  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 cummulative 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! */
  demultiplex_with_cmc (cmc, msg);
  /* 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;
  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!) */
  demultiplex_with_cmc (cmc, inbox);
}


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;

  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 reliabile channel */
    free_fragment_tree (pm);
    pos = pm->frag_parent;
    GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, pm);
    GNUNET_free (pm);
    /* check if subtree is done */
    while ((NULL == pos->head_frag) && (pos->frag_off == pos->bytes_msg) &&
           (pos != pm))
    {
      pm = pos;
      pos = pm->frag_parent;
      GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, pm);
      GNUNET_free (pm);
    }

    /* Was this the last applicable fragmment? */
    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);
    free_pending_message (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;

  /* 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);
    return;
  }
  /* Finally, deliver backchannel message to communicator */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Delivering backchannel message from %s of type %u to %s\n",
              GNUNET_i2s (&cmc->im.sender),
              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);
}


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
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)
  {
    /* Link was already up, remember dv is also now available and we are done */
    vl->dv = dv;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Virtual link to %s could now also use DV!\n",
                GNUNET_i2s (&dv->target));
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Creating new virtual link to %s using DV!\n",
              GNUNET_i2s (&dv->target));
  vl = GNUNET_new (struct VirtualLink);
  vl->message_uuid_ctr =
    GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
  vl->target = dv->target;
  vl->dv = dv;
  dv->vl = vl;
  vl->core_recv_window = RECV_WINDOW_SIZE;
  vl->available_fc_window_size = DEFAULT_WINDOW_SIZE;
  vl->visibility_task =
    GNUNET_SCHEDULER_add_at (hop->path_valid_until, &check_link_down, vl);
  GNUNET_break (GNUNET_YES ==
                GNUNET_CONTAINER_multipeermap_put (
                  links,
                  &vl->target,
                  vl,
                  GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  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);
}


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 - 2)
      shorter_distance++;
    /* Note that the distances in 'pos' excludes us (path[0]) and
       the next_hop (path[1]), so we need to subtract two
       and check next_hop explicitly */
    if ((pos->distance == path_len - 2) && (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, forwarding further\n",
                    GNUNET_i2s (&dv->target));
        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\n",
              GNUNET_i2s (&dv->target));
  hop = GNUNET_malloc (sizeof(struct DistanceVectorHop)
                       + sizeof(struct GNUNET_PeerIdentity) * (path_len - 2));
  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 - 2));
  hop->timeout = GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
  hop->path_valid_until = path_valid_until;
  hop->distance = path_len - 2;
  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 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;
  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 = dhops[nhops - 1].hop,
                           .succ = *next_hop,
                           .challenge = msg->challenge };

    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
                                             &dhp.purpose,
                                             &dhops[nhops].hop_sig));
  }
  route_control_message_without_fc (next_hop,
                                    &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 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.purpose,
                                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;

  (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 (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store other peer's monotonic time in peerstore!\n");
}


static void
handle_dv_learn (void *cls, const struct TransportDVLearnMessage *dvl)
{
  struct CommunicatorMessageContext *cmc = cls;
  enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
  int bi_hop;
  uint16_t nhops;
  uint16_t bi_history;
  const struct DVPathEntryP *hops;
  int do_fwd;
  int did_initiator;
  struct GNUNET_TIME_Absolute in_time;
  struct Neighbour *n;

  nhops = ntohs (dvl->bidirectional);  /* 0 = sender is initiator */
  bi_history = ntohs (dvl->bidirectional);
  hops = (const struct DVPathEntryP *) &dvl[1];
  if (0 == nhops)
  {
    /* sanity check */
    if (0 != GNUNET_memcmp (&dvl->initiator, &cmc->im.sender))
    {
      GNUNET_break (0);
      finish_cmc_handling (cmc);
      return;
    }
  }
  else
  {
    /* sanity check */
    if (0 != GNUNET_memcmp (&hops[nhops - 1].hop, &cmc->im.sender))
    {
      GNUNET_break (0);
      finish_cmc_handling (cmc);
      return;
    }
  }

  GNUNET_assert (CT_COMMUNICATOR == cmc->tc->type);
  cc = cmc->tc->details.communicator.cc;
  bi_hop = (GNUNET_TRANSPORT_CC_RELIABLE ==
            cc); // FIXME: add bi-directional flag to cc?
  in_time = GNUNET_TIME_absolute_get ();

  /* continue communicator here, everything else can happen asynchronous! */
  finish_cmc_handling (cmc);

  n = lookup_neighbour (&dvl->initiator);
  if (NULL != n)
  {
    if ((n->dv_monotime_available == GNUNET_YES) &&
        (GNUNET_TIME_absolute_ntoh (dvl->monotonic_time).abs_value_us <
         n->last_dv_learn_monotime.abs_value_us))
    {
      GNUNET_STATISTICS_update (GST_stats,
                                "# DV learn discarded due to time travel",
                                1,
                                GNUNET_NO);
      return;
    }
    if (GNUNET_OK != validate_dv_initiator_signature (dvl->monotonic_time,
                                                      &dvl->initiator,
                                                      &dvl->challenge,
                                                      &dvl->init_sig))
    {
      GNUNET_break_op (0);
      return;
    }
    n->last_dv_learn_monotime = GNUNET_TIME_absolute_ntoh (dvl->monotonic_time);
    if (GNUNET_YES == n->dv_monotime_available)
    {
      if (NULL != n->sc)
        GNUNET_PEERSTORE_store_cancel (n->sc);
      n->sc =
        GNUNET_PEERSTORE_store (peerstore,
                                "transport",
                                &dvl->initiator,
                                GNUNET_PEERSTORE_TRANSPORT_DVLEARN_MONOTIME,
                                &dvl->monotonic_time,
                                sizeof(dvl->monotonic_time),
                                GNUNET_TIME_UNIT_FOREVER_ABS,
                                GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                                &neighbour_store_dvmono_cb,
                                n);
    }
  }
  /* OPTIMIZE-FIXME: asynchronously (!) verify signatures!,
     If signature verification load too high, implement random drop strategy */
  for (unsigned int i = 0; i < nhops; i++)
  {
    struct DvHopPS dhp = { .purpose.purpose =
                             htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP),
                           .purpose.size = htonl (sizeof(dhp)),
                           .pred = (0 == i) ? dvl->initiator : hops[i - 1].hop,
                           .succ = (nhops == i + 1) ? GST_my_identity
                                   : hops[i + 1].hop,
                           .challenge = dvl->challenge };

    if (GNUNET_OK !=
        GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP,
                                    &dhp.purpose,
                                    &hops[i].hop_sig,
                                    &hops[i].hop.public_key))
    {
      GNUNET_break_op (0);
      return;
    }
  }

  if (GNUNET_EXTRA_LOGGING > 0)
  {
    char *path;

    path = GNUNET_strdup (GNUNET_i2s (&dvl->initiator));
    for (unsigned int i = 0; i < nhops; i++)
    {
      char *tmp;

      GNUNET_asprintf (&tmp,
                       "%s%s%s",
                       path,
                       (bi_history & (1 << (nhops - i))) ? "<->" : "-->",
                       GNUNET_i2s (&hops[i].hop));
      GNUNET_free (path);
      path = tmp;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received DVInit via %s%s%s\n",
                path,
                bi_hop ? "<->" : "-->",
                GNUNET_i2s (&GST_my_identity));
    GNUNET_free (path);
  }

  do_fwd = GNUNET_YES;
  if (0 == GNUNET_memcmp (&GST_my_identity, &dvl->initiator))
  {
    struct GNUNET_PeerIdentity path[nhops + 1];
    struct GNUNET_TIME_Relative host_latency_sum;
    struct GNUNET_TIME_Relative latency;
    struct GNUNET_TIME_Relative network_latency;

    /* We initiated this, learn the forward path! */
    path[0] = GST_my_identity;
    path[1] = hops[0].hop;
    host_latency_sum = GNUNET_TIME_relative_ntoh (dvl->non_network_delay);

    // Need also something to lookup initiation time
    // to compute RTT! -> add RTT argument here?
    latency = GNUNET_TIME_UNIT_FOREVER_REL;   // FIXME: initialize properly
    // (based on dvl->challenge, we can identify time of origin!)

    network_latency = GNUNET_TIME_relative_subtract (latency, host_latency_sum);
    /* assumption: latency on all links is the same */
    network_latency = GNUNET_TIME_relative_divide (network_latency, nhops);

    for (unsigned int i = 2; i <= nhops; i++)
    {
      struct GNUNET_TIME_Relative ilat;

      /* assumption: linear latency increase per hop */
      ilat = GNUNET_TIME_relative_multiply (network_latency, i);
      path[i] = hops[i - 1].hop;
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Learned path with %u hops to %s with latency %s\n",
                  i,
                  GNUNET_i2s (&path[i]),
                  GNUNET_STRINGS_relative_time_to_string (ilat, GNUNET_YES));
      learn_dv_path (path,
                     i,
                     ilat,
                     GNUNET_TIME_relative_to_absolute (
                       ADDRESS_VALIDATION_LIFETIME));
    }
    /* as we initiated, do not forward again (would be circular!) */
    do_fwd = GNUNET_NO;
    return;
  }
  if (bi_hop)
  {
    /* last hop was bi-directional, we could learn something here! */
    struct GNUNET_PeerIdentity path[nhops + 2];

    path[0] = GST_my_identity;
    path[1] = hops[nhops - 1].hop;   /* direct neighbour == predecessor! */
    for (unsigned int i = 0; i < nhops; i++)
    {
      int iret;

      if (0 == (bi_history & (1 << i)))
        break;     /* i-th hop not bi-directional, stop learning! */
      if (i == nhops - 1)
      {
        path[i + 2] = dvl->initiator;
      }
      else
      {
        path[i + 2] = hops[nhops - i - 2].hop;
      }

      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Learned inverse path with %u hops to %s\n",
                  i + 1,
                  GNUNET_i2s (&path[i + 2]));
      iret = learn_dv_path (path,
                            i + 2,
                            GNUNET_TIME_UNIT_FOREVER_REL,
                            GNUNET_TIME_UNIT_ZERO_ABS);
      if (GNUNET_SYSERR == iret)
      {
        /* path invalid or too long to be interesting for US, thus should also
           not be interesting to our neighbours, cut path when forwarding to
           'i' hops, except of course for the one that goes back to the
           initiator */
        GNUNET_STATISTICS_update (GST_stats,
                                  "# DV learn not forwarded due invalidity of path",
                                  1,
                                  GNUNET_NO);
        do_fwd = GNUNET_NO;
        break;
      }
      if ((GNUNET_NO == iret) && (nhops == i + 1))
      {
        /* we have better paths, and this is the longest target,
           so there cannot be anything interesting later */
        GNUNET_STATISTICS_update (GST_stats,
                                  "# DV learn not forwarded, got better paths",
                                  1,
                                  GNUNET_NO);
        do_fwd = GNUNET_NO;
        break;
      }
    }
  }

  if (MAX_DV_HOPS_ALLOWED == nhops)
  {
    /* At limit, we're out of here! */
    finish_cmc_handling (cmc);
    return;
  }

  /* Forward to initiator, if path non-trivial and possible */
  bi_history = (bi_history << 1) | (bi_hop ? 1 : 0);
  did_initiator = GNUNET_NO;
  if ((1 < nhops) &&
      (GNUNET_YES ==
       GNUNET_CONTAINER_multipeermap_contains (neighbours, &dvl->initiator)))
  {
    /* send back to origin! */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending DVL back to initiator %s\n",
                GNUNET_i2s (&dvl->initiator));
    forward_dv_learn (&dvl->initiator, dvl, bi_history, nhops, hops, in_time);
    did_initiator = GNUNET_YES;
  }
  /* We forward under two conditions: either we still learned something
     ourselves (do_fwd), or the path was darn short and thus the initiator is
     likely to still be very interested in this (and we did NOT already
     send it back to the initiator) */
  if ((do_fwd) || ((nhops < MIN_DV_PATH_LENGTH_FOR_INITIATOR) &&
                   (GNUNET_NO == did_initiator)))
  {
    /* Pick random neighbours that are not yet on the path */
    struct NeighbourSelectionContext nsc;
    unsigned int n_cnt;

    n_cnt = GNUNET_CONTAINER_multipeermap_size (neighbours);
    nsc.nhops = nhops;
    nsc.dvl = dvl;
    nsc.bi_history = bi_history;
    nsc.hops = hops;
    nsc.in_time = in_time;
    nsc.num_eligible = 0;
    GNUNET_CONTAINER_multipeermap_iterate (neighbours,
                                           &dv_neighbour_selection,
                                           &nsc);
    if (0 == nsc.num_eligible)
      return;   /* done here, cannot forward to anyone else */
    nsc.num_selections = calculate_fork_degree (nhops, n_cnt, nsc.num_eligible);
    nsc.num_selections =
      GNUNET_MIN (MAX_DV_DISCOVERY_SELECTION, nsc.num_selections);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Forwarding DVL to %u other peers\n",
                nsc.num_selections);
    for (unsigned int i = 0; i < nsc.num_selections; i++)
      nsc.selections[i] =
        (nsc.num_selections == n_cnt)
        ? i   /* all were selected, avoid collisions by chance */
        : GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, n_cnt);
    nsc.num_eligible = 0;
    GNUNET_CONTAINER_multipeermap_iterate (neighbours,
                                           &dv_neighbour_transmission,
                                           &nsc);
  }
}


static int
check_dv_box (void *cls, const struct TransportDVBoxMessage *dvb)
{
  uint16_t size = ntohs (dvb->header.size);
  uint16_t num_hops = ntohs (dvb->num_hops);
  const struct GNUNET_PeerIdentity *hops =
    (const struct GNUNET_PeerIdentity *) &dvb[1];

  (void) cls;
  if (size < sizeof(*dvb) + num_hops * sizeof(struct GNUNET_PeerIdentity)
      + sizeof(struct GNUNET_MessageHeader))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  /* This peer must not be on the path */
  for (unsigned int i = 0; i < num_hops; i++)
    if (0 == GNUNET_memcmp (&hops[i], &GST_my_identity))
    {
      GNUNET_break_op (0);
      return GNUNET_SYSERR;
    }
  return GNUNET_YES;
}


static void
forward_dv_box (struct Neighbour *next_hop,
                const struct TransportDVBoxMessage *hdr,
                uint16_t total_hops,
                uint16_t num_hops,
                const struct GNUNET_PeerIdentity *hops,
                const void *enc_payload,
                uint16_t enc_payload_size)
{
  struct VirtualLink *vl = next_hop->vl;
  struct PendingMessage *pm;
  size_t msg_size;
  char *buf;
  struct GNUNET_PeerIdentity *dhops;

  GNUNET_assert (NULL != vl);
  msg_size = sizeof(struct TransportDVBoxMessage)
             + num_hops * sizeof(struct GNUNET_PeerIdentity) + enc_payload_size;
  pm = GNUNET_malloc (sizeof(struct PendingMessage) + msg_size);
  pm->pmt = PMT_DV_BOX;
  pm->vl = vl;
  pm->timeout = GNUNET_TIME_relative_to_absolute (DV_FORWARD_TIMEOUT);
  pm->logging_uuid = logging_uuid_gen++;
  pm->prefs = GNUNET_MQ_PRIO_BACKGROUND;
  pm->bytes_msg = msg_size;
  buf = (char *) &pm[1];
  memcpy (buf, hdr, sizeof(*hdr));
  dhops =
    (struct GNUNET_PeerIdentity *) &buf[sizeof(struct TransportDVBoxMessage)];
  memcpy (dhops, hops, num_hops * sizeof(struct GNUNET_PeerIdentity));
  memcpy (&dhops[num_hops], enc_payload, enc_payload_size);
  GNUNET_CONTAINER_MDLL_insert (vl,
                                vl->pending_msg_head,
                                vl->pending_msg_tail,
                                pm);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Created pending message %llu for DV Box with next hop %s (%u/%u)\n",
              pm->logging_uuid,
              GNUNET_i2s (&next_hop->pid),
              (unsigned int) num_hops,
              (unsigned int) total_hops);
  check_vl_transmission (vl);
}


static void
free_backtalker (struct Backtalker *b)
{
  if (NULL != b->get)
  {
    GNUNET_PEERSTORE_iterate_cancel (b->get);
    b->get = NULL;
    GNUNET_assert (NULL != b->cmc);
    finish_cmc_handling (b->cmc);
    b->cmc = NULL;
  }
  if (NULL != b->task)
  {
    GNUNET_SCHEDULER_cancel (b->task);
    b->task = NULL;
  }
  if (NULL != b->sc)
  {
    GNUNET_PEERSTORE_store_cancel (b->sc);
    b->sc = NULL;
  }
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (backtalkers, &b->pid, b));
  GNUNET_free (b);
}


static int
free_backtalker_cb (void *cls,
                    const struct GNUNET_PeerIdentity *pid,
                    void *value)
{
  struct Backtalker *b = value;

  (void) cls;
  (void) pid;
  free_backtalker (b);
  return GNUNET_OK;
}


static void
backtalker_timeout_cb (void *cls)
{
  struct Backtalker *b = cls;

  b->task = NULL;
  if (0 != GNUNET_TIME_absolute_get_remaining (b->timeout).rel_value_us)
  {
    b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
    return;
  }
  GNUNET_assert (NULL == b->sc);
  free_backtalker (b);
}


static void
backtalker_monotime_cb (void *cls,
                        const struct GNUNET_PEERSTORE_Record *record,
                        const char *emsg)
{
  struct Backtalker *b = cls;
  struct GNUNET_TIME_AbsoluteNBO *mtbe;
  struct GNUNET_TIME_Absolute mt;

  (void) emsg;
  if (NULL == record)
  {
    /* we're done with #backtalker_monotime_cb() invocations,
       continue normal processing */
    b->get = NULL;
    GNUNET_assert (NULL != b->cmc);
    if (0 != b->body_size)
      demultiplex_with_cmc (b->cmc,
                            (const struct GNUNET_MessageHeader *) &b[1]);
    else
      finish_cmc_handling (b->cmc);
    b->cmc = NULL;
    return;
  }
  if (sizeof(*mtbe) != record->value_size)
  {
    GNUNET_break (0);
    return;
  }
  mtbe = record->value;
  mt = GNUNET_TIME_absolute_ntoh (*mtbe);
  if (mt.abs_value_us > b->monotonic_time.abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Backtalker message from %s dropped, monotime in the past\n",
                GNUNET_i2s (&b->pid));
    GNUNET_STATISTICS_update (
      GST_stats,
      "# Backchannel messages dropped: monotonic time not increasing",
      1,
      GNUNET_NO);
    b->monotonic_time = mt;
    /* Setting body_size to 0 prevents call to #forward_backchannel_payload()
     */
    b->body_size = 0;
    return;
  }
}


static void
backtalker_monotime_store_cb (void *cls, int success)
{
  struct Backtalker *b = cls;

  if (GNUNET_OK != success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store backtalker's monotonic time in PEERSTORE!\n");
  }
  b->sc = NULL;
  b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
}


static void
update_backtalker_monotime (struct Backtalker *b)
{
  struct GNUNET_TIME_AbsoluteNBO mtbe;

  if (NULL != b->sc)
  {
    GNUNET_PEERSTORE_store_cancel (b->sc);
    b->sc = NULL;
  }
  else
  {
    GNUNET_SCHEDULER_cancel (b->task);
    b->task = NULL;
  }
  mtbe = GNUNET_TIME_absolute_hton (b->monotonic_time);
  b->sc =
    GNUNET_PEERSTORE_store (peerstore,
                            "transport",
                            &b->pid,
                            GNUNET_PEERSTORE_TRANSPORT_BACKCHANNEL_MONOTIME,
                            &mtbe,
                            sizeof(mtbe),
                            GNUNET_TIME_UNIT_FOREVER_ABS,
                            GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                            &backtalker_monotime_store_cb,
                            b);
}


static void
handle_dv_box (void *cls, const struct TransportDVBoxMessage *dvb)
{
  struct CommunicatorMessageContext *cmc = cls;
  uint16_t size = ntohs (dvb->header.size) - sizeof(*dvb);
  uint16_t num_hops = ntohs (dvb->num_hops);
  const struct GNUNET_PeerIdentity *hops =
    (const struct GNUNET_PeerIdentity *) &dvb[1];
  const char *enc_payload = (const char *) &hops[num_hops];
  uint16_t enc_payload_size =
    size - (num_hops * sizeof(struct GNUNET_PeerIdentity));
  struct DVKeyState key;
  struct GNUNET_HashCode hmac;
  const char *hdr;
  size_t hdr_len;

  if (GNUNET_EXTRA_LOGGING > 0)
  {
    char *path;

    path = GNUNET_strdup (GNUNET_i2s (&GST_my_identity));
    for (unsigned int i = 0; i < num_hops; i++)
    {
      char *tmp;

      GNUNET_asprintf (&tmp, "%s->%s", path, GNUNET_i2s (&hops[i]));
      GNUNET_free (path);
      path = tmp;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received DVBox with remainig path %s\n",
                path);
    GNUNET_free (path);
  }

  if (num_hops > 0)
  {
    /* We're trying from the end of the hops array, as we may be
       able to find a shortcut unknown to the origin that way */
    for (int i = num_hops - 1; i >= 0; i--)
    {
      struct Neighbour *n;

      if (0 == GNUNET_memcmp (&hops[i], &GST_my_identity))
      {
        GNUNET_break_op (0);
        finish_cmc_handling (cmc);
        return;
      }
      n = lookup_neighbour (&hops[i]);
      if (NULL == n)
        continue;
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Skipping %u/%u hops ahead while routing DV Box\n",
                  i,
                  num_hops);
      forward_dv_box (n,
                      dvb,
                      ntohs (dvb->total_hops) + 1,
                      num_hops - i - 1,    /* number of hops left */
                      &hops[i + 1],    /* remaining hops */
                      enc_payload,
                      enc_payload_size);
      GNUNET_STATISTICS_update (GST_stats,
                                "# DV hops skipped routing boxes",
                                i,
                                GNUNET_NO);
      GNUNET_STATISTICS_update (GST_stats,
                                "# DV boxes routed (total)",
                                1,
                                GNUNET_NO);
      finish_cmc_handling (cmc);
      return;
    }
    /* Woopsie, next hop not in neighbours, drop! */
    GNUNET_STATISTICS_update (GST_stats,
                              "# DV Boxes dropped: next hop unknown",
                              1,
                              GNUNET_NO);
    finish_cmc_handling (cmc);
    return;
  }
  /* We are the target. Unbox and handle message. */
  GNUNET_STATISTICS_update (GST_stats,
                            "# DV boxes opened (ultimate target)",
                            1,
                            GNUNET_NO);
  cmc->total_hops = ntohs (dvb->total_hops);

  dh_key_derive_eph_pub (&dvb->ephemeral_key, &dvb->iv, &key);
  hdr = (const char *) &dvb[1];
  hdr_len = ntohs (dvb->header.size) - sizeof(*dvb);
  dv_hmac (&key, &hmac, hdr, hdr_len);
  if (0 != GNUNET_memcmp (&hmac, &dvb->hmac))
  {
    /* HMAC missmatch, disard! */
    GNUNET_break_op (0);
    finish_cmc_handling (cmc);
    return;
  }
  /* begin actual decryption */
  {
    struct Backtalker *b;
    struct GNUNET_TIME_Absolute monotime;
    struct TransportDVBoxPayloadP ppay;
    char body[hdr_len - sizeof(ppay)] GNUNET_ALIGN;
    const struct GNUNET_MessageHeader *mh =
      (const struct GNUNET_MessageHeader *) body;

    GNUNET_assert (hdr_len >=
                   sizeof(ppay) + sizeof(struct GNUNET_MessageHeader));
    dv_decrypt (&key, &ppay, hdr, sizeof(ppay));
    dv_decrypt (&key, &body, &hdr[sizeof(ppay)], hdr_len - sizeof(ppay));
    dv_key_clean (&key);
    if (ntohs (mh->size) != sizeof(body))
    {
      GNUNET_break_op (0);
      finish_cmc_handling (cmc);
      return;
    }
    /* need to prevent box-in-a-box (and DV_LEARN) so check inbox type! */
    switch (ntohs (mh->type))
    {
    case GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX:
      GNUNET_break_op (0);
      finish_cmc_handling (cmc);
      return;

    case GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN:
      GNUNET_break_op (0);
      finish_cmc_handling (cmc);
      return;

    default:
      /* permitted, continue */
      break;
    }
    monotime = GNUNET_TIME_absolute_ntoh (ppay.monotonic_time);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Decrypted backtalk from %s\n",
                GNUNET_i2s (&ppay.sender));
    b = GNUNET_CONTAINER_multipeermap_get (backtalkers, &ppay.sender);
    if ((NULL != b) && (monotime.abs_value_us < b->monotonic_time.abs_value_us))
    {
      GNUNET_STATISTICS_update (
        GST_stats,
        "# Backchannel messages dropped: monotonic time not increasing",
        1,
        GNUNET_NO);
      finish_cmc_handling (cmc);
      return;
    }
    if ((NULL == b) ||
        (0 != GNUNET_memcmp (&b->last_ephemeral, &dvb->ephemeral_key)))
    {
      /* Check signature */
      struct EphemeralConfirmationPS ec;

      ec.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL);
      ec.purpose.size = htonl (sizeof(ec));
      ec.target = GST_my_identity;
      ec.ephemeral_key = dvb->ephemeral_key;
      if (
        GNUNET_OK !=
        GNUNET_CRYPTO_eddsa_verify (
          GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL,
          &ec.purpose,
          &ppay.sender_sig,
          &ppay.sender.public_key))
      {
        /* Signature invalid, disard! */
        GNUNET_break_op (0);
        finish_cmc_handling (cmc);
        return;
      }
    }
    /* Update sender, we now know the real origin! */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "DVBox received for me from %s via %s\n",
                GNUNET_i2s2 (&ppay.sender),
                GNUNET_i2s (&cmc->im.sender));
    cmc->im.sender = ppay.sender;

    if (NULL != b)
    {
      /* update key cache and mono time */
      b->last_ephemeral = dvb->ephemeral_key;
      b->monotonic_time = monotime;
      update_backtalker_monotime (b);
      b->timeout =
        GNUNET_TIME_relative_to_absolute (BACKCHANNEL_INACTIVITY_TIMEOUT);

      demultiplex_with_cmc (cmc, mh);
      return;
    }
    /* setup data structure to cache signature AND check
       monotonic time with PEERSTORE before forwarding backchannel payload */
    b = GNUNET_malloc (sizeof(struct Backtalker) + sizeof(body));
    b->pid = ppay.sender;
    b->body_size = sizeof(body);
    memcpy (&b[1], body, sizeof(body));
    GNUNET_assert (GNUNET_YES ==
                   GNUNET_CONTAINER_multipeermap_put (
                     backtalkers,
                     &b->pid,
                     b,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    b->monotonic_time = monotime; /* NOTE: to be checked still! */
    b->cmc = cmc;
    b->timeout =
      GNUNET_TIME_relative_to_absolute (BACKCHANNEL_INACTIVITY_TIMEOUT);
    b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
    b->get =
      GNUNET_PEERSTORE_iterate (peerstore,
                                "transport",
                                &b->pid,
                                GNUNET_PEERSTORE_TRANSPORT_BACKCHANNEL_MONOTIME,
                                &backtalker_monotime_cb,
                                b);
  } /* end actual decryption */
}


static int
check_incoming_msg (void *cls,
                    const struct GNUNET_TRANSPORT_IncomingMessage *im)
{
  struct TransportClient *tc = cls;

  if (CT_COMMUNICATOR != tc->type)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  GNUNET_MQ_check_boxed_message (im);
  return GNUNET_OK;
}


struct CheckKnownAddressContext
{
  const char *address;

  struct ValidationState *vs;
};


static int
check_known_address (void *cls,
                     const struct GNUNET_PeerIdentity *pid,
                     void *value)
{
  struct CheckKnownAddressContext *ckac = cls;
  struct ValidationState *vs = value;

  (void) pid;
  if (0 != strcmp (vs->address, ckac->address))
    return GNUNET_OK;
  ckac->vs = vs;
  return GNUNET_NO;
}


static void
validation_start_cb (void *cls);


static void
update_next_challenge_time (struct ValidationState *vs,
                            struct GNUNET_TIME_Absolute new_time)
{
  struct GNUNET_TIME_Relative delta;

  if (new_time.abs_value_us == vs->next_challenge.abs_value_us)
    return; /* be lazy */
  vs->next_challenge = new_time;
  if (NULL == vs->hn)
    vs->hn =
      GNUNET_CONTAINER_heap_insert (validation_heap, vs, new_time.abs_value_us);
  else
    GNUNET_CONTAINER_heap_update_cost (vs->hn, new_time.abs_value_us);
  if ((vs != GNUNET_CONTAINER_heap_peek (validation_heap)) &&
      (NULL != validation_task))
    return;
  if (NULL != validation_task)
    GNUNET_SCHEDULER_cancel (validation_task);
  /* randomize a bit */
  delta.rel_value_us =
    GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
                              MIN_DELAY_ADDRESS_VALIDATION.rel_value_us);
  new_time = GNUNET_TIME_absolute_add (new_time, delta);
  validation_task =
    GNUNET_SCHEDULER_add_at (new_time, &validation_start_cb, NULL);
}


static void
start_address_validation (const struct GNUNET_PeerIdentity *pid,
                          const char *address)
{
  struct GNUNET_TIME_Absolute now;
  struct ValidationState *vs;
  struct CheckKnownAddressContext ckac = { .address = address, .vs = NULL };

  (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
                                                     pid,
                                                     &check_known_address,
                                                     &ckac);
  if (NULL != (vs = ckac.vs))
  {
    /* if 'vs' is not currently valid, we need to speed up retrying the
     * validation */
    if (vs->validated_until.abs_value_us < vs->next_challenge.abs_value_us)
    {
      /* reduce backoff as we got a fresh advertisement */
      vs->challenge_backoff =
        GNUNET_TIME_relative_min (FAST_VALIDATION_CHALLENGE_FREQ,
                                  GNUNET_TIME_relative_divide (
                                    vs->challenge_backoff,
                                    2));
      update_next_challenge_time (vs,
                                  GNUNET_TIME_relative_to_absolute (
                                    vs->challenge_backoff));
    }
    return;
  }
  now = GNUNET_TIME_absolute_get ();
  vs = GNUNET_new (struct ValidationState);
  vs->pid = *pid;
  vs->valid_until =
    GNUNET_TIME_relative_to_absolute (ADDRESS_VALIDATION_LIFETIME);
  vs->first_challenge_use = now;
  vs->validation_rtt = GNUNET_TIME_UNIT_FOREVER_REL;
  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
                              &vs->challenge,
                              sizeof(vs->challenge));
  vs->address = GNUNET_strdup (address);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting address validation `%s' of peer %s using challenge %s\n",
              address,
              GNUNET_i2s (pid),
              GNUNET_sh2s (&vs->challenge.value));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_put (
                   validation_map,
                   &vs->pid,
                   vs,
                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  update_next_challenge_time (vs, now);
}


static void
handle_hello_for_incoming (void *cls,
                           const struct GNUNET_PEERSTORE_Record *record,
                           const char *emsg)
{
  struct IncomingRequest *ir = cls;
  const char *val;

  if (NULL != emsg)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Got failure from PEERSTORE: %s\n",
                emsg);
    return;
  }
  val = record->value;
  if ((0 == record->value_size) || ('\0' != val[record->value_size - 1]))
  {
    GNUNET_break (0);
    return;
  }
  start_address_validation (&ir->pid, (const char *) record->value);
}


static void
handle_validation_challenge (
  void *cls,
  const struct TransportValidationChallengeMessage *tvc)
{
  struct CommunicatorMessageContext *cmc = cls;
  struct TransportValidationResponseMessage tvr;
  struct VirtualLink *vl;
  struct GNUNET_TIME_RelativeNBO validity_duration;
  struct IncomingRequest *ir;
  struct Neighbour *n;
  struct GNUNET_PeerIdentity sender;

  /* DV-routed messages are not allowed for validation challenges */
  if (cmc->total_hops > 0)
  {
    GNUNET_break_op (0);
    finish_cmc_handling (cmc);
    return;
  }
  validity_duration = cmc->im.expected_address_validity;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received address validation challenge %s\n",
              GNUNET_sh2s (&tvc->challenge.value));
  /* If we have a virtual link, we use this mechanism to signal the
     size of the flow control window, and to allow the sender
     to ask for increases. If for us the virtual link is still down,
     we will always give a window size of zero. */
  tvr.header.type =
    htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE);
  tvr.header.size = htons (sizeof(tvr));
  tvr.reserved = htonl (0);
  tvr.challenge = tvc->challenge;
  tvr.origin_time = tvc->sender_time;
  tvr.validity_duration = validity_duration;
  {
    /* create signature */
    struct TransportValidationPS tvp =
    { .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE),
      .purpose.size = htonl (sizeof(tvp)),
      .validity_duration = validity_duration,
      .challenge = tvc->challenge };

    GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
                                                          &tvp.purpose,
                                                          &tvr.signature));
  }
  route_control_message_without_fc (&cmc->im.sender,
                                    &tvr.header,
                                    RMO_ANYTHING_GOES | RMO_REDUNDANT);
  sender = cmc->im.sender;
  finish_cmc_handling (cmc);
  vl = lookup_virtual_link (&sender);
  if (NULL != vl)
    return;

  /* For us, the link is still down, but we need bi-directional
     connections (for flow-control and for this to be useful for
     CORE), so we must try to bring the link up! */

  /* (1) Check existing queues, if any, we may be lucky! */
  n = lookup_neighbour (&sender);
  if (NULL != n)
    for (struct Queue *q = n->queue_head; NULL != q; q = q->next_neighbour)
      start_address_validation (&sender, q->address);
  /* (2) Also try to see if we have addresses in PEERSTORE for this peer
     we could use */
  for (ir = ir_head; NULL != ir; ir = ir->next)
    if (0 == GNUNET_memcmp (&ir->pid, &sender))
      return;
  /* we are already trying */
  ir = GNUNET_new (struct IncomingRequest);
  ir->pid = sender;
  GNUNET_CONTAINER_DLL_insert (ir_head, ir_tail, ir);
  ir->wc = GNUNET_PEERSTORE_watch (peerstore,
                                   "transport",
                                   &ir->pid,
                                   GNUNET_PEERSTORE_TRANSPORT_URLADDRESS_KEY,
                                   &handle_hello_for_incoming,
                                   ir);
  ir_total++;
  /* Bound attempts we do in parallel here, might otherwise get excessive */
  while (ir_total > MAX_INCOMING_REQUEST)
    free_incoming_request (ir_head);
}


struct CheckKnownChallengeContext
{
  const struct ChallengeNonceP *challenge;

  struct ValidationState *vs;
};


static int
check_known_challenge (void *cls,
                       const struct GNUNET_PeerIdentity *pid,
                       void *value)
{
  struct CheckKnownChallengeContext *ckac = cls;
  struct ValidationState *vs = value;

  (void) pid;
  if (0 != GNUNET_memcmp (&vs->challenge, ckac->challenge))
    return GNUNET_OK;
  ckac->vs = vs;
  return GNUNET_NO;
}


static void
peerstore_store_validation_cb (void *cls, int success)
{
  struct ValidationState *vs = cls;

  vs->sc = NULL;
  if (GNUNET_YES == success)
    return;
  GNUNET_STATISTICS_update (GST_stats,
                            "# Peerstore failed to store foreign address",
                            1,
                            GNUNET_NO);
}


static struct Queue *
find_queue (const struct GNUNET_PeerIdentity *pid, const char *address)
{
  struct Neighbour *n;

  n = lookup_neighbour (pid);
  if (NULL == n)
    return NULL;
  for (struct Queue *pos = n->queue_head; NULL != pos;
       pos = pos->next_neighbour)
  {
    if (0 == strcmp (pos->address, address))
      return pos;
  }
  return NULL;
}


static void
handle_validation_response (
  void *cls,
  const struct TransportValidationResponseMessage *tvr)
{
  struct CommunicatorMessageContext *cmc = cls;
  struct ValidationState *vs;
  struct CheckKnownChallengeContext ckac = { .challenge = &tvr->challenge,
                                             .vs = NULL };
  struct GNUNET_TIME_Absolute origin_time;