TNG: Add first steps for communicator testing
[oweals/gnunet.git] / src / transport / gnunet-service-tng.c
1 /*
2  This file is part of GNUnet.
3  Copyright (C) 2010-2016, 2018, 2019 GNUnet e.V.
4
5  GNUnet is free software: you can redistribute it and/or modify it
6  under the terms of the GNU Affero General Public License as published
7  by the Free Software Foundation, either version 3 of the License,
8  or (at your option) any later version.
9
10  GNUnet is distributed in the hope that it will be useful, but
11  WITHOUT ANY WARRANTY; without even the implied warranty of
12  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  Affero General Public License for more details.
14
15  You should have received a copy of the GNU Affero General Public License
16  along with this program.  If not, see <http://www.gnu.org/licenses/>.
17
18      SPDX-License-Identifier: AGPL3.0-or-later
19  */
20 /**
21  * @file transport/gnunet-service-tng.c
22  * @brief main for gnunet-service-tng
23  * @author Christian Grothoff
24  *
25  * TODO:
26  * - figure out how to transmit (selective) ACKs in case of uni-directional
27  *   communicators (with/without core? DV-only?) When do we use ACKs?
28  *   => communicators use selective ACKs for flow control
29  *   => transport uses message-level ACKs for RTT, fragment confirmation
30  *   => integrate DV into transport, use neither core nor communicators
31  *      but rather give communicators transport-encapsulated messages
32  *      (which could be core-data, background-channel traffic, or
33  *       transport-to-transport traffic)
34  *
35  * Implement next:
36  * - route_message() implementation, including using DV data structures
37  *   (but not when routing certain message types, like DV learn,
38  *    looks like now like we need two flags (DV/no-DV, confirmed-only,
39  *    unconfirmed OK)
40  *   + NOTE: do NOT use PendingMessage for route_message(), as that is
41  *     for fragmentation/reliability and ultimately core flow control!
42  *     => route_message() should pick the queue
43  *     => in case of DV routing, route_message should BOX the message, too.
44  * - We currently do NEVER tell CORE also about DV-connections (core_visible
45  *    of `struct DistanceVector` is simply never set!)
46  *     + When? Easy if we initiated the DV and got the challenge; do that NOW
47  *        BUT what we passively learned DV (unconfirmed freshness)
48  *        => Do we trigger Challenge->Response there as well, or 'wait' for
49  *           our own DV initiations to discover?
50  *        => What about DV routes that expire? Do we also only count on
51  *           our own DV initiations for maintenance here, or do we
52  *           try to specifically re-confirm the existence of a particular path?
53  *        => OPITMIZATION-FIXME!
54  *   + Where do we track what we told core? Careful: need to check
55  *     the "core_visible' flag in both neighbours and DV before
56  *     sending out notifications to CORE!
57  * - retransmission logic
58  * - track RTT, distance, loss, etc. => requires extra data structures!
59  *
60  * Later:
61  * - change transport-core API to provide proper flow control in both
62  *   directions, allow multiple messages per peer simultaneously (tag
63  *   confirmations with unique message ID), and replace quota-out with
64  *   proper flow control;
65  * - if messages are below MTU, consider adding ACKs and other stuff
66  *   (requires planning at receiver, and additional MST-style demultiplex
67  *    at receiver!)
68  * - could avoid copying body of message into each fragment and keep
69  *   fragments as just pointers into the original message and only
70  *   fully build fragments just before transmission (optimization, should
71  *   reduce CPU and memory use)
72  *
73  * FIXME (without marks in the code!):
74  * - proper use/initialization of timestamps in messages exchanged
75  *   during DV learning
76  * - persistence of monotonic time obtained from other peers
77  *   in PEERSTORE (by message type)
78  *
79  * Optimizations:
80  * - use shorthashmap on msg_uuid's when matching reliability/fragment ACKs
81  *   against our pending message queue (requires additional per neighbour
82  *   hash map to be maintained, avoids possible linear scan on pending msgs)
83  * - queue_send_msg and route_message both by API design have to make copies
84  *   of the payload, and route_message on top of that requires a malloc/free.
85  *   Change design to approximate "zero" copy better...
86  *
87  * Design realizations / discussion:
88  * - communicators do flow control by calling MQ "notify sent"
89  *   when 'ready'. They determine flow implicitly (i.e. TCP blocking)
90  *   or explicitly via backchannel FC ACKs.  As long as the
91  *   channel is not full, they may 'notify sent' even if the other
92  *   peer has not yet confirmed receipt. The other peer confirming
93  *   is _only_ for FC, not for more reliable transmission; reliable
94  *   transmission (i.e. of fragments) is left to _transport_.
95  * - ACKs sent back in uni-directional communicators are done via
96  *   the background channel API; here transport _may_ initially
97  *   broadcast (with bounded # hops) if no path is known;
98  * - transport should _integrate_ DV-routing and build a view of
99  *   the network; then background channel traffic can be
100  *   routed via DV as well as explicit "DV" traffic.
101  * - background channel is also used for ACKs and NAT traversal support
102  * - transport service is responsible for AEAD'ing the background
103  *   channel, timestamps and monotonic time are used against replay
104  *   of old messages -> peerstore needs to be supplied with
105  *   "latest timestamps seen" data
106  * - if transport implements DV, we likely need a 3rd peermap
107  *   in addition to ephemerals and (direct) neighbours
108  *   ==> check if stuff needs to be moved out of "Neighbour"
109  * - transport should encapsualte core-level messages and do its
110  *   own ACKing for RTT/goodput/loss measurements _and_ fragment
111  *   for retransmission
112  */
113 #include "platform.h"
114 #include "gnunet_util_lib.h"
115 #include "gnunet_statistics_service.h"
116 #include "gnunet_transport_monitor_service.h"
117 #include "gnunet_peerstore_service.h"
118 #include "gnunet_hello_lib.h"
119 #include "gnunet_signatures.h"
120 #include "transport.h"
121
122
123 /**
124  * What is the size we assume for a read operation in the
125  * absence of an MTU for the purpose of flow control?
126  */
127 #define IN_PACKET_SIZE_WITHOUT_MTU 128
128
129 /**
130  * Minimum number of hops we should forward DV learn messages
131  * even if they are NOT useful for us in hope of looping
132  * back to the initiator?
133  *
134  * FIXME: allow initiator some control here instead?
135  */
136 #define MIN_DV_PATH_LENGTH_FOR_INITIATOR 3
137
138 /**
139  * Maximum DV distance allowed ever.
140  */
141 #define MAX_DV_HOPS_ALLOWED 16
142
143 /**
144  * Maximum number of DV learning activities we may
145  * have pending at the same time.
146  */
147 #define MAX_DV_LEARN_PENDING 64
148
149 /**
150  * Maximum number of DV paths we keep simultaneously to the same target.
151  */
152 #define MAX_DV_PATHS_TO_TARGET 3
153
154 /**
155  * If a queue delays the next message by more than this number
156  * of seconds we log a warning. Note: this is for testing,
157  * the value chosen here might be too aggressively low!
158  */
159 #define DELAY_WARN_THRESHOLD GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
160
161 /**
162  * We only consider queues as "quality" connections when
163  * suppressing the generation of DV initiation messages if
164  * the latency of the queue is below this threshold.
165  */
166 #define DV_QUALITY_RTT_THRESHOLD GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
167
168 /**
169  * How long do we consider a DV path valid if we see no
170  * further updates on it? Note: the value chosen here might be too low!
171  */
172 #define DV_PATH_VALIDITY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
173
174 /**
175  * How long before paths expire would we like to (re)discover DV paths? Should
176  * be below #DV_PATH_VALIDITY_TIMEOUT.
177  */
178 #define DV_PATH_DISCOVERY_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
179
180 /**
181  * How long are ephemeral keys valid?
182  */
183 #define EPHEMERAL_VALIDITY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
184
185 /**
186  * How long do we keep partially reassembled messages around before giving up?
187  */
188 #define REASSEMBLY_EXPIRATION GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
189
190 /**
191  * What is the fastest rate at which we send challenges *if* we keep learning
192  * an address (gossip, DHT, etc.)?
193  */
194 #define FAST_VALIDATION_CHALLENGE_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1)
195
196 /**
197  * What is the slowest rate at which we send challenges?
198  */
199 #define MAX_VALIDATION_CHALLENGE_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_DAYS, 1)
200
201 /**
202  * What is the non-randomized base frequency at which we
203  * would initiate DV learn messages?
204  */
205 #define DV_LEARN_BASE_FREQUENCY GNUNET_TIME_UNIT_MINUTES
206
207 /**
208  * How many good connections (confirmed, bi-directional, not DV)
209  * do we need to have to suppress initiating DV learn messages?
210  */
211 #define DV_LEARN_QUALITY_THRESHOLD 100
212
213 /**
214  * When do we forget an invalid address for sure?
215  */
216 #define MAX_ADDRESS_VALID_UNTIL GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MONTHS, 1)
217 /**
218  * How long do we consider an address valid if we just checked?
219  */
220 #define ADDRESS_VALIDATION_LIFETIME GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
221
222 /**
223  * What is the maximum frequency at which we do address validation?
224  * A random value between 0 and this value is added when scheduling
225  * the #validation_task (both to ensure we do not validate too often,
226  * and to randomize a bit).
227  */
228 #define MIN_DELAY_ADDRESS_VALIDATION GNUNET_TIME_UNIT_MILLISECONDS
229
230 /**
231  * How many network RTTs before an address validation expires should we begin
232  * trying to revalidate? (Note that the RTT used here is the one that we
233  * experienced during the last validation, not necessarily the latest RTT
234  * observed).
235  */
236 #define VALIDATION_RTT_BUFFER_FACTOR 3
237
238 /**
239  * How many messages can we have pending for a given communicator
240  * process before we start to throttle that communicator?
241  *
242  * Used if a communicator might be CPU-bound and cannot handle the traffic.
243  */
244 #define COMMUNICATOR_TOTAL_QUEUE_LIMIT 512
245
246 /**
247  * How many messages can we have pending for a given queue (queue to
248  * a particular peer via a communicator) process before we start to
249  * throttle that queue?
250  */
251 #define QUEUE_LENGTH_LIMIT 32
252
253
254 GNUNET_NETWORK_STRUCT_BEGIN
255
256 /**
257  * Outer layer of an encapsulated backchannel message.
258  */
259 struct TransportBackchannelEncapsulationMessage
260 {
261   /**
262    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION.
263    */
264   struct GNUNET_MessageHeader header;
265
266   /**
267    * Distance the backchannel message has traveled, to be updated at
268    * each hop.  Used to bound the number of hops in case a backchannel
269    * message is broadcast and thus travels without routing
270    * information (during initial backchannel discovery).
271    */
272   uint32_t distance;
273
274   /**
275    * Target's peer identity (as backchannels may be transmitted
276    * indirectly, or even be broadcast).
277    */
278   struct GNUNET_PeerIdentity target;
279
280   /**
281    * Ephemeral key setup by the sender for @e target, used
282    * to encrypt the payload.
283    */
284   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
285
286   /**
287    * We use an IV here as the @e ephemeral_key is re-used for
288    * #EPHEMERAL_VALIDITY time to avoid re-signing it all the time.
289    */
290   struct GNUNET_ShortHashCode iv;
291
292   /**
293    * HMAC over the ciphertext of the encrypted, variable-size
294    * body that follows.  Verified via DH of @e target and
295    * @e ephemeral_key
296    */
297   struct GNUNET_HashCode hmac;
298
299   /* Followed by encrypted, variable-size payload */
300 };
301
302
303 /**
304  * Body by which a peer confirms that it is using an ephemeral key.
305  */
306 struct EphemeralConfirmation
307 {
308
309   /**
310    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
311    */
312   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
313
314   /**
315    * How long is this signature over the ephemeral key valid?
316    *
317    * Note that the receiver MUST IGNORE the absolute time, and only interpret
318    * the value as a mononic time and reject "older" values than the last one
319    * observed.  This is necessary as we do not want to require synchronized
320    * clocks and may not have a bidirectional communication channel.
321    *
322    * Even with this, there is no real guarantee against replay achieved here,
323    * unless the latest timestamp is persisted.  While persistence should be
324    * provided via PEERSTORE, we do not consider the mechanism reliable!  Thus,
325    * communicators must protect against replay attacks when using backchannel
326    * communication!
327    */
328   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
329
330   /**
331    * Target's peer identity.
332    */
333   struct GNUNET_PeerIdentity target;
334
335   /**
336    * Ephemeral key setup by the sender for @e target, used
337    * to encrypt the payload.
338    */
339   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
340
341 };
342
343
344 /**
345  * Plaintext of the variable-size payload that is encrypted
346  * within a `struct TransportBackchannelEncapsulationMessage`
347  */
348 struct TransportBackchannelRequestPayload
349 {
350
351   /**
352    * Sender's peer identity.
353    */
354   struct GNUNET_PeerIdentity sender;
355
356   /**
357    * Signature of the sender over an
358    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL.
359    */
360   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
361
362   /**
363    * How long is this signature over the ephemeral key valid?
364    *
365    * Note that the receiver MUST IGNORE the absolute time, and only interpret
366    * the value as a mononic time and reject "older" values than the last one
367    * observed.  This is necessary as we do not want to require synchronized
368    * clocks and may not have a bidirectional communication channel.
369    *
370    * Even with this, there is no real guarantee against replay achieved here,
371    * unless the latest timestamp is persisted.  While persistence should be
372    * provided via PEERSTORE, we do not consider the mechanism reliable!  Thus,
373    * communicators must protect against replay attacks when using backchannel
374    * communication!
375    */
376   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
377
378   /**
379    * Current monotonic time of the sending transport service.  Used to
380    * detect replayed messages.  Note that the receiver should remember
381    * a list of the recently seen timestamps and only reject messages
382    * if the timestamp is in the list, or the list is "full" and the
383    * timestamp is smaller than the lowest in the list.
384    *
385    * Like the @e ephemeral_validity, the list of timestamps per peer should be
386    * persisted to guard against replays after restarts.
387    */
388   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
389
390   /* Followed by a `struct GNUNET_MessageHeader` with a message
391      for a communicator */
392
393   /* Followed by a 0-termianted string specifying the name of
394      the communicator which is to receive the message */
395
396 };
397
398
399 /**
400  * Outer layer of an encapsulated unfragmented application message sent
401  * over an unreliable channel.
402  */
403 struct TransportReliabilityBox
404 {
405   /**
406    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX
407    */
408   struct GNUNET_MessageHeader header;
409
410   /**
411    * Number of messages still to be sent before a commulative
412    * ACK is requested.  Zero if an ACK is requested immediately.
413    * In NBO.  Note that the receiver may send the ACK faster
414    * if it believes that is reasonable.
415    */
416   uint32_t ack_countdown GNUNET_PACKED;
417
418   /**
419    * Unique ID of the message used for signalling receipt of
420    * messages sent over possibly unreliable channels.  Should
421    * be a random.
422    */
423   struct GNUNET_ShortHashCode msg_uuid;
424 };
425
426
427 /**
428  * Confirmation that the receiver got a
429  * #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX. Note that the
430  * confirmation may be transmitted over a completely different queue,
431  * so ACKs are identified by a combination of PID of sender and
432  * message UUID, without the queue playing any role!
433  */
434 struct TransportReliabilityAckMessage
435 {
436   /**
437    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK
438    */
439   struct GNUNET_MessageHeader header;
440
441   /**
442    * Reserved. Zero.
443    */
444   uint32_t reserved GNUNET_PACKED;
445
446   /**
447    * How long was the ACK delayed relative to the average time of
448    * receipt of the messages being acknowledged?  Used to calculate
449    * the average RTT by taking the receipt time of the ack minus the
450    * average transmission time of the sender minus this value.
451    */
452   struct GNUNET_TIME_RelativeNBO avg_ack_delay;
453
454   /* followed by any number of `struct GNUNET_ShortHashCode`
455      messages providing ACKs */
456 };
457
458
459 /**
460  * Outer layer of an encapsulated fragmented application message.
461  */
462 struct TransportFragmentBox
463 {
464   /**
465    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT
466    */
467   struct GNUNET_MessageHeader header;
468
469   /**
470    * Unique ID of this fragment (and fragment transmission!). Will
471    * change even if a fragement is retransmitted to make each
472    * transmission attempt unique! Should be incremented by one for
473    * each fragment transmission. If a client receives a duplicate
474    * fragment (same @e frag_off), it must send
475    * #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK immediately.
476    */
477   uint32_t frag_uuid GNUNET_PACKED;
478
479   /**
480    * Original message ID for of the message that all the1
481    * fragments belong to.  Must be the same for all fragments.
482    */
483   struct GNUNET_ShortHashCode msg_uuid;
484
485   /**
486    * Offset of this fragment in the overall message.
487    */
488   uint16_t frag_off GNUNET_PACKED;
489
490   /**
491    * Total size of the message that is being fragmented.
492    */
493   uint16_t msg_size GNUNET_PACKED;
494
495 };
496
497
498 /**
499  * Outer layer of an fragmented application message sent over a queue
500  * with finite MTU.  When a #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT is
501  * received, the receiver has two RTTs or 64 further fragments with
502  * the same basic message time to send an acknowledgement, possibly
503  * acknowledging up to 65 fragments in one ACK.  ACKs must also be
504  * sent immediately once all fragments were sent.
505  */
506 struct TransportFragmentAckMessage
507 {
508   /**
509    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK
510    */
511   struct GNUNET_MessageHeader header;
512
513   /**
514    * Unique ID of the lowest fragment UUID being acknowledged.
515    */
516   uint32_t frag_uuid GNUNET_PACKED;
517
518   /**
519    * Bitfield of up to 64 additional fragments following the
520    * @e msg_uuid being acknowledged by this message.
521    */
522   uint64_t extra_acks GNUNET_PACKED;
523
524   /**
525    * Original message ID for of the message that all the
526    * fragments belong to.
527    */
528   struct GNUNET_ShortHashCode msg_uuid;
529
530   /**
531    * How long was the ACK delayed relative to the average time of
532    * receipt of the fragments being acknowledged?  Used to calculate
533    * the average RTT by taking the receipt time of the ack minus the
534    * average transmission time of the sender minus this value.
535    */
536   struct GNUNET_TIME_RelativeNBO avg_ack_delay;
537
538   /**
539    * How long until the receiver will stop trying reassembly
540    * of this message?
541    */
542   struct GNUNET_TIME_RelativeNBO reassembly_timeout;
543 };
544
545
546 /**
547  * Content signed by the initator during DV learning.
548  *
549  * The signature is required to prevent DDoS attacks. A peer sending out this
550  * message is potentially generating a lot of traffic that will go back to the
551  * initator, as peers receiving this message will try to let the initiator
552  * know that they got the message.
553  *
554  * Without this signature, an attacker could abuse this mechanism for traffic
555  * amplification, sending a lot of traffic to a peer by putting out this type
556  * of message with the victim's peer identity.
557  *
558  * Even with just a signature, traffic amplification would be possible via
559  * replay attacks. The @e monotonic_time limits such replay attacks, as every
560  * potential amplificator will check the @e monotonic_time and only respond
561  * (at most) once per message.
562  */
563 struct DvInitPS
564 {
565   /**
566    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
567    */
568   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
569
570   /**
571    * Time at the initiator when generating the signature.
572    *
573    * Note that the receiver MUST IGNORE the absolute time, and only interpret
574    * the value as a mononic time and reject "older" values than the last one
575    * observed.  This is necessary as we do not want to require synchronized
576    * clocks and may not have a bidirectional communication channel.
577    *
578    * Even with this, there is no real guarantee against replay achieved here,
579    * unless the latest timestamp is persisted.  Persistence should be
580    * provided via PEERSTORE if possible.
581    */
582   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
583
584   /**
585    * Challenge value used by the initiator to re-identify the path.
586    */
587   struct GNUNET_ShortHashCode challenge;
588
589 };
590
591
592 /**
593  * Content signed by each peer during DV learning.
594  *
595  * This assues the initiator of the DV learning operation that the hop from @e
596  * pred via the signing peer to @e succ actually exists.  This makes it
597  * impossible for an adversary to supply the network with bogus routes.
598  *
599  * The @e challenge is included to provide replay protection for the
600  * initiator. This way, the initiator knows that the hop existed after the
601  * original @e challenge was first transmitted, providing a freshness metric.
602  *
603  * Peers other than the initiator that passively learn paths by observing
604  * these messages do NOT benefit from this. Here, an adversary may indeed
605  * replay old messages.  Thus, passively learned paths should always be
606  * immediately marked as "potentially stale".
607  */
608 struct DvHopPS
609 {
610   /**
611    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP
612    */
613   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
614
615   /**
616    * Identity of the previous peer on the path.
617    */
618   struct GNUNET_PeerIdentity pred;
619
620   /**
621    * Identity of the next peer on the path.
622    */
623   struct GNUNET_PeerIdentity succ;
624
625   /**
626    * Challenge value used by the initiator to re-identify the path.
627    */
628   struct GNUNET_ShortHashCode challenge;
629
630 };
631
632
633 /**
634  * An entry describing a peer on a path in a
635  * `struct TransportDVLearn` message.
636  */
637 struct DVPathEntryP
638 {
639   /**
640    * Identity of a peer on the path.
641    */
642   struct GNUNET_PeerIdentity hop;
643
644   /**
645    * Signature of this hop over the path, of purpose
646    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP
647    */
648   struct GNUNET_CRYPTO_EddsaSignature hop_sig;
649
650 };
651
652
653 /**
654  * Internal message used by transport for distance vector learning.
655  * If @e num_hops does not exceed the threshold, peers should append
656  * themselves to the peer list and flood the message (possibly only
657  * to a subset of their neighbours to limit discoverability of the
658  * network topology).  To the extend that the @e bidirectional bits
659  * are set, peers may learn the inverse paths even if they did not
660  * initiate.
661  *
662  * Unless received on a bidirectional queue and @e num_hops just
663  * zero, peers that can forward to the initator should always try to
664  * forward to the initiator.
665  */
666 struct TransportDVLearn
667 {
668   /**
669    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN
670    */
671   struct GNUNET_MessageHeader header;
672
673   /**
674    * Number of hops this messages has travelled, in NBO. Zero if
675    * sent by initiator.
676    */
677   uint16_t num_hops GNUNET_PACKED;
678
679   /**
680    * Bitmask of the last 16 hops indicating whether they are confirmed
681    * available (without DV) in both directions or not, in NBO.  Used
682    * to possibly instantly learn a path in both directions.  Each peer
683    * should shift this value by one to the left, and then set the
684    * lowest bit IF the current sender can be reached from it (without
685    * DV routing).
686    */
687   uint16_t bidirectional GNUNET_PACKED;
688
689   /**
690    * Peers receiving this message and delaying forwarding to other
691    * peers for any reason should increment this value by the non-network
692    * delay created by the peer.
693    */
694   struct GNUNET_TIME_RelativeNBO non_network_delay;
695
696   /**
697    * Signature of this hop over the path, of purpose
698    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
699    */
700   struct GNUNET_CRYPTO_EddsaSignature init_sig;
701
702   /**
703    * Identity of the peer that started this learning activity.
704    */
705   struct GNUNET_PeerIdentity initiator;
706
707   /**
708    * Challenge value used by the initiator to re-identify the path.
709    */
710   struct GNUNET_ShortHashCode challenge;
711
712   /* Followed by @e num_hops `struct DVPathEntryP` values,
713      excluding the initiator of the DV trace; the last entry is the
714      current sender; the current peer must not be included. */
715
716 };
717
718
719 /**
720  * Outer layer of an encapsulated message send over multiple hops.
721  * The path given only includes the identities of the subsequent
722  * peers, i.e. it will be empty if we are the receiver. Each
723  * forwarding peer should scan the list from the end, and if it can,
724  * forward to the respective peer. The list should then be shortened
725  * by all the entries up to and including that peer.  Each hop should
726  * also increment @e total_hops to allow the receiver to get a precise
727  * estimate on the number of hops the message travelled.  Senders must
728  * provide a learned path that thus should work, but intermediaries
729  * know of a shortcut, they are allowed to send the message via that
730  * shortcut.
731  *
732  * If a peer finds itself still on the list, it must drop the message.
733  */
734 struct TransportDVBox
735 {
736   /**
737    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX
738    */
739   struct GNUNET_MessageHeader header;
740
741   /**
742    * Number of total hops this messages travelled. In NBO.
743    * @e origin sets this to zero, to be incremented at
744    * each hop.
745    */
746   uint16_t total_hops GNUNET_PACKED;
747
748   /**
749    * Number of hops this messages includes. In NBO.
750    */
751   uint16_t num_hops GNUNET_PACKED;
752
753   /**
754    * Identity of the peer that originated the message.
755    */
756   struct GNUNET_PeerIdentity origin;
757
758   /* Followed by @e num_hops `struct GNUNET_PeerIdentity` values;
759      excluding the @e origin and the current peer, the last must be
760      the ultimate target; if @e num_hops is zero, the receiver of this
761      message is the ultimate target. */
762
763   /* Followed by the actual message, which itself may be
764      another box, but not a DV_LEARN or DV_BOX message! */
765 };
766
767
768 /**
769  * Message send to another peer to validate that it can indeed
770  * receive messages at a particular address.
771  */
772 struct TransportValidationChallenge
773 {
774
775   /**
776    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE
777    */
778   struct GNUNET_MessageHeader header;
779
780   /**
781    * Zero.
782    */
783   uint32_t reserved GNUNET_PACKED;
784
785   /**
786    * Challenge to be signed by the receiving peer.
787    */
788   struct GNUNET_ShortHashCode challenge;
789
790   /**
791    * Timestamp of the sender, to be copied into the reply
792    * to allow sender to calculate RTT.
793    */
794   struct GNUNET_TIME_AbsoluteNBO sender_time;
795 };
796
797
798 /**
799  * Message signed by a peer to confirm that it can indeed
800  * receive messages at a particular address.
801  */
802 struct TransportValidationPS
803 {
804
805   /**
806    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE
807    */
808   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
809
810   /**
811    * How long does the sender believe the address on
812    * which the challenge was received to remain valid?
813    */
814   struct GNUNET_TIME_RelativeNBO validity_duration;
815
816   /**
817    * Challenge signed by the receiving peer.
818    */
819   struct GNUNET_ShortHashCode challenge;
820
821 };
822
823
824 /**
825  * Message send to a peer to respond to a
826  * #GNUNET_MESSAGE_TYPE_ADDRESS_VALIDATION_CHALLENGE
827  */
828 struct TransportValidationResponse
829 {
830
831   /**
832    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE
833    */
834   struct GNUNET_MessageHeader header;
835
836   /**
837    * Zero.
838    */
839   uint32_t reserved GNUNET_PACKED;
840
841   /**
842    * The peer's signature matching the
843    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE purpose.
844    */
845   struct GNUNET_CRYPTO_EddsaSignature signature;
846
847   /**
848    * The challenge that was signed by the receiving peer.
849    */
850   struct GNUNET_ShortHashCode challenge;
851
852   /**
853    * Original timestamp of the sender (was @code{sender_time}),
854    * copied into the reply to allow sender to calculate RTT.
855    */
856   struct GNUNET_TIME_AbsoluteNBO origin_time;
857
858   /**
859    * How long does the sender believe this address to remain
860    * valid?
861    */
862   struct GNUNET_TIME_RelativeNBO validity_duration;
863 };
864
865
866
867 GNUNET_NETWORK_STRUCT_END
868
869
870 /**
871  * What type of client is the `struct TransportClient` about?
872  */
873 enum ClientType
874 {
875   /**
876    * We do not know yet (client is fresh).
877    */
878   CT_NONE = 0,
879
880   /**
881    * Is the CORE service, we need to forward traffic to it.
882    */
883   CT_CORE = 1,
884
885   /**
886    * It is a monitor, forward monitor data.
887    */
888   CT_MONITOR = 2,
889
890   /**
891    * It is a communicator, use for communication.
892    */
893   CT_COMMUNICATOR = 3,
894
895   /**
896    * "Application" telling us where to connect (i.e. TOPOLOGY, DHT or CADET).
897    */
898   CT_APPLICATION = 4
899 };
900
901
902 /**
903  * When did we launch this DV learning activity?
904  */
905 struct LearnLaunchEntry
906 {
907
908   /**
909    * Kept (also) in a DLL sorted by launch time.
910    */
911   struct LearnLaunchEntry *prev;
912
913   /**
914    * Kept (also) in a DLL sorted by launch time.
915    */
916   struct LearnLaunchEntry *next;
917
918   /**
919    * Challenge that uniquely identifies this activity.
920    */
921   struct GNUNET_ShortHashCode challenge;
922
923   /**
924    * When did we transmit the DV learn message (used to calculate RTT) and
925    * determine freshness of paths learned via this operation.
926    */
927   struct GNUNET_TIME_Absolute launch_time;
928
929 };
930
931
932 /**
933  * Entry in our cache of ephemeral keys we currently use.  This way, we only
934  * sign an ephemeral once per @e target, and then can re-use it over multiple
935  * #GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION messages (as
936  * signing is expensive and in some cases we may use backchannel messages a
937  * lot).
938  */
939 struct EphemeralCacheEntry
940 {
941
942   /**
943    * Target's peer identity (we don't re-use ephemerals
944    * to limit linkability of messages).
945    */
946   struct GNUNET_PeerIdentity target;
947
948   /**
949    * Signature affirming @e ephemeral_key of type
950    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
951    */
952   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
953
954   /**
955    * How long is @e sender_sig valid
956    */
957   struct GNUNET_TIME_Absolute ephemeral_validity;
958
959   /**
960    * Our ephemeral key.
961    */
962   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
963
964   /**
965    * Our private ephemeral key.
966    */
967   struct GNUNET_CRYPTO_EcdhePrivateKey private_key;
968
969   /**
970    * Node in the ephemeral cache for this entry.
971    * Used for expiration.
972    */
973   struct GNUNET_CONTAINER_HeapNode *hn;
974 };
975
976
977 /**
978  * Client connected to the transport service.
979  */
980 struct TransportClient;
981
982
983 /**
984  * A neighbour that at least one communicator is connected to.
985  */
986 struct Neighbour;
987
988
989 /**
990  * Entry in our #dv_routes table, representing a (set of) distance
991  * vector routes to a particular peer.
992  */
993 struct DistanceVector;
994
995 /**
996  * One possible hop towards a DV target.
997  */
998 struct DistanceVectorHop
999 {
1000
1001   /**
1002    * Kept in a MDLL, sorted by @e timeout.
1003    */
1004   struct DistanceVectorHop *next_dv;
1005
1006   /**
1007    * Kept in a MDLL, sorted by @e timeout.
1008    */
1009   struct DistanceVectorHop *prev_dv;
1010
1011   /**
1012    * Kept in a MDLL.
1013    */
1014   struct DistanceVectorHop *next_neighbour;
1015
1016   /**
1017    * Kept in a MDLL.
1018    */
1019   struct DistanceVectorHop *prev_neighbour;
1020
1021   /**
1022    * What would be the next hop to @e target?
1023    */
1024   struct Neighbour *next_hop;
1025
1026   /**
1027    * Distance vector entry this hop belongs with.
1028    */
1029   struct DistanceVector *dv;
1030
1031   /**
1032    * Array of @e distance hops to the target, excluding @e next_hop.
1033    * NULL if the entire path is us to @e next_hop to `target`. Allocated
1034    * at the end of this struct.
1035    */
1036   const struct GNUNET_PeerIdentity *path;
1037
1038   /**
1039    * At what time do we forget about this path unless we see it again
1040    * while learning?
1041    */
1042   struct GNUNET_TIME_Absolute timeout;
1043
1044   /**
1045    * After what time do we know for sure that the path must have existed?
1046    * Set to ZERO if the path is learned by snooping on DV learn messages
1047    * initiated by other peers, and to the time at which we generated the
1048    * challenge for DV learn operations this peer initiated.
1049    */
1050   struct GNUNET_TIME_Absolute freshness;
1051
1052   /**
1053    * How many hops in total to the `target` (excluding @e next_hop and `target` itself),
1054    * thus 0 still means a distance of 2 hops (to @e next_hop and then to `target`)?
1055    */
1056   unsigned int distance;
1057 };
1058
1059
1060 /**
1061  * Entry in our #dv_routes table, representing a (set of) distance
1062  * vector routes to a particular peer.
1063  */
1064 struct DistanceVector
1065 {
1066
1067   /**
1068    * To which peer is this a route?
1069    */
1070   struct GNUNET_PeerIdentity target;
1071
1072   /**
1073    * Known paths to @e target.
1074    */
1075   struct DistanceVectorHop *dv_head;
1076
1077   /**
1078    * Known paths to @e target.
1079    */
1080   struct DistanceVectorHop *dv_tail;
1081
1082   /**
1083    * Task scheduled to purge expired paths from @e dv_head MDLL.
1084    */
1085   struct GNUNET_SCHEDULER_Task *timeout_task;
1086
1087   /**
1088    * Is one of the DV paths in this struct 'confirmed' and thus
1089    * the cause for CORE to see this peer as connected? (Note that
1090    * the same may apply to a `struct Neighbour` at the same time.)
1091    */
1092   int core_visible;
1093 };
1094
1095
1096 /**
1097  * A queue is a message queue provided by a communicator
1098  * via which we can reach a particular neighbour.
1099  */
1100 struct Queue;
1101
1102
1103 /**
1104  * Entry identifying transmission in one of our `struct
1105  * Queue` which still awaits an ACK.  This is used to
1106  * ensure we do not overwhelm a communicator and limit the number of
1107  * messages outstanding per communicator (say in case communicator is
1108  * CPU bound) and per queue (in case bandwidth allocation exceeds
1109  * what the communicator can actually provide towards a particular
1110  * peer/target).
1111  */
1112 struct QueueEntry
1113 {
1114
1115   /**
1116    * Kept as a DLL.
1117    */
1118   struct QueueEntry *next;
1119
1120   /**
1121    * Kept as a DLL.
1122    */
1123   struct QueueEntry *prev;
1124
1125   /**
1126    * Queue this entry is queued with.
1127    */
1128   struct Queue *queue;
1129
1130   /**
1131    * Message ID used for this message with the queue used for transmission.
1132    */
1133   uint64_t mid;
1134 };
1135
1136
1137 /**
1138  * A queue is a message queue provided by a communicator
1139  * via which we can reach a particular neighbour.
1140  */
1141 struct Queue
1142 {
1143   /**
1144    * Kept in a MDLL.
1145    */
1146   struct Queue *next_neighbour;
1147
1148   /**
1149    * Kept in a MDLL.
1150    */
1151   struct Queue *prev_neighbour;
1152
1153   /**
1154    * Kept in a MDLL.
1155    */
1156   struct Queue *prev_client;
1157
1158   /**
1159    * Kept in a MDLL.
1160    */
1161   struct Queue *next_client;
1162
1163   /**
1164    * Head of DLL of unacked transmission requests.
1165    */
1166   struct QueueEntry *queue_head;
1167
1168   /**
1169    * End of DLL of unacked transmission requests.
1170    */
1171   struct QueueEntry *queue_tail;
1172
1173   /**
1174    * Which neighbour is this queue for?
1175    */
1176   struct Neighbour *neighbour;
1177
1178   /**
1179    * Which communicator offers this queue?
1180    */
1181   struct TransportClient *tc;
1182
1183   /**
1184    * Address served by the queue.
1185    */
1186   const char *address;
1187
1188   /**
1189    * Task scheduled for the time when this queue can (likely) transmit the
1190    * next message. Still needs to check with the @e tracker_out to be sure.
1191    */
1192   struct GNUNET_SCHEDULER_Task *transmit_task;
1193
1194   /**
1195    * Task scheduled to possibly notfiy core that this queue is no longer
1196    * counting as confirmed.  Runs the #core_queue_visibility_check().
1197    */
1198   struct GNUNET_SCHEDULER_Task *visibility_task;
1199
1200   /**
1201    * Our current RTT estimate for this queue.
1202    */
1203   struct GNUNET_TIME_Relative rtt;
1204
1205   /**
1206    * How long do *we* consider this @e address to be valid?  In the past or
1207    * zero if we have not yet validated it.  Can be updated based on
1208    * challenge-response validations (via address validation logic), or when we
1209    * receive ACKs that we can definitively map to transmissions via this
1210    * queue.
1211    */
1212   struct GNUNET_TIME_Absolute validated_until;
1213
1214   /**
1215    * Message ID generator for transmissions on this queue.
1216    */
1217   uint64_t mid_gen;
1218
1219   /**
1220    * Unique identifier of this queue with the communicator.
1221    */
1222   uint32_t qid;
1223
1224   /**
1225    * Maximum transmission unit supported by this queue.
1226    */
1227   uint32_t mtu;
1228
1229   /**
1230    * Distance to the target of this queue.
1231    * FIXME: needed? DV is done differently these days...
1232    */
1233   uint32_t distance;
1234
1235   /**
1236    * Messages pending.
1237    */
1238   uint32_t num_msg_pending;
1239
1240   /**
1241    * Bytes pending.
1242    */
1243   uint32_t num_bytes_pending;
1244
1245   /**
1246    * Length of the DLL starting at @e queue_head.
1247    */
1248   unsigned int queue_length;
1249
1250   /**
1251    * Network type offered by this queue.
1252    */
1253   enum GNUNET_NetworkType nt;
1254
1255   /**
1256    * Connection status for this queue.
1257    */
1258   enum GNUNET_TRANSPORT_ConnectionStatus cs;
1259
1260   /**
1261    * How much outbound bandwidth do we have available for this queue?
1262    */
1263   struct GNUNET_BANDWIDTH_Tracker tracker_out;
1264
1265   /**
1266    * How much inbound bandwidth do we have available for this queue?
1267    */
1268   struct GNUNET_BANDWIDTH_Tracker tracker_in;
1269 };
1270
1271
1272 /**
1273  * Information we keep for a message that we are reassembling.
1274  */
1275 struct ReassemblyContext
1276 {
1277
1278   /**
1279    * Original message ID for of the message that all the
1280    * fragments belong to.
1281    */
1282   struct GNUNET_ShortHashCode msg_uuid;
1283
1284   /**
1285    * Which neighbour is this context for?
1286    */
1287   struct Neighbour *neighbour;
1288
1289   /**
1290    * Entry in the reassembly heap (sorted by expiration).
1291    */
1292   struct GNUNET_CONTAINER_HeapNode *hn;
1293
1294   /**
1295    * Bitfield with @e msg_size bits representing the positions
1296    * where we have received fragments.  When we receive a fragment,
1297    * we check the bits in @e bitfield before incrementing @e msg_missing.
1298    *
1299    * Allocated after the reassembled message.
1300    */
1301   uint8_t *bitfield;
1302
1303   /**
1304    * Task for sending ACK. We may send ACKs either because of hitting
1305    * the @e extra_acks limit, or based on time and @e num_acks.  This
1306    * task is for the latter case.
1307    */
1308   struct GNUNET_SCHEDULER_Task *ack_task;
1309
1310   /**
1311    * At what time will we give up reassembly of this message?
1312    */
1313   struct GNUNET_TIME_Absolute reassembly_timeout;
1314
1315   /**
1316    * Average delay of all acks in @e extra_acks and @e frag_uuid.
1317    * Should be reset to zero when @e num_acks is set to 0.
1318    */
1319   struct GNUNET_TIME_Relative avg_ack_delay;
1320
1321   /**
1322    * Time we received the last fragment.  @e avg_ack_delay must be
1323    * incremented by now - @e last_frag multiplied by @e num_acks.
1324    */
1325   struct GNUNET_TIME_Absolute last_frag;
1326
1327   /**
1328    * Bitfield of up to 64 additional fragments following @e frag_uuid
1329    * to be acknowledged in the next cummulative ACK.
1330    */
1331   uint64_t extra_acks;
1332
1333   /**
1334    * Unique ID of the lowest fragment UUID to be acknowledged in the
1335    * next cummulative ACK.  Only valid if @e num_acks > 0.
1336    */
1337   uint32_t frag_uuid;
1338
1339   /**
1340    * Number of ACKs we have accumulated so far.  Reset to 0
1341    * whenever we send a #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK.
1342    */
1343   unsigned int num_acks;
1344
1345   /**
1346    * How big is the message we are reassembling in total?
1347    */
1348   uint16_t msg_size;
1349
1350   /**
1351    * How many bytes of the message are still missing?  Defragmentation
1352    * is complete when @e msg_missing == 0.
1353    */
1354   uint16_t msg_missing;
1355
1356   /* Followed by @e msg_size bytes of the (partially) defragmented original message */
1357
1358   /* Followed by @e bitfield data */
1359 };
1360
1361
1362 /**
1363  * A neighbour that at least one communicator is connected to.
1364  */
1365 struct Neighbour
1366 {
1367
1368   /**
1369    * Which peer is this about?
1370    */
1371   struct GNUNET_PeerIdentity pid;
1372
1373   /**
1374    * Map with `struct ReassemblyContext` structs for fragments under
1375    * reassembly. May be NULL if we currently have no fragments from
1376    * this @e pid (lazy initialization).
1377    */
1378   struct GNUNET_CONTAINER_MultiShortmap *reassembly_map;
1379
1380   /**
1381    * Heap with `struct ReassemblyContext` structs for fragments under
1382    * reassembly. May be NULL if we currently have no fragments from
1383    * this @e pid (lazy initialization).
1384    */
1385   struct GNUNET_CONTAINER_Heap *reassembly_heap;
1386
1387   /**
1388    * Task to free old entries from the @e reassembly_heap and @e reassembly_map.
1389    */
1390   struct GNUNET_SCHEDULER_Task *reassembly_timeout_task;
1391
1392   /**
1393    * Head of list of messages pending for this neighbour.
1394    */
1395   struct PendingMessage *pending_msg_head;
1396
1397   /**
1398    * Tail of list of messages pending for this neighbour.
1399    */
1400   struct PendingMessage *pending_msg_tail;
1401
1402   /**
1403    * Head of MDLL of DV hops that have this neighbour as next hop. Must be
1404    * purged if this neighbour goes down.
1405    */
1406   struct DistanceVectorHop *dv_head;
1407
1408   /**
1409    * Tail of MDLL of DV hops that have this neighbour as next hop. Must be
1410    * purged if this neighbour goes down.
1411    */
1412   struct DistanceVectorHop *dv_tail;
1413
1414   /**
1415    * Head of DLL of queues to this peer.
1416    */
1417   struct Queue *queue_head;
1418
1419   /**
1420    * Tail of DLL of queues to this peer.
1421    */
1422   struct Queue *queue_tail;
1423
1424   /**
1425    * Task run to cleanup pending messages that have exceeded their timeout.
1426    */
1427   struct GNUNET_SCHEDULER_Task *timeout_task;
1428
1429   /**
1430    * Quota at which CORE is allowed to transmit to this peer.
1431    *
1432    * FIXME: not yet used, tricky to get right given multiple queues!
1433    *        (=> Idea: measure???)
1434    * FIXME: how do we set this value initially when we tell CORE?
1435    *    Options: start at a minimum value or at literally zero?
1436    *         (=> Current thought: clean would be zero!)
1437    */
1438   struct GNUNET_BANDWIDTH_Value32NBO quota_out;
1439
1440   /**
1441    * What is the earliest timeout of any message in @e pending_msg_tail?
1442    */
1443   struct GNUNET_TIME_Absolute earliest_timeout;
1444
1445   /**
1446    * Do we have a confirmed working queue and are thus visible to
1447    * CORE?
1448    */
1449   int core_visible;
1450 };
1451
1452
1453 /**
1454  * A peer that an application (client) would like us to talk to directly.
1455  */
1456 struct PeerRequest
1457 {
1458
1459   /**
1460    * Which peer is this about?
1461    */
1462   struct GNUNET_PeerIdentity pid;
1463
1464   /**
1465    * Client responsible for the request.
1466    */
1467   struct TransportClient *tc;
1468
1469   /**
1470    * Handle for watching the peerstore for HELLOs for this peer.
1471    */
1472   struct GNUNET_PEERSTORE_WatchContext *wc;
1473
1474   /**
1475    * What kind of performance preference does this @e tc have?
1476    */
1477   enum GNUNET_MQ_PreferenceKind pk;
1478
1479   /**
1480    * How much bandwidth would this @e tc like to see?
1481    */
1482   struct GNUNET_BANDWIDTH_Value32NBO bw;
1483
1484 };
1485
1486
1487 /**
1488  * Types of different pending messages.
1489  */
1490 enum PendingMessageType
1491 {
1492
1493   /**
1494    * Ordinary message received from the CORE service.
1495    */
1496   PMT_CORE = 0,
1497
1498   /**
1499    * Fragment box.
1500    */
1501   PMT_FRAGMENT_BOX = 1,
1502
1503   /**
1504    * Reliability box.
1505    */
1506   PMT_RELIABILITY_BOX = 2,
1507
1508   /**
1509    * Any type of acknowledgement.
1510    */
1511   PMT_ACKNOWLEDGEMENT = 3,
1512
1513   /**
1514    * Control traffic generated by the TRANSPORT service itself.
1515    */
1516   PMT_CONTROL = 4
1517
1518 };
1519
1520
1521 /**
1522  * Transmission request that is awaiting delivery.  The original
1523  * transmission requests from CORE may be too big for some queues.
1524  * In this case, a *tree* of fragments is created.  At each
1525  * level of the tree, fragments are kept in a DLL ordered by which
1526  * fragment should be sent next (at the head).  The tree is searched
1527  * top-down, with the original message at the root.
1528  *
1529  * To select a node for transmission, first it is checked if the
1530  * current node's message fits with the MTU.  If it does not, we
1531  * either calculate the next fragment (based on @e frag_off) from the
1532  * current node, or, if all fragments have already been created,
1533  * descend to the @e head_frag.  Even though the node was already
1534  * fragmented, the fragment may be too big if the fragment was
1535  * generated for a queue with a larger MTU. In this case, the node
1536  * may be fragmented again, thus creating a tree.
1537  *
1538  * When acknowledgements for fragments are received, the tree
1539  * must be pruned, removing those parts that were already
1540  * acknowledged.  When fragments are sent over a reliable
1541  * channel, they can be immediately removed.
1542  *
1543  * If a message is ever fragmented, then the original "full" message
1544  * is never again transmitted (even if it fits below the MTU), and
1545  * only (remaining) fragments are sent.
1546  */
1547 struct PendingMessage
1548 {
1549   /**
1550    * Kept in a MDLL of messages for this @a target.
1551    */
1552   struct PendingMessage *next_neighbour;
1553
1554   /**
1555    * Kept in a MDLL of messages for this @a target.
1556    */
1557   struct PendingMessage *prev_neighbour;
1558
1559   /**
1560    * Kept in a MDLL of messages from this @a client (if @e pmt is #PMT_CORE)
1561    */
1562   struct PendingMessage *next_client;
1563
1564   /**
1565    * Kept in a MDLL of messages from this @a client  (if @e pmt is #PMT_CORE)
1566    */
1567   struct PendingMessage *prev_client;
1568
1569   /**
1570    * Kept in a MDLL of messages from this @a cpm (if @e pmt is #PMT_FRAGMENT_BOx)
1571    */
1572   struct PendingMessage *next_frag;
1573
1574   /**
1575    * Kept in a MDLL of messages from this @a cpm  (if @e pmt is #PMT_FRAGMENT_BOX)
1576    */
1577   struct PendingMessage *prev_frag;
1578
1579   /**
1580    * This message, reliability boxed. Only possibly available if @e pmt is #PMT_CORE.
1581    */
1582   struct PendingMessage *bpm;
1583
1584   /**
1585    * Target of the request.
1586    */
1587   struct Neighbour *target;
1588
1589   /**
1590    * Client that issued the transmission request, if @e pmt is #PMT_CORE.
1591    */
1592   struct TransportClient *client;
1593
1594   /**
1595    * Head of a MDLL of fragments created for this core message.
1596    */
1597   struct PendingMessage *head_frag;
1598
1599   /**
1600    * Tail of a MDLL of fragments created for this core message.
1601    */
1602   struct PendingMessage *tail_frag;
1603
1604   /**
1605    * Our parent in the fragmentation tree.
1606    */
1607   struct PendingMessage *frag_parent;
1608
1609   /**
1610    * At what time should we give up on the transmission (and no longer retry)?
1611    */
1612   struct GNUNET_TIME_Absolute timeout;
1613
1614   /**
1615    * What is the earliest time for us to retry transmission of this message?
1616    */
1617   struct GNUNET_TIME_Absolute next_attempt;
1618
1619   /**
1620    * UUID to use for this message (used for reassembly of fragments, only
1621    * initialized if @e msg_uuid_set is #GNUNET_YES).
1622    */
1623   struct GNUNET_ShortHashCode msg_uuid;
1624
1625   /**
1626    * Counter incremented per generated fragment.
1627    */
1628   uint32_t frag_uuidgen;
1629
1630   /**
1631    * Type of the pending message.
1632    */
1633   enum PendingMessageType pmt;
1634
1635   /**
1636    * Size of the original message.
1637    */
1638   uint16_t bytes_msg;
1639
1640   /**
1641    * Offset at which we should generate the next fragment.
1642    */
1643   uint16_t frag_off;
1644
1645   /**
1646    * #GNUNET_YES once @e msg_uuid was initialized
1647    */
1648   int16_t msg_uuid_set;
1649
1650   /* Followed by @e bytes_msg to transmit */
1651 };
1652
1653
1654 /**
1655  * One of the addresses of this peer.
1656  */
1657 struct AddressListEntry
1658 {
1659
1660   /**
1661    * Kept in a DLL.
1662    */
1663   struct AddressListEntry *next;
1664
1665   /**
1666    * Kept in a DLL.
1667    */
1668   struct AddressListEntry *prev;
1669
1670   /**
1671    * Which communicator provides this address?
1672    */
1673   struct TransportClient *tc;
1674
1675   /**
1676    * The actual address.
1677    */
1678   const char *address;
1679
1680   /**
1681    * Current context for storing this address in the peerstore.
1682    */
1683   struct GNUNET_PEERSTORE_StoreContext *sc;
1684
1685   /**
1686    * Task to periodically do @e st operation.
1687    */
1688   struct GNUNET_SCHEDULER_Task *st;
1689
1690   /**
1691    * What is a typical lifetime the communicator expects this
1692    * address to have? (Always from now.)
1693    */
1694   struct GNUNET_TIME_Relative expiration;
1695
1696   /**
1697    * Address identifier used by the communicator.
1698    */
1699   uint32_t aid;
1700
1701   /**
1702    * Network type offered by this address.
1703    */
1704   enum GNUNET_NetworkType nt;
1705
1706 };
1707
1708
1709 /**
1710  * Client connected to the transport service.
1711  */
1712 struct TransportClient
1713 {
1714
1715   /**
1716    * Kept in a DLL.
1717    */
1718   struct TransportClient *next;
1719
1720   /**
1721    * Kept in a DLL.
1722    */
1723   struct TransportClient *prev;
1724
1725   /**
1726    * Handle to the client.
1727    */
1728   struct GNUNET_SERVICE_Client *client;
1729
1730   /**
1731    * Message queue to the client.
1732    */
1733   struct GNUNET_MQ_Handle *mq;
1734
1735   /**
1736    * What type of client is this?
1737    */
1738   enum ClientType type;
1739
1740   union
1741   {
1742
1743     /**
1744      * Information for @e type #CT_CORE.
1745      */
1746     struct {
1747
1748       /**
1749        * Head of list of messages pending for this client, sorted by
1750        * transmission time ("next_attempt" + possibly internal prioritization).
1751        */
1752       struct PendingMessage *pending_msg_head;
1753
1754       /**
1755        * Tail of list of messages pending for this client.
1756        */
1757       struct PendingMessage *pending_msg_tail;
1758
1759     } core;
1760
1761     /**
1762      * Information for @e type #CT_MONITOR.
1763      */
1764     struct {
1765
1766       /**
1767        * Peer identity to monitor the addresses of.
1768        * Zero to monitor all neighbours.  Valid if
1769        * @e type is #CT_MONITOR.
1770        */
1771       struct GNUNET_PeerIdentity peer;
1772
1773       /**
1774        * Is this a one-shot monitor?
1775        */
1776       int one_shot;
1777
1778     } monitor;
1779
1780
1781     /**
1782      * Information for @e type #CT_COMMUNICATOR.
1783      */
1784     struct {
1785       /**
1786        * If @e type is #CT_COMMUNICATOR, this communicator
1787        * supports communicating using these addresses.
1788        */
1789       char *address_prefix;
1790
1791       /**
1792        * Head of DLL of queues offered by this communicator.
1793        */
1794       struct Queue *queue_head;
1795
1796       /**
1797        * Tail of DLL of queues offered by this communicator.
1798        */
1799       struct Queue *queue_tail;
1800
1801       /**
1802        * Head of list of the addresses of this peer offered by this communicator.
1803        */
1804       struct AddressListEntry *addr_head;
1805
1806       /**
1807        * Tail of list of the addresses of this peer offered by this communicator.
1808        */
1809       struct AddressListEntry *addr_tail;
1810
1811       /**
1812        * Number of queue entries in all queues to this communicator. Used
1813        * throttle sending to a communicator if we see that the communicator
1814        * is globally unable to keep up.
1815        */
1816       unsigned int total_queue_length;
1817
1818       /**
1819        * Characteristics of this communicator.
1820        */
1821       enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
1822
1823     } communicator;
1824
1825     /**
1826      * Information for @e type #CT_APPLICATION
1827      */
1828     struct {
1829
1830       /**
1831        * Map of requests for peers the given client application would like to
1832        * see connections for.  Maps from PIDs to `struct PeerRequest`.
1833        */
1834       struct GNUNET_CONTAINER_MultiPeerMap *requests;
1835
1836     } application;
1837
1838   } details;
1839
1840 };
1841
1842
1843 /**
1844  * State we keep for validation activities.  Each of these
1845  * is both in the #validation_heap and the #validation_map.
1846  */
1847 struct ValidationState
1848 {
1849
1850   /**
1851    * For which peer is @a address to be validated (or possibly valid)?
1852    * Serves as key in the #validation_map.
1853    */
1854   struct GNUNET_PeerIdentity pid;
1855
1856   /**
1857    * How long did the peer claim this @e address to be valid? Capped at
1858    * minimum of #MAX_ADDRESS_VALID_UNTIL relative to the time where we last
1859    * were told about the address and the value claimed by the other peer at
1860    * that time.  May be updated similarly when validation succeeds.
1861    */
1862   struct GNUNET_TIME_Absolute valid_until;
1863
1864   /**
1865    * How long do *we* consider this @e address to be valid?
1866    * In the past or zero if we have not yet validated it.
1867    */
1868   struct GNUNET_TIME_Absolute validated_until;
1869
1870   /**
1871    * When did we FIRST use the current @e challenge in a message?
1872    * Used to sanity-check @code{origin_time} in the response when
1873    * calculating the RTT. If the @code{origin_time} is not in
1874    * the expected range, the response is discarded as malicious.
1875    */
1876   struct GNUNET_TIME_Absolute first_challenge_use;
1877
1878   /**
1879    * When did we LAST use the current @e challenge in a message?
1880    * Used to sanity-check @code{origin_time} in the response when
1881    * calculating the RTT.  If the @code{origin_time} is not in
1882    * the expected range, the response is discarded as malicious.
1883    */
1884   struct GNUNET_TIME_Absolute last_challenge_use;
1885
1886   /**
1887    * Next time we will send the @e challenge to the peer, if this time is past
1888    * @e valid_until, this validation state is released at this time.  If the
1889    * address is valid, @e next_challenge is set to @e validated_until MINUS @e
1890    * validation_delay * #VALIDATION_RTT_BUFFER_FACTOR, such that we will try
1891    * to re-validate before the validity actually expires.
1892    */
1893   struct GNUNET_TIME_Absolute next_challenge;
1894
1895   /**
1896    * Current backoff factor we're applying for sending the @a challenge.
1897    * Reset to 0 if the @a challenge is confirmed upon validation.
1898    * Reduced to minimum of #FAST_VALIDATION_CHALLENGE_FREQ and half of the
1899    * existing value if we receive an unvalidated address again over
1900    * another channel (and thus should consider the information "fresh").
1901    * Maximum is #MAX_VALIDATION_CHALLENGE_FREQ.
1902    */
1903   struct GNUNET_TIME_Relative challenge_backoff;
1904
1905   /**
1906    * Initially set to "forever". Once @e validated_until is set, this value is
1907    * set to the RTT that tells us how long it took to receive the validation.
1908    */
1909   struct GNUNET_TIME_Relative validation_rtt;
1910
1911   /**
1912    * The challenge we sent to the peer to get it to validate the address. Note
1913    * that we rotate the challenge whenever we update @e validated_until to
1914    * avoid attacks where a peer simply replays an old challenge in the future.
1915    * (We must not rotate more often as otherwise we may discard valid answers
1916    * due to packet losses, latency and reorderings on the network).
1917    */
1918   struct GNUNET_ShortHashCode challenge;
1919
1920   /**
1921    * Claimed address of the peer.
1922    */
1923   char *address;
1924
1925   /**
1926    * Entry in the #validation_heap, which is sorted by @e next_challenge. The
1927    * heap is used to figure out when the next validation activity should be
1928    * run.
1929    */
1930   struct GNUNET_CONTAINER_HeapNode *hn;
1931
1932   /**
1933    * Handle to a PEERSTORE store operation for this @e address.  NULL if
1934    * no PEERSTORE operation is pending.
1935    */
1936   struct GNUNET_PEERSTORE_StoreContext *sc;
1937
1938   /**
1939    * We are technically ready to send the challenge, but we are waiting for
1940    * the respective queue to become available for transmission.
1941    */
1942   int awaiting_queue;
1943
1944 };
1945
1946
1947 /**
1948  * Head of linked list of all clients to this service.
1949  */
1950 static struct TransportClient *clients_head;
1951
1952 /**
1953  * Tail of linked list of all clients to this service.
1954  */
1955 static struct TransportClient *clients_tail;
1956
1957 /**
1958  * Statistics handle.
1959  */
1960 static struct GNUNET_STATISTICS_Handle *GST_stats;
1961
1962 /**
1963  * Configuration handle.
1964  */
1965 static const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
1966
1967 /**
1968  * Our public key.
1969  */
1970 static struct GNUNET_PeerIdentity GST_my_identity;
1971
1972 /**
1973  * Our private key.
1974  */
1975 static struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
1976
1977 /**
1978  * Map from PIDs to `struct Neighbour` entries.  A peer is
1979  * a neighbour if we have an MQ to it from some communicator.
1980  */
1981 static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
1982
1983 /**
1984  * Map from PIDs to `struct DistanceVector` entries describing
1985  * known paths to the peer.
1986  */
1987 static struct GNUNET_CONTAINER_MultiPeerMap *dv_routes;
1988
1989 /**
1990  * Map from PIDs to `struct ValidationState` entries describing
1991  * addresses we are aware of and their validity state.
1992  */
1993 static struct GNUNET_CONTAINER_MultiPeerMap *validation_map;
1994
1995 /**
1996  * Map from challenges to `struct LearnLaunchEntry` values.
1997  */
1998 static struct GNUNET_CONTAINER_MultiShortmap *dvlearn_map;
1999
2000 /**
2001  * Head of a DLL sorted by launch time.
2002  */
2003 static struct LearnLaunchEntry *lle_head;
2004
2005 /**
2006  * Tail of a DLL sorted by launch time.
2007  */
2008 static struct LearnLaunchEntry *lle_tail;
2009
2010 /**
2011  * MIN Heap sorted by "next_challenge" to `struct ValidationState` entries
2012  * sorting addresses we are aware of by when we should next try to (re)validate
2013  * (or expire) them.
2014  */
2015 static struct GNUNET_CONTAINER_Heap *validation_heap;
2016
2017 /**
2018  * Database for peer's HELLOs.
2019  */
2020 static struct GNUNET_PEERSTORE_Handle *peerstore;
2021
2022 /**
2023  * Heap sorting `struct EphemeralCacheEntry` by their
2024  * key/signature validity.
2025  */
2026 static struct GNUNET_CONTAINER_Heap *ephemeral_heap;
2027
2028 /**
2029  * Hash map for looking up `struct EphemeralCacheEntry`s
2030  * by peer identity. (We may have ephemerals in our
2031  * cache for which we do not have a neighbour entry,
2032  * and similar many neighbours may not need ephemerals,
2033  * so we use a second map.)
2034  */
2035 static struct GNUNET_CONTAINER_MultiPeerMap *ephemeral_map;
2036
2037 /**
2038  * Task to free expired ephemerals.
2039  */
2040 static struct GNUNET_SCHEDULER_Task *ephemeral_task;
2041
2042 /**
2043  * Task run to initiate DV learning.
2044  */
2045 static struct GNUNET_SCHEDULER_Task *dvlearn_task;
2046
2047 /**
2048  * Task to run address validation.
2049  */
2050 static struct GNUNET_SCHEDULER_Task *validation_task;
2051
2052
2053 /**
2054  * Free cached ephemeral key.
2055  *
2056  * @param ece cached signature to free
2057  */
2058 static void
2059 free_ephemeral (struct EphemeralCacheEntry *ece)
2060 {
2061   GNUNET_CONTAINER_multipeermap_remove (ephemeral_map,
2062                                         &ece->target,
2063                                         ece);
2064   GNUNET_CONTAINER_heap_remove_node (ece->hn);
2065   GNUNET_free (ece);
2066 }
2067
2068
2069 /**
2070  * Free validation state.
2071  *
2072  * @param vs validation state to free
2073  */
2074 static void
2075 free_validation_state (struct ValidationState *vs)
2076 {
2077   GNUNET_CONTAINER_multipeermap_remove (validation_map,
2078                                         &vs->pid,
2079                                         vs);
2080   GNUNET_CONTAINER_heap_remove_node (vs->hn);
2081   vs->hn = NULL;
2082   if (NULL != vs->sc)
2083   {
2084     GNUNET_PEERSTORE_store_cancel (vs->sc);
2085     vs->sc = NULL;
2086   }
2087   GNUNET_free (vs->address);
2088   GNUNET_free (vs);
2089 }
2090
2091
2092 /**
2093  * Lookup neighbour record for peer @a pid.
2094  *
2095  * @param pid neighbour to look for
2096  * @return NULL if we do not have this peer as a neighbour
2097  */
2098 static struct Neighbour *
2099 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
2100 {
2101   return GNUNET_CONTAINER_multipeermap_get (neighbours,
2102                                             pid);
2103 }
2104
2105
2106 /**
2107  * Details about what to notify monitors about.
2108  */
2109 struct MonitorEvent
2110 {
2111   /**
2112    * @deprecated To be discussed if we keep these...
2113    */
2114   struct GNUNET_TIME_Absolute last_validation;
2115   struct GNUNET_TIME_Absolute valid_until;
2116   struct GNUNET_TIME_Absolute next_validation;
2117
2118   /**
2119    * Current round-trip time estimate.
2120    */
2121   struct GNUNET_TIME_Relative rtt;
2122
2123   /**
2124    * Connection status.
2125    */
2126   enum GNUNET_TRANSPORT_ConnectionStatus cs;
2127
2128   /**
2129    * Messages pending.
2130    */
2131   uint32_t num_msg_pending;
2132
2133   /**
2134    * Bytes pending.
2135    */
2136   uint32_t num_bytes_pending;
2137
2138
2139 };
2140
2141
2142 /**
2143  * Free a @dvh. Callers MAY want to check if this was the last path to the
2144  * `target`, and if so call #free_dv_route to also free the associated DV
2145  * entry in #dv_routes (if not, the associated scheduler job should eventually
2146  * take care of it).
2147  *
2148  * @param dvh hop to free
2149  */
2150 static void
2151 free_distance_vector_hop (struct DistanceVectorHop *dvh)
2152 {
2153   struct Neighbour *n = dvh->next_hop;
2154   struct DistanceVector *dv = dvh->dv;
2155
2156   GNUNET_CONTAINER_MDLL_remove (neighbour,
2157                                 n->dv_head,
2158                                 n->dv_tail,
2159                                 dvh);
2160   GNUNET_CONTAINER_MDLL_remove (dv,
2161                                 dv->dv_head,
2162                                 dv->dv_tail,
2163                                 dvh);
2164   GNUNET_free (dvh);
2165 }
2166
2167
2168 /**
2169  * Free entry in #dv_routes.  First frees all hops to the target, and
2170  * if there are no entries left, frees @a dv as well.
2171  *
2172  * @param dv route to free
2173  */
2174 static void
2175 free_dv_route (struct DistanceVector *dv)
2176 {
2177   struct DistanceVectorHop *dvh;
2178
2179   while (NULL != (dvh = dv->dv_head))
2180     free_distance_vector_hop (dvh);
2181   if (NULL == dv->dv_head)
2182   {
2183     GNUNET_assert (GNUNET_YES ==
2184                    GNUNET_CONTAINER_multipeermap_remove (dv_routes,
2185                                                          &dv->target,
2186                                                          dv));
2187     if (NULL != dv->timeout_task)
2188       GNUNET_SCHEDULER_cancel (dv->timeout_task);
2189     GNUNET_free (dv);
2190   }
2191 }
2192
2193
2194 /**
2195  * Notify monitor @a tc about an event.  That @a tc
2196  * cares about the event has already been checked.
2197  *
2198  * Send @a tc information in @a me about a @a peer's status with
2199  * respect to some @a address to all monitors that care.
2200  *
2201  * @param tc monitor to inform
2202  * @param peer peer the information is about
2203  * @param address address the information is about
2204  * @param nt network type associated with @a address
2205  * @param me detailed information to transmit
2206  */
2207 static void
2208 notify_monitor (struct TransportClient *tc,
2209                 const struct GNUNET_PeerIdentity *peer,
2210                 const char *address,
2211                 enum GNUNET_NetworkType nt,
2212                 const struct MonitorEvent *me)
2213 {
2214   struct GNUNET_MQ_Envelope *env;
2215   struct GNUNET_TRANSPORT_MonitorData *md;
2216   size_t addr_len = strlen (address) + 1;
2217
2218   env = GNUNET_MQ_msg_extra (md,
2219                              addr_len,
2220                              GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_DATA);
2221   md->nt = htonl ((uint32_t) nt);
2222   md->peer = *peer;
2223   md->last_validation = GNUNET_TIME_absolute_hton (me->last_validation);
2224   md->valid_until = GNUNET_TIME_absolute_hton (me->valid_until);
2225   md->next_validation = GNUNET_TIME_absolute_hton (me->next_validation);
2226   md->rtt = GNUNET_TIME_relative_hton (me->rtt);
2227   md->cs = htonl ((uint32_t) me->cs);
2228   md->num_msg_pending = htonl (me->num_msg_pending);
2229   md->num_bytes_pending = htonl (me->num_bytes_pending);
2230   memcpy (&md[1],
2231           address,
2232           addr_len);
2233   GNUNET_MQ_send (tc->mq,
2234                   env);
2235 }
2236
2237
2238 /**
2239  * Send information in @a me about a @a peer's status with respect
2240  * to some @a address to all monitors that care.
2241  *
2242  * @param peer peer the information is about
2243  * @param address address the information is about
2244  * @param nt network type associated with @a address
2245  * @param me detailed information to transmit
2246  */
2247 static void
2248 notify_monitors (const struct GNUNET_PeerIdentity *peer,
2249                  const char *address,
2250                  enum GNUNET_NetworkType nt,
2251                  const struct MonitorEvent *me)
2252 {
2253   for (struct TransportClient *tc = clients_head;
2254        NULL != tc;
2255        tc = tc->next)
2256   {
2257     if (CT_MONITOR != tc->type)
2258       continue;
2259     if (tc->details.monitor.one_shot)
2260       continue;
2261     if ( (0 != GNUNET_is_zero (&tc->details.monitor.peer)) &&
2262          (0 != GNUNET_memcmp (&tc->details.monitor.peer,
2263                               peer)) )
2264       continue;
2265     notify_monitor (tc,
2266                     peer,
2267                     address,
2268                     nt,
2269                     me);
2270   }
2271 }
2272
2273
2274 /**
2275  * Called whenever a client connects.  Allocates our
2276  * data structures associated with that client.
2277  *
2278  * @param cls closure, NULL
2279  * @param client identification of the client
2280  * @param mq message queue for the client
2281  * @return our `struct TransportClient`
2282  */
2283 static void *
2284 client_connect_cb (void *cls,
2285                    struct GNUNET_SERVICE_Client *client,
2286                    struct GNUNET_MQ_Handle *mq)
2287 {
2288   struct TransportClient *tc;
2289
2290   (void) cls;
2291   tc = GNUNET_new (struct TransportClient);
2292   tc->client = client;
2293   tc->mq = mq;
2294   GNUNET_CONTAINER_DLL_insert (clients_head,
2295                                clients_tail,
2296                                tc);
2297   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2298               "Client %p connected\n",
2299               tc);
2300   return tc;
2301 }
2302
2303
2304 /**
2305  * Free @a rc
2306  *
2307  * @param rc data structure to free
2308  */
2309 static void
2310 free_reassembly_context (struct ReassemblyContext *rc)
2311 {
2312   struct Neighbour *n = rc->neighbour;
2313
2314   GNUNET_assert (rc ==
2315                  GNUNET_CONTAINER_heap_remove_node (rc->hn));
2316   GNUNET_assert (GNUNET_OK ==
2317                  GNUNET_CONTAINER_multishortmap_remove (n->reassembly_map,
2318                                                         &rc->msg_uuid,
2319                                                         rc));
2320   GNUNET_free (rc);
2321 }
2322
2323
2324 /**
2325  * Task run to clean up reassembly context of a neighbour that have expired.
2326  *
2327  * @param cls a `struct Neighbour`
2328  */
2329 static void
2330 reassembly_cleanup_task (void *cls)
2331 {
2332   struct Neighbour *n = cls;
2333   struct ReassemblyContext *rc;
2334
2335   n->reassembly_timeout_task = NULL;
2336   while (NULL != (rc = GNUNET_CONTAINER_heap_peek (n->reassembly_heap)))
2337   {
2338     if (0 == GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout).rel_value_us)
2339     {
2340       free_reassembly_context (rc);
2341       continue;
2342     }
2343     GNUNET_assert (NULL == n->reassembly_timeout_task);
2344     n->reassembly_timeout_task = GNUNET_SCHEDULER_add_at (rc->reassembly_timeout,
2345                                                           &reassembly_cleanup_task,
2346                                                           n);
2347     return;
2348   }
2349 }
2350
2351
2352 /**
2353  * function called to #free_reassembly_context().
2354  *
2355  * @param cls NULL
2356  * @param key unused
2357  * @param value a `struct ReassemblyContext` to free
2358  * @return #GNUNET_OK (continue iteration)
2359  */
2360 static int
2361 free_reassembly_cb (void *cls,
2362                     const struct GNUNET_ShortHashCode *key,
2363                     void *value)
2364 {
2365   struct ReassemblyContext *rc = value;
2366   (void) cls;
2367   (void) key;
2368
2369   free_reassembly_context (rc);
2370   return GNUNET_OK;
2371 }
2372
2373
2374 /**
2375  * Release memory used by @a neighbour.
2376  *
2377  * @param neighbour neighbour entry to free
2378  */
2379 static void
2380 free_neighbour (struct Neighbour *neighbour)
2381 {
2382   struct DistanceVectorHop *dvh;
2383
2384   GNUNET_assert (NULL == neighbour->queue_head);
2385   GNUNET_assert (GNUNET_YES ==
2386                  GNUNET_CONTAINER_multipeermap_remove (neighbours,
2387                                                        &neighbour->pid,
2388                                                        neighbour));
2389   if (NULL != neighbour->timeout_task)
2390     GNUNET_SCHEDULER_cancel (neighbour->timeout_task);
2391   if (NULL != neighbour->reassembly_map)
2392   {
2393     GNUNET_CONTAINER_multishortmap_iterate (neighbour->reassembly_map,
2394                                             &free_reassembly_cb,
2395                                             NULL);
2396     GNUNET_CONTAINER_multishortmap_destroy (neighbour->reassembly_map);
2397     neighbour->reassembly_map = NULL;
2398     GNUNET_CONTAINER_heap_destroy (neighbour->reassembly_heap);
2399     neighbour->reassembly_heap = NULL;
2400   }
2401   while (NULL != (dvh = neighbour->dv_head))
2402   {
2403     struct DistanceVector *dv = dvh->dv;
2404
2405     free_distance_vector_hop (dvh);
2406     if (NULL == dv->dv_head)
2407       free_dv_route (dv);
2408   }
2409   if (NULL != neighbour->reassembly_timeout_task)
2410     GNUNET_SCHEDULER_cancel (neighbour->reassembly_timeout_task);
2411   GNUNET_free (neighbour);
2412 }
2413
2414
2415 /**
2416  * Send message to CORE clients that we lost a connection.
2417  *
2418  * @param tc client to inform (must be CORE client)
2419  * @param pid peer the connection is for
2420  * @param quota_out current quota for the peer
2421  */
2422 static void
2423 core_send_connect_info (struct TransportClient *tc,
2424                         const struct GNUNET_PeerIdentity *pid,
2425                         struct GNUNET_BANDWIDTH_Value32NBO quota_out)
2426 {
2427   struct GNUNET_MQ_Envelope *env;
2428   struct ConnectInfoMessage *cim;
2429
2430   GNUNET_assert (CT_CORE == tc->type);
2431   env = GNUNET_MQ_msg (cim,
2432                        GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
2433   cim->quota_out = quota_out;
2434   cim->id = *pid;
2435   GNUNET_MQ_send (tc->mq,
2436                   env);
2437 }
2438
2439
2440 /**
2441  * Send message to CORE clients that we gained a connection
2442  *
2443  * @param pid peer the queue was for
2444  * @param quota_out current quota for the peer
2445  */
2446 static void
2447 cores_send_connect_info (const struct GNUNET_PeerIdentity *pid,
2448                          struct GNUNET_BANDWIDTH_Value32NBO quota_out)
2449 {
2450   for (struct TransportClient *tc = clients_head;
2451        NULL != tc;
2452        tc = tc->next)
2453   {
2454     if (CT_CORE != tc->type)
2455       continue;
2456     core_send_connect_info (tc,
2457                             pid,
2458                             quota_out);
2459   }
2460 }
2461
2462
2463 /**
2464  * Send message to CORE clients that we lost a connection.
2465  *
2466  * @param pid peer the connection was for
2467  */
2468 static void
2469 cores_send_disconnect_info (const struct GNUNET_PeerIdentity *pid)
2470 {
2471   for (struct TransportClient *tc = clients_head;
2472        NULL != tc;
2473        tc = tc->next)
2474   {
2475     struct GNUNET_MQ_Envelope *env;
2476     struct DisconnectInfoMessage *dim;
2477
2478     if (CT_CORE != tc->type)
2479       continue;
2480     env = GNUNET_MQ_msg (dim,
2481                          GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
2482     dim->peer = *pid;
2483     GNUNET_MQ_send (tc->mq,
2484                     env);
2485   }
2486 }
2487
2488
2489 /**
2490  * We believe we are ready to transmit a message on a queue. Double-checks
2491  * with the queue's "tracker_out" and then gives the message to the
2492  * communicator for transmission (updating the tracker, and re-scheduling
2493  * itself if applicable).
2494  *
2495  * @param cls the `struct Queue` to process transmissions for
2496  */
2497 static void
2498 transmit_on_queue (void *cls);
2499
2500
2501 /**
2502  * Schedule next run of #transmit_on_queue().  Does NOTHING if
2503  * we should run immediately or if the message queue is empty.
2504  * Test for no task being added AND queue not being empty to
2505  * transmit immediately afterwards!  This function must only
2506  * be called if the message queue is non-empty!
2507  *
2508  * @param queue the queue to do scheduling for
2509  */
2510 static void
2511 schedule_transmit_on_queue (struct Queue *queue)
2512 {
2513   struct Neighbour *n = queue->neighbour;
2514   struct PendingMessage *pm = n->pending_msg_head;
2515   struct GNUNET_TIME_Relative out_delay;
2516   unsigned int wsize;
2517
2518   GNUNET_assert (NULL != pm);
2519   if (queue->tc->details.communicator.total_queue_length >=
2520       COMMUNICATOR_TOTAL_QUEUE_LIMIT)
2521   {
2522     GNUNET_STATISTICS_update (GST_stats,
2523                               "# Transmission throttled due to communicator queue limit",
2524                               1,
2525                               GNUNET_NO);
2526     return;
2527   }
2528   if (queue->queue_length >= QUEUE_LENGTH_LIMIT)
2529   {
2530     GNUNET_STATISTICS_update (GST_stats,
2531                               "# Transmission throttled due to queue queue limit",
2532                               1,
2533                               GNUNET_NO);
2534     return;
2535   }
2536
2537   wsize = (0 == queue->mtu)
2538     ? pm->bytes_msg /* FIXME: add overheads? */
2539     : queue->mtu;
2540   out_delay = GNUNET_BANDWIDTH_tracker_get_delay (&queue->tracker_out,
2541                                                   wsize);
2542   out_delay = GNUNET_TIME_relative_max (GNUNET_TIME_absolute_get_remaining (pm->next_attempt),
2543                                         out_delay);
2544   if (0 == out_delay.rel_value_us)
2545     return; /* we should run immediately! */
2546   /* queue has changed since we were scheduled, reschedule again */
2547   queue->transmit_task
2548     = GNUNET_SCHEDULER_add_delayed (out_delay,
2549                                     &transmit_on_queue,
2550                                     queue);
2551   if (out_delay.rel_value_us > DELAY_WARN_THRESHOLD.rel_value_us)
2552     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2553                 "Next transmission on queue `%s' in %s (high delay)\n",
2554                 queue->address,
2555                 GNUNET_STRINGS_relative_time_to_string (out_delay,
2556                                                         GNUNET_YES));
2557   else
2558     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2559                 "Next transmission on queue `%s' in %s\n",
2560                 queue->address,
2561                 GNUNET_STRINGS_relative_time_to_string (out_delay,
2562                                                         GNUNET_YES));
2563 }
2564
2565
2566 /**
2567  * Check whether the CORE visibility of @a n changed. If so,
2568  * check whether we need to notify CORE.
2569  *
2570  * @param n neighbour to perform the check for
2571  */
2572 static void
2573 update_neighbour_core_visibility (struct Neighbour *n);
2574
2575
2576 /**
2577  * Free @a queue.
2578  *
2579  * @param queue the queue to free
2580  */
2581 static void
2582 free_queue (struct Queue *queue)
2583 {
2584   struct Neighbour *neighbour = queue->neighbour;
2585   struct TransportClient *tc = queue->tc;
2586   struct MonitorEvent me = {
2587     .cs = GNUNET_TRANSPORT_CS_DOWN,
2588     .rtt = GNUNET_TIME_UNIT_FOREVER_REL
2589   };
2590   struct QueueEntry *qe;
2591   int maxxed;
2592
2593   if (NULL != queue->transmit_task)
2594   {
2595     GNUNET_SCHEDULER_cancel (queue->transmit_task);
2596     queue->transmit_task = NULL;
2597   }
2598   if (NULL != queue->visibility_task)
2599   {
2600     GNUNET_SCHEDULER_cancel (queue->visibility_task);
2601     queue->visibility_task = NULL;
2602   }
2603   GNUNET_CONTAINER_MDLL_remove (neighbour,
2604                                 neighbour->queue_head,
2605                                 neighbour->queue_tail,
2606                                 queue);
2607   GNUNET_CONTAINER_MDLL_remove (client,
2608                                 tc->details.communicator.queue_head,
2609                                 tc->details.communicator.queue_tail,
2610                                 queue);
2611   maxxed = (COMMUNICATOR_TOTAL_QUEUE_LIMIT >= tc->details.communicator.total_queue_length);
2612   while (NULL != (qe = queue->queue_head))
2613   {
2614     GNUNET_CONTAINER_DLL_remove (queue->queue_head,
2615                                  queue->queue_tail,
2616                                  qe);
2617     queue->queue_length--;
2618     tc->details.communicator.total_queue_length--;
2619     GNUNET_free (qe);
2620   }
2621   GNUNET_assert (0 == queue->queue_length);
2622   if ( (maxxed) &&
2623        (COMMUNICATOR_TOTAL_QUEUE_LIMIT < tc->details.communicator.total_queue_length) )
2624   {
2625     /* Communicator dropped below threshold, resume all queues */
2626     GNUNET_STATISTICS_update (GST_stats,
2627                               "# Transmission throttled due to communicator queue limit",
2628                               -1,
2629                               GNUNET_NO);
2630     for (struct Queue *s = tc->details.communicator.queue_head;
2631          NULL != s;
2632          s = s->next_client)
2633       schedule_transmit_on_queue (s);
2634   }
2635   notify_monitors (&neighbour->pid,
2636                    queue->address,
2637                    queue->nt,
2638                    &me);
2639   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_in);
2640   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_out);
2641   GNUNET_free (queue);
2642
2643   update_neighbour_core_visibility (neighbour);
2644     cores_send_disconnect_info (&neighbour->pid);
2645
2646   if (NULL == neighbour->queue_head)
2647   {
2648     free_neighbour (neighbour);
2649   }
2650 }
2651
2652
2653 /**
2654  * Free @a ale
2655  *
2656  * @param ale address list entry to free
2657  */
2658 static void
2659 free_address_list_entry (struct AddressListEntry *ale)
2660 {
2661   struct TransportClient *tc = ale->tc;
2662
2663   GNUNET_CONTAINER_DLL_remove (tc->details.communicator.addr_head,
2664                                tc->details.communicator.addr_tail,
2665                                ale);
2666   if (NULL != ale->sc)
2667   {
2668     GNUNET_PEERSTORE_store_cancel (ale->sc);
2669     ale->sc = NULL;
2670   }
2671   if (NULL != ale->st)
2672   {
2673     GNUNET_SCHEDULER_cancel (ale->st);
2674     ale->st = NULL;
2675   }
2676   GNUNET_free (ale);
2677 }
2678
2679
2680 /**
2681  * Stop the peer request in @a value.
2682  *
2683  * @param cls a `struct TransportClient` that no longer makes the request
2684  * @param pid the peer's identity
2685  * @param value a `struct PeerRequest`
2686  * @return #GNUNET_YES (always)
2687  */
2688 static int
2689 stop_peer_request (void *cls,
2690                    const struct GNUNET_PeerIdentity *pid,
2691                    void *value)
2692 {
2693   struct TransportClient *tc = cls;
2694   struct PeerRequest *pr = value;
2695
2696   GNUNET_PEERSTORE_watch_cancel (pr->wc);
2697   GNUNET_assert (GNUNET_YES ==
2698                  GNUNET_CONTAINER_multipeermap_remove (tc->details.application.requests,
2699                                                        pid,
2700                                                        pr));
2701   GNUNET_free (pr);
2702
2703   return GNUNET_OK;
2704 }
2705
2706
2707 /**
2708  * Called whenever a client is disconnected.  Frees our
2709  * resources associated with that client.
2710  *
2711  * @param cls closure, NULL
2712  * @param client identification of the client
2713  * @param app_ctx our `struct TransportClient`
2714  */
2715 static void
2716 client_disconnect_cb (void *cls,
2717                       struct GNUNET_SERVICE_Client *client,
2718                       void *app_ctx)
2719 {
2720   struct TransportClient *tc = app_ctx;
2721
2722   (void) cls;
2723   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2724               "Client %p disconnected, cleaning up.\n",
2725               tc);
2726   GNUNET_CONTAINER_DLL_remove (clients_head,
2727                                clients_tail,
2728                                tc);
2729   switch (tc->type)
2730   {
2731   case CT_NONE:
2732     break;
2733   case CT_CORE:
2734     {
2735       struct PendingMessage *pm;
2736
2737       while (NULL != (pm = tc->details.core.pending_msg_head))
2738       {
2739         GNUNET_CONTAINER_MDLL_remove (client,
2740                                       tc->details.core.pending_msg_head,
2741                                       tc->details.core.pending_msg_tail,
2742                                       pm);
2743         pm->client = NULL;
2744       }
2745     }
2746     break;
2747   case CT_MONITOR:
2748     break;
2749   case CT_COMMUNICATOR:
2750     {
2751       struct Queue *q;
2752       struct AddressListEntry *ale;
2753
2754       while (NULL != (q = tc->details.communicator.queue_head))
2755         free_queue (q);
2756       while (NULL != (ale = tc->details.communicator.addr_head))
2757         free_address_list_entry (ale);
2758       GNUNET_free (tc->details.communicator.address_prefix);
2759     }
2760     break;
2761   case CT_APPLICATION:
2762     GNUNET_CONTAINER_multipeermap_iterate (tc->details.application.requests,
2763                                            &stop_peer_request,
2764                                            tc);
2765     GNUNET_CONTAINER_multipeermap_destroy (tc->details.application.requests);
2766     break;
2767   }
2768   GNUNET_free (tc);
2769 }
2770
2771
2772 /**
2773  * Iterator telling new CORE client about all existing
2774  * connections to peers.
2775  *
2776  * @param cls the new `struct TransportClient`
2777  * @param pid a connected peer
2778  * @param value the `struct Neighbour` with more information
2779  * @return #GNUNET_OK (continue to iterate)
2780  */
2781 static int
2782 notify_client_connect_info (void *cls,
2783                             const struct GNUNET_PeerIdentity *pid,
2784                             void *value)
2785 {
2786   struct TransportClient *tc = cls;
2787   struct Neighbour *neighbour = value;
2788
2789   core_send_connect_info (tc,
2790                           pid,
2791                           neighbour->quota_out);
2792   return GNUNET_OK;
2793 }
2794
2795
2796 /**
2797  * Initialize a "CORE" client.  We got a start message from this
2798  * client, so add it to the list of clients for broadcasting of
2799  * inbound messages.
2800  *
2801  * @param cls the client
2802  * @param start the start message that was sent
2803  */
2804 static void
2805 handle_client_start (void *cls,
2806                      const struct StartMessage *start)
2807 {
2808   struct TransportClient *tc = cls;
2809   uint32_t options;
2810
2811   options = ntohl (start->options);
2812   if ( (0 != (1 & options)) &&
2813        (0 !=
2814         GNUNET_memcmp (&start->self,
2815                        &GST_my_identity)) )
2816   {
2817     /* client thinks this is a different peer, reject */
2818     GNUNET_break (0);
2819     GNUNET_SERVICE_client_drop (tc->client);
2820     return;
2821   }
2822   if (CT_NONE != tc->type)
2823   {
2824     GNUNET_break (0);
2825     GNUNET_SERVICE_client_drop (tc->client);
2826     return;
2827   }
2828   tc->type = CT_CORE;
2829   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
2830                                          &notify_client_connect_info,
2831                                          tc);
2832   GNUNET_SERVICE_client_continue (tc->client);
2833 }
2834
2835
2836 /**
2837  * Client asked for transmission to a peer.  Process the request.
2838  *
2839  * @param cls the client
2840  * @param obm the send message that was sent
2841  */
2842 static int
2843 check_client_send (void *cls,
2844                    const struct OutboundMessage *obm)
2845 {
2846   struct TransportClient *tc = cls;
2847   uint16_t size;
2848   const struct GNUNET_MessageHeader *obmm;
2849
2850   if (CT_CORE != tc->type)
2851   {
2852     GNUNET_break (0);
2853     return GNUNET_SYSERR;
2854   }
2855   size = ntohs (obm->header.size) - sizeof (struct OutboundMessage);
2856   if (size < sizeof (struct GNUNET_MessageHeader))
2857   {
2858     GNUNET_break (0);
2859     return GNUNET_SYSERR;
2860   }
2861   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
2862   if (size != ntohs (obmm->size))
2863   {
2864     GNUNET_break (0);
2865     return GNUNET_SYSERR;
2866   }
2867   return GNUNET_OK;
2868 }
2869
2870
2871 /**
2872  * Free fragment tree below @e root, excluding @e root itself.
2873  *
2874  * @param root root of the tree to free
2875  */
2876 static void
2877 free_fragment_tree (struct PendingMessage *root)
2878 {
2879   struct PendingMessage *frag;
2880
2881   while (NULL != (frag = root->head_frag))
2882   {
2883     free_fragment_tree (frag);
2884     GNUNET_CONTAINER_MDLL_remove (frag,
2885                                   root->head_frag,
2886                                   root->tail_frag,
2887                                   frag);
2888     GNUNET_free (frag);
2889   }
2890 }
2891
2892
2893 /**
2894  * Release memory associated with @a pm and remove @a pm from associated
2895  * data structures.  @a pm must be a top-level pending message and not
2896  * a fragment in the tree.  The entire tree is freed (if applicable).
2897  *
2898  * @param pm the pending message to free
2899  */
2900 static void
2901 free_pending_message (struct PendingMessage *pm)
2902 {
2903   struct TransportClient *tc = pm->client;
2904   struct Neighbour *target = pm->target;
2905
2906   if (NULL != tc)
2907   {
2908     GNUNET_CONTAINER_MDLL_remove (client,
2909                                   tc->details.core.pending_msg_head,
2910                                   tc->details.core.pending_msg_tail,
2911                                   pm);
2912   }
2913   GNUNET_CONTAINER_MDLL_remove (neighbour,
2914                                 target->pending_msg_head,
2915                                 target->pending_msg_tail,
2916                                 pm);
2917   free_fragment_tree (pm);
2918   GNUNET_free_non_null (pm->bpm);
2919   GNUNET_free (pm);
2920 }
2921
2922
2923 /**
2924  * Send a response to the @a pm that we have processed a
2925  * "send" request with status @a success. We
2926  * transmitted @a bytes_physical on the actual wire.
2927  * Sends a confirmation to the "core" client responsible
2928  * for the original request and free's @a pm.
2929  *
2930  * @param pm handle to the original pending message
2931  * @param success status code, #GNUNET_OK on success, #GNUNET_SYSERR
2932  *          for transmission failure
2933  * @param bytes_physical amount of bandwidth consumed
2934  */
2935 static void
2936 client_send_response (struct PendingMessage *pm,
2937                       int success,
2938                       uint32_t bytes_physical)
2939 {
2940   struct TransportClient *tc = pm->client;
2941   struct Neighbour *target = pm->target;
2942   struct GNUNET_MQ_Envelope *env;
2943   struct SendOkMessage *som;
2944
2945   if (NULL != tc)
2946   {
2947     env = GNUNET_MQ_msg (som,
2948                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
2949     som->success = htonl ((uint32_t) success);
2950     som->bytes_msg = htons (pm->bytes_msg);
2951     som->bytes_physical = htonl (bytes_physical);
2952     som->peer = target->pid;
2953     GNUNET_MQ_send (tc->mq,
2954                     env);
2955   }
2956   free_pending_message (pm);
2957 }
2958
2959
2960 /**
2961  * Checks the message queue for a neighbour for messages that have timed
2962  * out and purges them.
2963  *
2964  * @param cls a `struct Neighbour`
2965  */
2966 static void
2967 check_queue_timeouts (void *cls)
2968 {
2969   struct Neighbour *n = cls;
2970   struct PendingMessage *pm;
2971   struct GNUNET_TIME_Absolute now;
2972   struct GNUNET_TIME_Absolute earliest_timeout;
2973
2974   n->timeout_task = NULL;
2975   earliest_timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
2976   now = GNUNET_TIME_absolute_get ();
2977   for (struct PendingMessage *pos = n->pending_msg_head;
2978        NULL != pos;
2979        pos = pm)
2980   {
2981     pm = pos->next_neighbour;
2982     if (pos->timeout.abs_value_us <= now.abs_value_us)
2983     {
2984       GNUNET_STATISTICS_update (GST_stats,
2985                                 "# messages dropped (timeout before confirmation)",
2986                                 1,
2987                                 GNUNET_NO);
2988       client_send_response (pm,
2989                             GNUNET_NO,
2990                             0);
2991       continue;
2992     }
2993     earliest_timeout = GNUNET_TIME_absolute_min (earliest_timeout,
2994                                                  pos->timeout);
2995   }
2996   n->earliest_timeout = earliest_timeout;
2997   if (NULL != n->pending_msg_head)
2998     n->timeout_task = GNUNET_SCHEDULER_add_at (earliest_timeout,
2999                                                &check_queue_timeouts,
3000                                                n);
3001 }
3002
3003
3004 /**
3005  * Client asked for transmission to a peer.  Process the request.
3006  *
3007  * @param cls the client
3008  * @param obm the send message that was sent
3009  */
3010 static void
3011 handle_client_send (void *cls,
3012                     const struct OutboundMessage *obm)
3013 {
3014   struct TransportClient *tc = cls;
3015   struct PendingMessage *pm;
3016   const struct GNUNET_MessageHeader *obmm;
3017   struct Neighbour *target;
3018   uint32_t bytes_msg;
3019   int was_empty;
3020
3021   GNUNET_assert (CT_CORE == tc->type);
3022   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
3023   bytes_msg = ntohs (obmm->size);
3024   target = lookup_neighbour (&obm->peer);
3025   if (NULL == target)
3026   {
3027     /* Failure: don't have this peer as a neighbour (anymore).
3028        Might have gone down asynchronously, so this is NOT
3029        a protocol violation by CORE. Still count the event,
3030        as this should be rare. */
3031     struct GNUNET_MQ_Envelope *env;
3032     struct SendOkMessage *som;
3033
3034     env = GNUNET_MQ_msg (som,
3035                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
3036     som->success = htonl (GNUNET_SYSERR);
3037     som->bytes_msg = htonl (bytes_msg);
3038     som->bytes_physical = htonl (0);
3039     som->peer = obm->peer;
3040     GNUNET_MQ_send (tc->mq,
3041                     env);
3042     GNUNET_SERVICE_client_continue (tc->client);
3043     GNUNET_STATISTICS_update (GST_stats,
3044                               "# messages dropped (neighbour unknown)",
3045                               1,
3046                               GNUNET_NO);
3047     return;
3048   }
3049   was_empty = (NULL == target->pending_msg_head);
3050   pm = GNUNET_malloc (sizeof (struct PendingMessage) + bytes_msg);
3051   pm->client = tc;
3052   pm->target = target;
3053   pm->bytes_msg = bytes_msg;
3054   pm->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (obm->timeout));
3055   memcpy (&pm[1],
3056           &obm[1],
3057           bytes_msg);
3058   GNUNET_CONTAINER_MDLL_insert (neighbour,
3059                                 target->pending_msg_head,
3060                                 target->pending_msg_tail,
3061                                 pm);
3062   GNUNET_CONTAINER_MDLL_insert (client,
3063                                 tc->details.core.pending_msg_head,
3064                                 tc->details.core.pending_msg_tail,
3065                                 pm);
3066   if (target->earliest_timeout.abs_value_us > pm->timeout.abs_value_us)
3067   {
3068     target->earliest_timeout.abs_value_us = pm->timeout.abs_value_us;
3069     if (NULL != target->timeout_task)
3070       GNUNET_SCHEDULER_cancel (target->timeout_task);
3071     target->timeout_task
3072       = GNUNET_SCHEDULER_add_at (target->earliest_timeout,
3073                                  &check_queue_timeouts,
3074                                  target);
3075   }
3076   if (! was_empty)
3077     return; /* all queues must already be busy */
3078   for (struct Queue *queue = target->queue_head;
3079        NULL != queue;
3080        queue = queue->next_neighbour)
3081   {
3082     /* try transmission on any queue that is idle */
3083     if (NULL == queue->transmit_task)
3084       queue->transmit_task = GNUNET_SCHEDULER_add_now (&transmit_on_queue,
3085                                                        queue);
3086   }
3087 }
3088
3089
3090 /**
3091  * Communicator started.  Test message is well-formed.
3092  *
3093  * @param cls the client
3094  * @param cam the send message that was sent
3095  */
3096 static int
3097 check_communicator_available (void *cls,
3098                               const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
3099 {
3100   struct TransportClient *tc = cls;
3101   uint16_t size;
3102
3103   if (CT_NONE != tc->type)
3104   {
3105     GNUNET_break (0);
3106     return GNUNET_SYSERR;
3107   }
3108   tc->type = CT_COMMUNICATOR;
3109   size = ntohs (cam->header.size) - sizeof (*cam);
3110   if (0 == size)
3111     return GNUNET_OK; /* receive-only communicator */
3112   GNUNET_MQ_check_zero_termination (cam);
3113   return GNUNET_OK;
3114 }
3115
3116
3117 /**
3118  * Communicator started.  Process the request.
3119  *
3120  * @param cls the client
3121  * @param cam the send message that was sent
3122  */
3123 static void
3124 handle_communicator_available (void *cls,
3125                                const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
3126 {
3127   struct TransportClient *tc = cls;
3128   uint16_t size;
3129
3130   size = ntohs (cam->header.size) - sizeof (*cam);
3131   if (0 == size)
3132     return; /* receive-only communicator */
3133   tc->details.communicator.address_prefix
3134     = GNUNET_strdup ((const char *) &cam[1]);
3135   tc->details.communicator.cc
3136     = (enum GNUNET_TRANSPORT_CommunicatorCharacteristics) ntohl (cam->cc);
3137   GNUNET_SERVICE_client_continue (tc->client);
3138 }
3139
3140
3141 /**
3142  * Communicator requests backchannel transmission.  Check the request.
3143  *
3144  * @param cls the client
3145  * @param cb the send message that was sent
3146  * @return #GNUNET_OK if message is well-formed
3147  */
3148 static int
3149 check_communicator_backchannel (void *cls,
3150                                 const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
3151 {
3152   const struct GNUNET_MessageHeader *inbox;
3153   const char *is;
3154   uint16_t msize;
3155   uint16_t isize;
3156
3157   (void) cls;
3158   msize = ntohs (cb->header.size) - sizeof (*cb);
3159   if (UINT16_MAX - msize >
3160       sizeof (struct TransportBackchannelEncapsulationMessage) +
3161       sizeof (struct TransportBackchannelRequestPayload) )
3162   {
3163     GNUNET_break (0);
3164     return GNUNET_SYSERR;
3165   }
3166   inbox = (const struct GNUNET_MessageHeader *) &cb[1];
3167   isize = ntohs (inbox->size);
3168   if (isize >= msize)
3169   {
3170     GNUNET_break (0);
3171     return GNUNET_SYSERR;
3172   }
3173   is = (const char *) inbox;
3174   is += isize;
3175   msize -= isize;
3176   GNUNET_assert (msize > 0);
3177   if ('\0' != is[msize-1])
3178   {
3179     GNUNET_break (0);
3180     return GNUNET_SYSERR;
3181   }
3182   return GNUNET_OK;
3183 }
3184
3185
3186 /**
3187  * Remove memory used by expired ephemeral keys.
3188  *
3189  * @param cls NULL
3190  */
3191 static void
3192 expire_ephemerals (void *cls)
3193 {
3194   struct EphemeralCacheEntry *ece;
3195
3196   (void) cls;
3197   ephemeral_task = NULL;
3198   while (NULL != (ece = GNUNET_CONTAINER_heap_peek (ephemeral_heap)))
3199   {
3200     if (0 == GNUNET_TIME_absolute_get_remaining (ece->ephemeral_validity).rel_value_us)
3201     {
3202       free_ephemeral (ece);
3203       continue;
3204     }
3205     ephemeral_task = GNUNET_SCHEDULER_add_at (ece->ephemeral_validity,
3206                                               &expire_ephemerals,
3207                                               NULL);
3208     return;
3209   }
3210 }
3211
3212
3213 /**
3214  * Lookup ephemeral key in our #ephemeral_map. If no valid one exists, generate
3215  * one, cache it and return it.
3216  *
3217  * @param pid peer to look up ephemeral for
3218  * @param private_key[out] set to the private key
3219  * @param ephemeral_key[out] set to the key
3220  * @param ephemeral_sender_sig[out] set to the signature
3221  * @param ephemeral_validity[out] set to the validity expiration time
3222  */
3223 static void
3224 lookup_ephemeral (const struct GNUNET_PeerIdentity *pid,
3225                   struct GNUNET_CRYPTO_EcdhePrivateKey *private_key,
3226                   struct GNUNET_CRYPTO_EcdhePublicKey *ephemeral_key,
3227                   struct GNUNET_CRYPTO_EddsaSignature *ephemeral_sender_sig,
3228                   struct GNUNET_TIME_Absolute *ephemeral_validity)
3229 {
3230   struct EphemeralCacheEntry *ece;
3231   struct EphemeralConfirmation ec;
3232
3233   ece = GNUNET_CONTAINER_multipeermap_get (ephemeral_map,
3234                                            pid);
3235   if ( (NULL != ece) &&
3236        (0 == GNUNET_TIME_absolute_get_remaining (ece->ephemeral_validity).rel_value_us) )
3237   {
3238     free_ephemeral (ece);
3239     ece = NULL;
3240   }
3241   if (NULL == ece)
3242   {
3243     ece = GNUNET_new (struct EphemeralCacheEntry);
3244     ece->target = *pid;
3245     ece->ephemeral_validity = GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get_monotonic (GST_cfg),
3246                                                         EPHEMERAL_VALIDITY);
3247     GNUNET_assert (GNUNET_OK ==
3248                    GNUNET_CRYPTO_ecdhe_key_create2 (&ece->private_key));
3249     GNUNET_CRYPTO_ecdhe_key_get_public (&ece->private_key,
3250                                         &ece->ephemeral_key);
3251     ec.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL);
3252     ec.purpose.size = htonl (sizeof (ec));
3253     ec.target = *pid;
3254     ec.ephemeral_key = ece->ephemeral_key;
3255     GNUNET_assert (GNUNET_OK ==
3256                    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
3257                                              &ec.purpose,
3258                                              &ece->sender_sig));
3259     ece->hn = GNUNET_CONTAINER_heap_insert (ephemeral_heap,
3260                                             ece,
3261                                             ece->ephemeral_validity.abs_value_us);
3262     GNUNET_assert (GNUNET_OK ==
3263                    GNUNET_CONTAINER_multipeermap_put (ephemeral_map,
3264                                                       &ece->target,
3265                                                       ece,
3266                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
3267     if (NULL == ephemeral_task)
3268       ephemeral_task = GNUNET_SCHEDULER_add_at (ece->ephemeral_validity,
3269                                                 &expire_ephemerals,
3270                                                 NULL);
3271   }
3272   *private_key = ece->private_key;
3273   *ephemeral_key = ece->ephemeral_key;
3274   *ephemeral_sender_sig = ece->sender_sig;
3275   *ephemeral_validity = ece->ephemeral_validity;
3276 }
3277
3278
3279 /**
3280  * Send the control message @a payload on @a queue.
3281  *
3282  * @param queue the queue to use for transmission
3283  * @param pm pending message to update once transmission is done, may be NULL!
3284  * @param payload the payload to send (encapsulated in a
3285  *        #GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG).
3286  * @param payload_size number of bytes in @a payload
3287  */
3288 static void
3289 queue_send_msg (struct Queue *queue,
3290                 struct PendingMessage *pm,
3291                 const void *payload,
3292                 size_t payload_size)
3293 {
3294   struct Neighbour *n = queue->neighbour;
3295   struct GNUNET_TRANSPORT_SendMessageTo *smt;
3296   struct GNUNET_MQ_Envelope *env;
3297
3298   env = GNUNET_MQ_msg_extra (smt,
3299                              payload_size,
3300                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
3301   smt->qid = queue->qid;
3302   smt->mid = queue->mid_gen;
3303   smt->receiver = n->pid;
3304   memcpy (&smt[1],
3305           payload,
3306           payload_size);
3307   {
3308     /* Pass the env to the communicator of queue for transmission. */
3309     struct QueueEntry *qe;
3310
3311     qe = GNUNET_new (struct QueueEntry);
3312     qe->mid = queue->mid_gen++;
3313     qe->queue = queue;
3314     // qe->pm = pm; // FIXME: not so easy, reference management on 'free(s)'!
3315     // (also, note that pm may be NULL!)
3316     GNUNET_CONTAINER_DLL_insert (queue->queue_head,
3317                                  queue->queue_tail,
3318                                  qe);
3319     GNUNET_assert (CT_COMMUNICATOR == queue->tc->type);
3320     queue->queue_length++;
3321     queue->tc->details.communicator.total_queue_length++;
3322     GNUNET_MQ_send (queue->tc->mq,
3323                     env);
3324   }
3325 }
3326
3327
3328 /**
3329  * Which transmission options are allowable for transmission?
3330  * Interpreted bit-wise!
3331  */
3332 enum RouteMessageOptions {
3333   /**
3334    * Only confirmed, non-DV direct neighbours.
3335    */
3336   RMO_NONE = 0,
3337
3338   /**
3339    * We are allowed to use DV routing for this @a hdr
3340    */
3341   RMO_DV_ALLOWED = 1,
3342
3343   /**
3344    * We are allowed to use unconfirmed queues or DV routes for this message
3345    */
3346   RMO_UNCONFIRMED_ALLOWED = 2,
3347
3348   /**
3349    * Reliable and unreliable, DV and non-DV are all acceptable.
3350    */
3351   RMO_ANYTHING_GOES = (RMO_DV_ALLOWED | RMO_UNCONFIRMED_ALLOWED),
3352
3353   /**
3354    * If we have multiple choices, it is OK to send this message
3355    * over multiple channels at the same time to improve loss tolerance.
3356    * (We do at most 2 transmissions.)
3357    */
3358   RMO_REDUNDANT = 4
3359 };
3360
3361
3362 /**
3363  * Pick a queue of @a n under constraints @a options and schedule
3364  * transmission of @a hdr.
3365  *
3366  * @param n neighbour to send to
3367  * @param hdr message to send as payload
3368  * @param options whether queues must be confirmed or not,
3369  *        and whether we may pick multiple (2) queues
3370  */
3371 static void
3372 route_via_neighbour (const struct Neighbour *n,
3373                      const struct GNUNET_MessageHeader *hdr,
3374                      enum RouteMessageOptions options)
3375 {
3376   // FIXME: pick on or two 'random' queue (under constraints of options)
3377   // Then add wrapper and enqueue message!
3378 }
3379
3380
3381 /**
3382  * Pick a path of @a dv under constraints @a options and schedule
3383  * transmission of @a hdr.
3384  *
3385  * @param n neighbour to send to
3386  * @param hdr message to send as payload
3387  * @param options whether path must be confirmed or not
3388  *        and whether we may pick multiple (2) paths
3389  */
3390 static void
3391 route_via_dv (const struct DistanceVector *dv,
3392               const struct GNUNET_MessageHeader *hdr,
3393               enum RouteMessageOptions options)
3394 {
3395   // FIXME: pick on or two 'random' paths (under constraints of options)
3396   // Then add DVBox and enqueue message (possibly using
3397   // route_via_neighbour for 1st hop?)
3398 }
3399
3400
3401 /**
3402  * We need to transmit @a hdr to @a target.  If necessary, this may
3403  * involve DV routing.
3404  *
3405  * @param target peer to receive @a hdr
3406  * @param hdr header of the message to route and #GNUNET_free()
3407  * @param options which transmission channels are allowed
3408  */
3409 static void
3410 route_message (const struct GNUNET_PeerIdentity *target,
3411                struct GNUNET_MessageHeader *hdr,
3412                enum RouteMessageOptions options)
3413 {
3414   struct Neighbour *n;
3415   struct DistanceVector *dv;
3416
3417   n = GNUNET_CONTAINER_multipeermap_get (neighbours,
3418                                          target);
3419   dv = (0 != (options & RMO_DV_ALLOWED))
3420     ? GNUNET_CONTAINER_multipeermap_get (dv_routes,
3421                                          target)
3422     : NULL;
3423   if (0 == (options & RMO_UNCONFIRMED_ALLOWED))
3424   {
3425     /* if confirmed is required, and we do not have anything
3426        confirmed, drop respective options */
3427     if ( (NULL != n) &&
3428          (GNUNET_NO == n->core_visible) )
3429       n = NULL;
3430     if ( (NULL != dv) &&
3431          (GNUNET_NO == dv->core_visible) )
3432       dv = NULL;
3433   }
3434   if ( (NULL == n) &&
3435        (NULL == dv) )
3436   {
3437     GNUNET_STATISTICS_update (GST_stats,
3438                               "# Messages dropped in routing: no acceptable method",
3439                               1,
3440                               GNUNET_NO);
3441     GNUNET_free (hdr);
3442     return;
3443   }
3444   /* If both dv and n are possible and we must choose:
3445      flip a coin for the choice between the two; for now 50/50 */
3446   if ( (NULL != n) &&
3447        (NULL != dv) &&
3448        (0 == (options & RMO_REDUNDANT)) )
3449   {
3450     if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 2))
3451       n = NULL;
3452     else
3453       dv = NULL;
3454   }
3455   if ( (NULL != n) &&
3456        (NULL != dv) )
3457     options &= ~RMO_REDUNDANT; /* We will do one DV and one direct, that's
3458                                   enough for redunancy, so clear the flag. */
3459   if (NULL != n)
3460   {
3461     route_via_neighbour (n,
3462                          hdr,
3463                          options);
3464   }
3465   if (NULL != dv)
3466   {
3467     route_via_dv (dv,
3468                   hdr,
3469                   options);
3470   }
3471   GNUNET_free (hdr);
3472 }
3473
3474
3475 /**
3476  * Structure of the key material used to encrypt backchannel messages.
3477  */
3478 struct BackchannelKeyState
3479 {
3480   /**
3481    * State of our block cipher.
3482    */
3483   gcry_cipher_hd_t cipher;
3484
3485   /**
3486    * Actual key material.
3487    */
3488   struct {
3489
3490     /**
3491      * Key used for HMAC calculations (via #GNUNET_CRYPTO_hmac()).
3492      */
3493     struct GNUNET_CRYPTO_AuthKey hmac_key;
3494
3495     /**
3496      * Symmetric key to use for encryption.
3497      */
3498     char aes_key[256/8];
3499
3500     /**
3501      * Counter value to use during setup.
3502      */
3503     char aes_ctr[128/8];
3504
3505   } material;
3506 };
3507
3508
3509 static void
3510 bc_setup_key_state_from_km (const struct GNUNET_HashCode *km,
3511                             const struct GNUNET_ShortHashCode *iv,
3512                             struct BackchannelKeyState *key)
3513 {
3514   /* must match #dh_key_derive_eph_pub */
3515   GNUNET_assert (GNUNET_YES ==
3516                  GNUNET_CRYPTO_kdf (&key->material,
3517                                     sizeof (key->material),
3518                                     "transport-backchannel-key",
3519                                     strlen ("transport-backchannel-key"),
3520                                     &km,
3521                                     sizeof (km),
3522                                     iv,
3523                                     sizeof (*iv)));
3524   gcry_cipher_open (&key->cipher,
3525                     GCRY_CIPHER_AES256 /* low level: go for speed */,
3526                     GCRY_CIPHER_MODE_CTR,
3527                     0 /* flags */);
3528   gcry_cipher_setkey (key->cipher,
3529                       &key->material.aes_key,
3530                       sizeof (key->material.aes_key));
3531   gcry_cipher_setctr (key->cipher,
3532                       &key->material.aes_ctr,
3533                       sizeof (key->material.aes_ctr));
3534 }
3535
3536
3537 /**
3538  * Derive backchannel encryption key material from @a priv_ephemeral
3539  * and @a target and @a iv.
3540  *
3541  * @param priv_ephemeral ephemeral private key to use
3542  * @param target the target peer to encrypt to
3543  * @param iv unique IV to use
3544  * @param key[out] set to the key material
3545  */
3546 static void
3547 dh_key_derive_eph_pid (const struct GNUNET_CRYPTO_EcdhePrivateKey *priv_ephemeral,
3548                        const struct GNUNET_PeerIdentity *target,
3549                        const struct GNUNET_ShortHashCode *iv,
3550                        struct BackchannelKeyState *key)
3551 {
3552   struct GNUNET_HashCode km;
3553
3554   GNUNET_assert (GNUNET_YES ==
3555                  GNUNET_CRYPTO_ecdh_eddsa (priv_ephemeral,
3556                                            &target->public_key,
3557                                            &km));
3558   bc_setup_key_state_from_km (&km,
3559                               iv,
3560                               key);
3561 }
3562
3563
3564 /**
3565  * Derive backchannel encryption key material from #GST_my_private_key
3566  * and @a pub_ephemeral and @a iv.
3567  *
3568  * @param priv_ephemeral ephemeral private key to use
3569  * @param target the target peer to encrypt to
3570  * @param iv unique IV to use
3571  * @param key[out] set to the key material
3572  */
3573 static void
3574 dh_key_derive_eph_pub (const struct GNUNET_CRYPTO_EcdhePublicKey *pub_ephemeral,
3575                        const struct GNUNET_ShortHashCode *iv,
3576                        struct BackchannelKeyState *key)
3577 {
3578   struct GNUNET_HashCode km;
3579
3580   GNUNET_assert (GNUNET_YES ==
3581                  GNUNET_CRYPTO_eddsa_ecdh (GST_my_private_key,
3582                                            pub_ephemeral,
3583                                            &km));
3584   bc_setup_key_state_from_km (&km,
3585                               iv,
3586                               key);
3587 }
3588
3589
3590 /**
3591  * Do HMAC calculation for backchannel messages over @a data using key
3592  * material from @a key.
3593  *
3594  * @param key key material (from DH)
3595  * @param hmac[out] set to the HMAC
3596  * @param data data to perform HMAC calculation over
3597  * @param data_size number of bytes in @a data
3598  */
3599 static void
3600 bc_hmac (const struct BackchannelKeyState *key,
3601          struct GNUNET_HashCode *hmac,
3602          const void *data,
3603          size_t data_size)
3604 {
3605   GNUNET_CRYPTO_hmac (&key->material.hmac_key,
3606                       data,
3607                       data_size,
3608                       hmac);
3609 }
3610
3611
3612 /**
3613  * Perform backchannel encryption using symmetric secret in @a key
3614  * to encrypt data from @a in to @a dst.
3615  *
3616  * @param key[in,out] key material to use
3617  * @param dst where to write the result
3618  * @param in input data to encrypt (plaintext)
3619  * @param in_size number of bytes of input in @a in and available at @a dst
3620  */
3621 static void
3622 bc_encrypt (struct BackchannelKeyState *key,
3623             const void *in,
3624             void *dst,
3625             size_t in_size)
3626 {
3627   GNUNET_assert (0 ==
3628                  gcry_cipher_encrypt (key->cipher,
3629                                       dst,
3630                                       in_size,
3631                                       in,
3632                                       in_size));
3633 }
3634
3635
3636 /**
3637  * Perform backchannel encryption using symmetric secret in @a key
3638  * to encrypt data from @a in to @a dst.
3639  *
3640  * @param key[in,out] key material to use
3641  * @param ciph cipher text to decrypt
3642  * @param out[out] output data to generate (plaintext)
3643  * @param out_size number of bytes of input in @a ciph and available in @a out
3644  */
3645 static void
3646 bc_decrypt (struct BackchannelKeyState *key,
3647             void *out,
3648             const void *ciph,
3649             size_t out_size)
3650 {
3651   GNUNET_assert (0 ==
3652                  gcry_cipher_decrypt (key->cipher,
3653                                       out,
3654                                       out_size,
3655                                       ciph,
3656                                       out_size));
3657 }
3658
3659
3660 /**
3661  * Clean up key material in @a key.
3662  *
3663  * @param key key material to clean up (memory must not be free'd!)
3664  */
3665 static void
3666 bc_key_clean (struct BackchannelKeyState *key)
3667 {
3668   gcry_cipher_close (key->cipher);
3669   GNUNET_CRYPTO_zero_keys (&key->material,
3670                            sizeof (key->material));
3671 }
3672
3673
3674 /**
3675  * Communicator requests backchannel transmission.  Process the request.
3676  *
3677  * @param cls the client
3678  * @param cb the send message that was sent
3679  */
3680 static void
3681 handle_communicator_backchannel (void *cls,
3682                                  const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
3683 {
3684   struct TransportClient *tc = cls;
3685   struct GNUNET_CRYPTO_EcdhePrivateKey private_key;
3686   struct GNUNET_TIME_Absolute ephemeral_validity;
3687   struct TransportBackchannelEncapsulationMessage *enc;
3688   struct TransportBackchannelRequestPayload ppay;
3689   struct BackchannelKeyState key;
3690   char *mpos;
3691   uint16_t msize;
3692
3693   /* encapsulate and encrypt message */
3694   msize = ntohs (cb->header.size) - sizeof (*cb) + sizeof (struct TransportBackchannelRequestPayload);
3695   enc = GNUNET_malloc (sizeof (*enc) + msize);
3696   enc->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION);
3697   enc->header.size = htons (sizeof (*enc) + msize);
3698   enc->target = cb->pid;
3699   lookup_ephemeral (&cb->pid,
3700                     &private_key,
3701                     &enc->ephemeral_key,
3702                     &ppay.sender_sig,
3703                     &ephemeral_validity);
3704   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
3705                               &enc->iv,
3706                               sizeof (enc->iv));
3707   dh_key_derive_eph_pid (&private_key,
3708                          &cb->pid,
3709                          &enc->iv,
3710                          &key);
3711   ppay.ephemeral_validity = GNUNET_TIME_absolute_hton (ephemeral_validity);
3712   ppay.monotonic_time = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (GST_cfg));
3713   mpos = (char *) &enc[1];
3714   bc_encrypt (&key,
3715               &ppay,
3716               mpos,
3717               sizeof (ppay));
3718   bc_encrypt (&key,
3719               &cb[1],
3720               &mpos[sizeof (ppay)],
3721               ntohs (cb->header.size) - sizeof (*cb));
3722   bc_hmac (&key,
3723            &enc->hmac,
3724            mpos,
3725            sizeof (ppay) + ntohs (cb->header.size) - sizeof (*cb));
3726   bc_key_clean (&key);
3727   route_message (&cb->pid,
3728                  &enc->header,
3729                  RMO_DV_ALLOWED);
3730   GNUNET_SERVICE_client_continue (tc->client);
3731 }
3732
3733
3734 /**
3735  * Address of our peer added.  Test message is well-formed.
3736  *
3737  * @param cls the client
3738  * @param aam the send message that was sent
3739  * @return #GNUNET_OK if message is well-formed
3740  */
3741 static int
3742 check_add_address (void *cls,
3743                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
3744 {
3745   struct TransportClient *tc = cls;
3746
3747   if (CT_COMMUNICATOR != tc->type)
3748   {
3749     GNUNET_break (0);
3750     return GNUNET_SYSERR;
3751   }
3752   GNUNET_MQ_check_zero_termination (aam);
3753   return GNUNET_OK;
3754 }
3755
3756
3757 /**
3758  * Ask peerstore to store our address.
3759  *
3760  * @param cls an `struct AddressListEntry *`
3761  */
3762 static void
3763 store_pi (void *cls);
3764
3765
3766 /**
3767  * Function called when peerstore is done storing our address.
3768  *
3769  * @param cls a `struct AddressListEntry`
3770  * @param success #GNUNET_YES if peerstore was successful
3771  */
3772 static void
3773 peerstore_store_own_cb (void *cls,
3774                         int success)
3775 {
3776   struct AddressListEntry *ale = cls;
3777
3778   ale->sc = NULL;
3779   if (GNUNET_YES != success)
3780     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3781                 "Failed to store our own address `%s' in peerstore!\n",
3782                 ale->address);
3783   /* refresh period is 1/4 of expiration time, that should be plenty
3784      without being excessive. */
3785   ale->st = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_divide (ale->expiration,
3786                                                                        4ULL),
3787                                           &store_pi,
3788                                           ale);
3789 }
3790
3791
3792 /**
3793  * Ask peerstore to store our address.
3794  *
3795  * @param cls an `struct AddressListEntry *`
3796  */
3797 static void
3798 store_pi (void *cls)
3799 {
3800   struct AddressListEntry *ale = cls;
3801   void *addr;
3802   size_t addr_len;
3803   struct GNUNET_TIME_Absolute expiration;
3804
3805   ale->st = NULL;
3806   expiration = GNUNET_TIME_relative_to_absolute (ale->expiration);
3807   GNUNET_HELLO_sign_address (ale->address,
3808                              ale->nt,
3809                              expiration,
3810                              GST_my_private_key,
3811                              &addr,
3812                              &addr_len);
3813   ale->sc = GNUNET_PEERSTORE_store (peerstore,
3814                                     "transport",
3815                                     &GST_my_identity,
3816                                     GNUNET_PEERSTORE_TRANSPORT_HELLO_KEY,
3817                                     addr,
3818                                     addr_len,
3819                                     expiration,
3820                                     GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
3821                                     &peerstore_store_own_cb,
3822                                     ale);
3823   GNUNET_free (addr);
3824   if (NULL == ale->sc)
3825   {
3826     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
3827                 "Failed to store our address `%s' with peerstore\n",
3828                 ale->address);
3829     ale->st = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
3830                                             &store_pi,
3831                                             ale);
3832   }
3833 }
3834
3835
3836 /**
3837  * Address of our peer added.  Process the request.
3838  *
3839  * @param cls the client
3840  * @param aam the send message that was sent
3841  */
3842 static void
3843 handle_add_address (void *cls,
3844                     const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
3845 {
3846   struct TransportClient *tc = cls;
3847   struct AddressListEntry *ale;
3848   size_t slen;
3849
3850   slen = ntohs (aam->header.size) - sizeof (*aam);
3851   ale = GNUNET_malloc (sizeof (struct AddressListEntry) + slen);
3852   ale->tc = tc;
3853   ale->address = (const char *) &ale[1];
3854   ale->expiration = GNUNET_TIME_relative_ntoh (aam->expiration);
3855   ale->aid = aam->aid;
3856   ale->nt = (enum GNUNET_NetworkType) ntohl (aam->nt);
3857   memcpy (&ale[1],
3858           &aam[1],
3859           slen);
3860   GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
3861                                tc->details.communicator.addr_tail,
3862                                ale);
3863   ale->st = GNUNET_SCHEDULER_add_now (&store_pi,
3864                                       ale);
3865   GNUNET_SERVICE_client_continue (tc->client);
3866 }
3867
3868
3869 /**
3870  * Address of our peer deleted.  Process the request.
3871  *
3872  * @param cls the client
3873  * @param dam the send message that was sent
3874  */
3875 static void
3876 handle_del_address (void *cls,
3877                     const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
3878 {
3879   struct TransportClient *tc = cls;
3880
3881   if (CT_COMMUNICATOR != tc->type)
3882   {
3883     GNUNET_break (0);
3884     GNUNET_SERVICE_client_drop (tc->client);
3885     return;
3886   }
3887   for (struct AddressListEntry *ale = tc->details.communicator.addr_head;
3888        NULL != ale;
3889        ale = ale->next)
3890   {
3891     if (dam->aid != ale->aid)
3892       continue;
3893     GNUNET_assert (ale->tc == tc);
3894     free_address_list_entry (ale);
3895     GNUNET_SERVICE_client_continue (tc->client);
3896   }
3897   GNUNET_break (0);
3898   GNUNET_SERVICE_client_drop (tc->client);
3899 }
3900
3901
3902 /**
3903  * Context from #handle_incoming_msg().  Closure for many
3904  * message handlers below.
3905  */
3906 struct CommunicatorMessageContext
3907 {
3908   /**
3909    * Which communicator provided us with the message.
3910    */
3911   struct TransportClient *tc;
3912
3913   /**
3914    * Additional information for flow control and about the sender.
3915    */
3916   struct GNUNET_TRANSPORT_IncomingMessage im;
3917
3918   /**
3919    * Number of hops the message has travelled (if DV-routed).
3920    * FIXME: make use of this in ACK handling!
3921    */
3922   uint16_t total_hops;
3923 };
3924
3925
3926 /**
3927  * Given an inbound message @a msg from a communicator @a cmc,
3928  * demultiplex it based on the type calling the right handler.
3929  *
3930  * @param cmc context for demultiplexing
3931  * @param msg message to demultiplex
3932  */
3933 static void
3934 demultiplex_with_cmc (struct CommunicatorMessageContext *cmc,
3935                       const struct GNUNET_MessageHeader *msg);
3936
3937
3938 /**
3939  * Send ACK to communicator (if requested) and free @a cmc.
3940  *
3941  * @param cmc context for which we are done handling the message
3942  */
3943 static void
3944 finish_cmc_handling (struct CommunicatorMessageContext *cmc)
3945 {
3946   if (0 != ntohl (cmc->im.fc_on))
3947   {
3948     /* send ACK when done to communicator for flow control! */
3949     struct GNUNET_MQ_Envelope *env;
3950     struct GNUNET_TRANSPORT_IncomingMessageAck *ack;
3951
3952     env = GNUNET_MQ_msg (ack,
3953                          GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG_ACK);
3954     ack->reserved = htonl (0);
3955     ack->fc_id = cmc->im.fc_id;
3956     ack->sender = cmc->im.sender;
3957     GNUNET_MQ_send (cmc->tc->mq,
3958                     env);
3959   }
3960   GNUNET_SERVICE_client_continue (cmc->tc->client);
3961   GNUNET_free (cmc);
3962 }
3963
3964
3965 /**
3966  * Communicator gave us an unencapsulated message to pass as-is to
3967  * CORE.  Process the request.
3968  *
3969  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
3970  * @param mh the message that was received
3971  */
3972 static void
3973 handle_raw_message (void *cls,
3974                     const struct GNUNET_MessageHeader *mh)
3975 {
3976   struct CommunicatorMessageContext *cmc = cls;
3977   uint16_t size = ntohs (mh->size);
3978
3979   if ( (size > UINT16_MAX - sizeof (struct InboundMessage)) ||
3980        (size < sizeof (struct GNUNET_MessageHeader)) )
3981   {
3982     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
3983
3984     GNUNET_break (0);
3985     finish_cmc_handling (cmc);
3986     GNUNET_SERVICE_client_drop (client);
3987     return;
3988   }
3989   /* Forward to all CORE clients */
3990   for (struct TransportClient *tc = clients_head;
3991        NULL != tc;
3992        tc = tc->next)
3993   {
3994     struct GNUNET_MQ_Envelope *env;
3995     struct InboundMessage *im;
3996
3997     if (CT_CORE != tc->type)
3998       continue;
3999     env = GNUNET_MQ_msg_extra (im,
4000                                size,
4001                                GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
4002     im->peer = cmc->im.sender;
4003     memcpy (&im[1],
4004             mh,
4005             size);
4006     GNUNET_MQ_send (tc->mq,
4007                     env);
4008   }
4009   /* FIXME: consider doing this _only_ once the message
4010      was drained from the CORE MQs to extend flow control to CORE!
4011      (basically, increment counter in cmc, decrement on MQ send continuation! */
4012   finish_cmc_handling (cmc);
4013 }
4014
4015
4016 /**
4017  * Communicator gave us a fragment box.  Check the message.
4018  *
4019  * @param cls a `struct CommunicatorMessageContext`
4020  * @param fb the send message that was sent
4021  * @return #GNUNET_YES if message is well-formed
4022  */
4023 static int
4024 check_fragment_box (void *cls,
4025                     const struct TransportFragmentBox *fb)
4026 {
4027   uint16_t size = ntohs (fb->header.size);
4028   uint16_t bsize = size - sizeof (*fb);
4029
4030   if (0 == bsize)
4031   {
4032     GNUNET_break_op (0);
4033     return GNUNET_SYSERR;
4034   }
4035   if (bsize + ntohs (fb->frag_off) > ntohs (fb->msg_size))
4036   {
4037     GNUNET_break_op (0);
4038     return GNUNET_SYSERR;
4039   }
4040   if (ntohs (fb->frag_off) >= ntohs (fb->msg_size))
4041   {
4042     GNUNET_break_op (0);
4043     return GNUNET_SYSERR;
4044   }
4045   return GNUNET_YES;
4046 }
4047
4048
4049 /**
4050  * Generate a fragment acknowledgement for an @a rc.
4051  *
4052  * @param rc context to generate ACK for, @a rc ACK state is reset
4053  */
4054 static void
4055 send_fragment_ack (struct ReassemblyContext *rc)
4056 {
4057   struct TransportFragmentAckMessage *ack;
4058
4059   ack = GNUNET_new (struct TransportFragmentAckMessage);
4060   ack->header.size = htons (sizeof (struct TransportFragmentAckMessage));
4061   ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK);
4062   ack->frag_uuid = htonl (rc->frag_uuid);
4063   ack->extra_acks = GNUNET_htonll (rc->extra_acks);
4064   ack->msg_uuid = rc->msg_uuid;
4065   ack->avg_ack_delay = GNUNET_TIME_relative_hton (rc->avg_ack_delay);
4066   if (0 == rc->msg_missing)
4067     ack->reassembly_timeout
4068       = GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_FOREVER_REL); /* signal completion */
4069   else
4070     ack->reassembly_timeout
4071       = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout));
4072   route_message (&rc->neighbour->pid,
4073                  &ack->header,
4074                  RMO_DV_ALLOWED);
4075   rc->avg_ack_delay = GNUNET_TIME_UNIT_ZERO;
4076   rc->num_acks = 0;
4077   rc->extra_acks = 0LLU;
4078 }
4079
4080
4081 /**
4082  * Communicator gave us a fragment.  Process the request.
4083  *
4084  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4085  * @param fb the message that was received
4086  */
4087 static void
4088 handle_fragment_box (void *cls,
4089                      const struct TransportFragmentBox *fb)
4090 {
4091   struct CommunicatorMessageContext *cmc = cls;
4092   struct Neighbour *n;
4093   struct ReassemblyContext *rc;
4094   const struct GNUNET_MessageHeader *msg;
4095   uint16_t msize;
4096   uint16_t fsize;
4097   uint16_t frag_off;
4098   uint32_t frag_uuid;
4099   char *target;
4100   struct GNUNET_TIME_Relative cdelay;
4101   int ack_now;
4102
4103   n = GNUNET_CONTAINER_multipeermap_get (neighbours,
4104                                          &cmc->im.sender);
4105   if (NULL == n)
4106   {
4107     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4108
4109     GNUNET_break (0);
4110     finish_cmc_handling (cmc);
4111     GNUNET_SERVICE_client_drop (client);
4112     return;
4113   }
4114   if (NULL == n->reassembly_map)
4115   {
4116     n->reassembly_map = GNUNET_CONTAINER_multishortmap_create (8,
4117                                                                GNUNET_YES);
4118     n->reassembly_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
4119     n->reassembly_timeout_task = GNUNET_SCHEDULER_add_delayed (REASSEMBLY_EXPIRATION,
4120                                                                &reassembly_cleanup_task,
4121                                                                n);
4122   }
4123   msize = ntohs (fb->msg_size);
4124   rc = GNUNET_CONTAINER_multishortmap_get (n->reassembly_map,
4125                                            &fb->msg_uuid);
4126   if (NULL == rc)
4127   {
4128     rc = GNUNET_malloc (sizeof (*rc) +
4129                         msize + /* reassembly payload buffer */
4130                         (msize + 7) / 8 * sizeof (uint8_t) /* bitfield */);
4131     rc->msg_uuid = fb->msg_uuid;
4132     rc->neighbour = n;
4133     rc->msg_size = msize;
4134     rc->reassembly_timeout = GNUNET_TIME_relative_to_absolute (REASSEMBLY_EXPIRATION);
4135     rc->last_frag = GNUNET_TIME_absolute_get ();
4136     rc->hn = GNUNET_CONTAINER_heap_insert (n->reassembly_heap,
4137                                            rc,
4138                                            rc->reassembly_timeout.abs_value_us);
4139     GNUNET_assert (GNUNET_OK ==
4140                    GNUNET_CONTAINER_multishortmap_put (n->reassembly_map,
4141                                                        &rc->msg_uuid,
4142                                                        rc,
4143                                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
4144     target = (char *) &rc[1];
4145     rc->bitfield = (uint8_t *) (target + rc->msg_size);
4146     rc->msg_missing = rc->msg_size;
4147   }
4148   else
4149   {
4150     target = (char *) &rc[1];
4151   }
4152   if (msize != rc->msg_size)
4153   {
4154     GNUNET_break (0);
4155     finish_cmc_handling (cmc);
4156     return;
4157   }
4158
4159   /* reassemble */
4160   fsize = ntohs (fb->header.size) - sizeof (*fb);
4161   frag_off = ntohs (fb->frag_off);
4162   memcpy (&target[frag_off],
4163           &fb[1],
4164           fsize);
4165   /* update bitfield and msg_missing */
4166   for (unsigned int i=frag_off;i<frag_off+fsize;i++)
4167   {
4168     if (0 == (rc->bitfield[i / 8] & (1 << (i % 8))))
4169     {
4170       rc->bitfield[i / 8] |= (1 << (i % 8));
4171       rc->msg_missing--;
4172     }
4173   }
4174
4175   /* Compute cummulative ACK */
4176   frag_uuid = ntohl (fb->frag_uuid);
4177   cdelay = GNUNET_TIME_absolute_get_duration (rc->last_frag);
4178   cdelay = GNUNET_TIME_relative_multiply (cdelay,
4179                                           rc->num_acks);
4180   rc->last_frag = GNUNET_TIME_absolute_get ();
4181   rc->avg_ack_delay = GNUNET_TIME_relative_add (rc->avg_ack_delay,
4182                                                 cdelay);
4183   ack_now = GNUNET_NO;
4184   if (0 == rc->num_acks)
4185   {
4186     /* case one: first ack */
4187     rc->frag_uuid = frag_uuid;
4188     rc->extra_acks = 0LLU;
4189     rc->num_acks = 1;
4190   }
4191   else if ( (frag_uuid >= rc->frag_uuid) &&
4192             (frag_uuid <= rc->frag_uuid + 64) )
4193   {
4194     /* case two: ack fits after existing min UUID */
4195     if ( (frag_uuid == rc->frag_uuid) ||
4196          (0 != (rc->extra_acks & (1LLU << (frag_uuid - rc->frag_uuid - 1)))) )
4197     {
4198       /* duplicate fragment, ack now! */
4199       ack_now = GNUNET_YES;
4200     }
4201     else
4202     {
4203       rc->extra_acks |= (1LLU << (frag_uuid - rc->frag_uuid - 1));
4204       rc->num_acks++;
4205     }
4206   }
4207   else if ( (rc->frag_uuid > frag_uuid) &&
4208             ( ( (rc->frag_uuid == frag_uuid + 64) &&
4209                 (0 == rc->extra_acks) ) ||
4210               ( (rc->frag_uuid < frag_uuid + 64) &&
4211                 (rc->extra_acks == (rc->extra_acks & ~ ((1LLU << (64 - (rc->frag_uuid - frag_uuid))) - 1LLU))) ) ) )
4212   {
4213     /* can fit ack by shifting extra acks and starting at
4214        frag_uid, test above esured that the bits we will
4215        shift 'extra_acks' by are all zero. */
4216     rc->extra_acks <<= (rc->frag_uuid - frag_uuid);
4217     rc->extra_acks |= (1LLU << (rc->frag_uuid - frag_uuid - 1));
4218     rc->frag_uuid = frag_uuid;
4219     rc->num_acks++;
4220   }
4221   if (65 == rc->num_acks) /* FIXME: maybe use smaller threshold? This is very aggressive. */
4222     ack_now = GNUNET_YES; /* maximum acks received */
4223   // FIXME: possibly also ACK based on RTT (but for that we'd need to
4224   // determine the queue used for the ACK first!)
4225
4226   /* is reassembly complete? */
4227   if (0 != rc->msg_missing)
4228   {
4229     if (ack_now)
4230       send_fragment_ack (rc);
4231     finish_cmc_handling (cmc);
4232     return;
4233   }
4234   /* reassembly is complete, verify result */
4235   msg = (const struct GNUNET_MessageHeader *) &rc[1];
4236   if (ntohs (msg->size) != rc->msg_size)
4237   {
4238     GNUNET_break (0);
4239     free_reassembly_context (rc);
4240     finish_cmc_handling (cmc);
4241     return;
4242   }
4243   /* successful reassembly */
4244   send_fragment_ack (rc);
4245   demultiplex_with_cmc (cmc,
4246                         msg);
4247   /* FIXME: really free here? Might be bad if fragments are still
4248      en-route and we forget that we finished this reassembly immediately!
4249      -> keep around until timeout?
4250      -> shorten timeout based on ACK? */
4251   free_reassembly_context (rc);
4252 }
4253
4254
4255 /**
4256  * Check the @a fa against the fragments associated with @a pm.
4257  * If it matches, remove the matching fragments from the transmission
4258  * list.
4259  *
4260  * @param pm pending message to check against the ack
4261  * @param fa the ack that was received
4262  * @return #GNUNET_YES if @a fa matched, #GNUNET_NO if not
4263  */
4264 static int
4265 check_ack_against_pm (struct PendingMessage *pm,
4266                       const struct TransportFragmentAckMessage *fa)
4267 {
4268   int match;
4269   struct PendingMessage *nxt;
4270   uint32_t fs = ntohl (fa->frag_uuid);
4271   uint64_t xtra = GNUNET_ntohll (fa->extra_acks);
4272
4273   match = GNUNET_NO;
4274   for (struct PendingMessage *frag = pm->head_frag;
4275        NULL != frag;
4276        frag = nxt)
4277   {
4278     const struct TransportFragmentBox *tfb
4279       = (const struct TransportFragmentBox *) &pm[1];
4280     uint32_t fu = ntohl (tfb->frag_uuid);
4281
4282     GNUNET_assert (PMT_FRAGMENT_BOX == frag->pmt);
4283     nxt = frag->next_frag;
4284     /* Check for exact match or match in the 'xtra' bitmask */
4285     if ( (fu == fs) ||
4286          ( (fu > fs) &&
4287            (fu <= fs + 64) &&
4288            (0 != (1LLU << (fu - fs - 1) & xtra)) ) )
4289     {
4290       match = GNUNET_YES;
4291       free_fragment_tree (frag);
4292     }
4293   }
4294   return match;
4295 }
4296
4297
4298 /**
4299  * Communicator gave us a fragment acknowledgement.  Process the request.
4300  *
4301  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4302  * @param fa the message that was received
4303  */
4304 static void
4305 handle_fragment_ack (void *cls,
4306                      const struct TransportFragmentAckMessage *fa)
4307 {
4308   struct CommunicatorMessageContext *cmc = cls;
4309   struct Neighbour *n;
4310   int matched;
4311
4312   n = GNUNET_CONTAINER_multipeermap_get (neighbours,
4313                                          &cmc->im.sender);
4314   if (NULL == n)
4315   {
4316     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4317
4318     GNUNET_break (0);
4319     finish_cmc_handling (cmc);
4320     GNUNET_SERVICE_client_drop (client);
4321     return;
4322   }
4323   /* FIXME-OPTIMIZE: maybe use another hash map here? */
4324   matched = GNUNET_NO;
4325   for (struct PendingMessage *pm = n->pending_msg_head;
4326        NULL != pm;
4327        pm = pm->prev_neighbour)
4328   {
4329     if (0 !=
4330         GNUNET_memcmp (&fa->msg_uuid,
4331                        &pm->msg_uuid))
4332       continue;
4333     matched = GNUNET_YES;
4334     if (GNUNET_YES ==
4335         check_ack_against_pm (pm,
4336                               fa))
4337     {
4338       struct GNUNET_TIME_Relative avg_ack_delay
4339         = GNUNET_TIME_relative_ntoh (fa->avg_ack_delay);
4340       // FIXME: update RTT and other reliability data!
4341       // ISSUE: we don't know which of n's queues the message(s)
4342       // took (and in fact the different messages might have gone
4343       // over different queues and possibly over multiple).
4344       // => track queues with PendingMessages, and update RTT only if
4345       //    the queue used is unique?
4346       //    -> how can we get loss rates?
4347       //    -> or, add extra state to Box and ACK to identify queue?
4348       // IDEA: generate MULTIPLE frag-uuids per fragment and track
4349       //    the queue with the fragment! (-> this logic must
4350       //    be moved into check_ack_against_pm!)
4351       (void) avg_ack_delay;
4352     }
4353     else
4354     {
4355       GNUNET_STATISTICS_update (GST_stats,
4356                                 "# FRAGMENT_ACKS dropped, no matching fragment",
4357                                 1,
4358                                 GNUNET_NO);
4359     }
4360     if (NULL == pm->head_frag)
4361     {
4362       // if entire message is ACKed, handle that as well.
4363       // => clean up PM, any post actions?
4364       free_pending_message (pm);
4365     }
4366     else
4367     {
4368       struct GNUNET_TIME_Relative reassembly_timeout
4369         = GNUNET_TIME_relative_ntoh (fa->reassembly_timeout);
4370       // OPTIMIZE-FIXME: adjust retransmission strategy based on reassembly_timeout!
4371       (void) reassembly_timeout;
4372     }
4373     break;
4374   }
4375   if (GNUNET_NO == matched)
4376   {
4377     GNUNET_STATISTICS_update (GST_stats,
4378                               "# FRAGMENT_ACKS dropped, no matching pending message",
4379                               1,
4380                               GNUNET_NO);
4381   }
4382   finish_cmc_handling (cmc);
4383 }
4384
4385
4386 /**
4387  * Communicator gave us a reliability box.  Check the message.
4388  *
4389  * @param cls a `struct CommunicatorMessageContext`
4390  * @param rb the send message that was sent
4391  * @return #GNUNET_YES if message is well-formed
4392  */
4393 static int
4394 check_reliability_box (void *cls,
4395                        const struct TransportReliabilityBox *rb)
4396 {
4397   GNUNET_MQ_check_boxed_message (rb);
4398   return GNUNET_YES;
4399 }
4400
4401
4402 /**
4403  * Communicator gave us a reliability box.  Process the request.
4404  *
4405  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4406  * @param rb the message that was received
4407  */
4408 static void
4409 handle_reliability_box (void *cls,
4410                         const struct TransportReliabilityBox *rb)
4411 {
4412   struct CommunicatorMessageContext *cmc = cls;
4413   const struct GNUNET_MessageHeader *inbox = (const struct GNUNET_MessageHeader *) &rb[1];
4414
4415   if (0 == ntohl (rb->ack_countdown))
4416   {
4417     struct TransportReliabilityAckMessage *ack;
4418
4419     /* FIXME: implement cummulative ACKs and ack_countdown,
4420        then setting the avg_ack_delay field below: */
4421     ack = GNUNET_malloc (sizeof (*ack) +
4422                          sizeof (struct GNUNET_ShortHashCode));
4423     ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK);
4424     ack->header.size = htons (sizeof (*ack) +
4425                               sizeof (struct GNUNET_ShortHashCode));
4426     memcpy (&ack[1],
4427             &rb->msg_uuid,
4428             sizeof (struct GNUNET_ShortHashCode));
4429     route_message (&cmc->im.sender,
4430                    &ack->header,
4431                    RMO_DV_ALLOWED);
4432   }
4433   /* continue with inner message */
4434   demultiplex_with_cmc (cmc,
4435                         inbox);
4436 }
4437
4438
4439 /**
4440  * Communicator gave us a reliability ack.  Process the request.
4441  *
4442  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4443  * @param ra the message that was received
4444  */
4445 static void
4446 handle_reliability_ack (void *cls,
4447                         const struct TransportReliabilityAckMessage *ra)
4448 {
4449   struct CommunicatorMessageContext *cmc = cls;
4450   struct Neighbour *n;
4451   unsigned int n_acks;
4452   const struct GNUNET_ShortHashCode *msg_uuids;
4453   struct PendingMessage *nxt;
4454   int matched;
4455
4456   n = GNUNET_CONTAINER_multipeermap_get (neighbours,
4457                                          &cmc->im.sender);
4458   if (NULL == n)
4459   {
4460     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4461
4462     GNUNET_break (0);
4463     finish_cmc_handling (cmc);
4464     GNUNET_SERVICE_client_drop (client);
4465     return;
4466   }
4467   n_acks = (ntohs (ra->header.size) - sizeof (*ra))
4468     / sizeof (struct GNUNET_ShortHashCode);
4469   msg_uuids = (const struct GNUNET_ShortHashCode *) &ra[1];
4470
4471   /* FIXME-OPTIMIZE: maybe use another hash map here? */
4472   matched = GNUNET_NO;
4473   for (struct PendingMessage *pm = n->pending_msg_head;
4474        NULL != pm;
4475        pm = nxt)
4476   {
4477     int in_list;
4478
4479     nxt = pm->next_neighbour;
4480     in_list = GNUNET_NO;
4481     for (unsigned int i=0;i<n_acks;i++)
4482     {
4483       if (0 !=
4484           GNUNET_memcmp (&msg_uuids[i],
4485                          &pm->msg_uuid))
4486         continue;
4487       in_list = GNUNET_YES;
4488       break;
4489     }
4490     if (GNUNET_NO == in_list)
4491       continue;
4492
4493     /* this pm was acked! */
4494     matched = GNUNET_YES;
4495     free_pending_message (pm);
4496
4497     {
4498       struct GNUNET_TIME_Relative avg_ack_delay
4499         = GNUNET_TIME_relative_ntoh (ra->avg_ack_delay);
4500       // FIXME: update RTT and other reliability data!
4501       // ISSUE: we don't know which of n's queues the message(s)
4502       // took (and in fact the different messages might have gone
4503       // over different queues and possibly over multiple).
4504       // => track queues with PendingMessages, and update RTT only if
4505       //    the queue used is unique?
4506       //    -> how can we get loss rates?
4507       //    -> or, add extra state to MSG and ACKs to identify queue?
4508       //    -> if we do this, might just do the same for the avg_ack_delay!
4509       (void) avg_ack_delay;
4510     }
4511   }
4512   if (GNUNET_NO == matched)
4513   {
4514     GNUNET_STATISTICS_update (GST_stats,
4515                               "# FRAGMENT_ACKS dropped, no matching pending message",
4516                               1,
4517                               GNUNET_NO);
4518   }
4519   finish_cmc_handling (cmc);
4520 }
4521
4522
4523 /**
4524  * Communicator gave us a backchannel encapsulation.  Check the message.
4525  *
4526  * @param cls a `struct CommunicatorMessageContext`
4527  * @param be the send message that was sent
4528  * @return #GNUNET_YES if message is well-formed
4529  */
4530 static int
4531 check_backchannel_encapsulation (void *cls,
4532                                  const struct TransportBackchannelEncapsulationMessage *be)
4533 {
4534   uint16_t size = ntohs (be->header.size);
4535
4536   (void) cls;
4537   if (size - sizeof (*be) <
4538       sizeof (struct TransportBackchannelRequestPayload) +
4539       sizeof (struct GNUNET_MessageHeader) )
4540   {
4541     GNUNET_break_op (0);
4542     return GNUNET_SYSERR;
4543   }
4544   return GNUNET_YES;
4545 }
4546
4547
4548 /**
4549  * Communicator gave us a backchannel encapsulation.  Process the request.
4550  * (We are not the origin of the backchannel here, the communicator simply
4551  * received a backchannel message and we are expected to forward it.)
4552  *
4553  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4554  * @param be the message that was received
4555  */
4556 static void
4557 handle_backchannel_encapsulation (void *cls,
4558                                   const struct TransportBackchannelEncapsulationMessage *be)
4559 {
4560   struct CommunicatorMessageContext *cmc = cls;
4561   struct BackchannelKeyState key;
4562   struct GNUNET_HashCode hmac;
4563   const char *hdr;
4564   size_t hdr_len;
4565
4566   if (0 != GNUNET_memcmp (&be->target,
4567                           &GST_my_identity))
4568   {
4569     /* not for me, try to route to target */
4570     /* FIXME: someone needs to update be->distance! */
4571     /* FIXME: BE routing can be special, should we put all of this
4572        on 'route_message'? Maybe at least pass some more arguments? */
4573     route_message (&be->target,
4574                    GNUNET_copy_message (&be->header),
4575                    RMO_DV_ALLOWED);
4576     finish_cmc_handling (cmc);
4577     return;
4578   }
4579   dh_key_derive_eph_pub (&be->ephemeral_key,
4580                          &be->iv,
4581                          &key);
4582   hdr = (const char *) &be[1];
4583   hdr_len = ntohs (be->header.size) - sizeof (*be);
4584   bc_hmac (&key,
4585            &hmac,
4586            hdr,
4587            hdr_len);
4588   if (0 !=
4589       GNUNET_memcmp (&hmac,
4590                      &be->hmac))
4591   {
4592     /* HMAC missmatch, disard! */
4593     GNUNET_break_op (0);
4594     finish_cmc_handling (cmc);
4595     return;
4596   }
4597   /* begin actual decryption */
4598   {
4599     struct TransportBackchannelRequestPayload ppay;
4600     char body[hdr_len - sizeof (ppay)];
4601
4602     GNUNET_assert (hdr_len >= sizeof (ppay) + sizeof (struct GNUNET_MessageHeader));
4603     bc_decrypt (&key,
4604                 &ppay,
4605                 hdr,
4606                 sizeof (ppay));
4607     bc_decrypt (&key,
4608                 &body,
4609                 &hdr[sizeof (ppay)],
4610                 hdr_len - sizeof (ppay));
4611     bc_key_clean (&key);
4612     // FIXME: verify signatures in ppay!
4613     // => check if ephemeral key is known & valid, if not
4614     // => verify sig, cache ephemeral key
4615     // => update monotonic_time of sender for replay detection
4616
4617     // FIXME: forward to specified communicator!
4618     // (using GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL_INCOMING)
4619   }
4620   finish_cmc_handling (cmc);
4621 }
4622
4623
4624 /**
4625  * Task called when we should check if any of the DV paths
4626  * we have learned to a target are due for garbage collection.
4627  *
4628  * Collects stale paths, and possibly frees the entire DV
4629  * entry if no paths are left. Otherwise re-schedules itself.
4630  *
4631  * @param cls a `struct DistanceVector`
4632  */
4633 static void
4634 path_cleanup_cb (void *cls)
4635 {
4636   struct DistanceVector *dv = cls;
4637   struct DistanceVectorHop *pos;
4638
4639   dv->timeout_task = NULL;
4640   while (NULL != (pos = dv->dv_head))
4641   {
4642     GNUNET_assert (dv == pos->dv);
4643     if (GNUNET_TIME_absolute_get_remaining (pos->timeout).rel_value_us > 0)
4644       break;
4645     free_distance_vector_hop (pos);
4646   }
4647   if (NULL == pos)
4648   {
4649     free_dv_route (dv);
4650     return;
4651   }
4652   dv->timeout_task = GNUNET_SCHEDULER_add_at (pos->timeout,
4653                                               &path_cleanup_cb,
4654                                               dv);
4655 }
4656
4657
4658 /**
4659  * We have learned a @a path through the network to some other peer, add it to
4660  * our DV data structure (returning #GNUNET_YES on success).
4661  *
4662  * We do not add paths if we have a sufficient number of shorter
4663  * paths to this target already (returning #GNUNET_NO).
4664  *
4665  * We also do not add problematic paths, like those where we lack the first
4666  * hop in our neighbour list (i.e. due to a topology change) or where some
4667  * non-first hop is in our neighbour list (returning #GNUNET_SYSERR).
4668  *
4669  * @param path the path we learned, path[0] should be us,
4670  *             and then path contains a valid path from us to `path[path_len-1]`
4671  *             path[1] should be a direct neighbour (we should check!)
4672  * @param path_len number of entries on the @a path, at least three!
4673  * @param network_latency how long does the message take from us to `path[path_len-1]`?
4674  *          set to "forever" if unknown
4675  * @return #GNUNET_YES on success,
4676  *         #GNUNET_NO if we have better path(s) to the target
4677  *         #GNUNET_SYSERR if the path is useless and/or invalid
4678  *                         (i.e. path[1] not a direct neighbour
4679  *                        or path[i+1] is a direct neighbour for i>0)
4680  */
4681 static int
4682 learn_dv_path (const struct GNUNET_PeerIdentity *path,
4683                unsigned int path_len,
4684                struct GNUNET_TIME_Relative network_latency)
4685 {
4686   struct DistanceVectorHop *hop;
4687   struct DistanceVector *dv;
4688   struct Neighbour *next_hop;
4689   unsigned int shorter_distance;
4690
4691   if (path_len < 3)
4692   {
4693     /* what a boring path! not allowed! */
4694     GNUNET_break (0);
4695     return GNUNET_SYSERR;
4696   }
4697   GNUNET_assert (0 ==
4698                  GNUNET_memcmp (&GST_my_identity,
4699                                 &path[0]));
4700   next_hop = GNUNET_CONTAINER_multipeermap_get (neighbours,
4701                                                 &path[1]);
4702   if (NULL == next_hop)
4703   {
4704     /* next hop must be a neighbour, otherwise this whole thing is useless! */
4705     GNUNET_break (0);
4706     return GNUNET_SYSERR;
4707   }
4708   for (unsigned int i=2;i<path_len;i++)
4709     if (NULL !=
4710         GNUNET_CONTAINER_multipeermap_get (neighbours,
4711                                            &path[i]))
4712     {
4713       /* Useless path, we have a direct connection to some hop
4714          in the middle of the path, so this one doesn't even
4715          seem terribly useful for redundancy */
4716       return GNUNET_SYSERR;
4717     }
4718   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes,
4719                                           &path[path_len - 1]);
4720   if (NULL == dv)
4721   {
4722     dv = GNUNET_new (struct DistanceVector);
4723     dv->target = path[path_len - 1];
4724     dv->timeout_task = GNUNET_SCHEDULER_add_delayed (DV_PATH_VALIDITY_TIMEOUT,
4725                                                      &path_cleanup_cb,
4726                                                      dv);
4727     GNUNET_assert (GNUNET_OK ==
4728                    GNUNET_CONTAINER_multipeermap_put (dv_routes,
4729                                                       &dv->target,
4730                                                       dv,
4731                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
4732   }
4733   /* Check if we have this path already! */
4734   shorter_distance = 0;
4735   for (struct DistanceVectorHop *pos = dv->dv_head;
4736        NULL != pos;
4737        pos = pos->next_dv)
4738   {
4739     if (pos->distance < path_len - 2)
4740       shorter_distance++;
4741     /* Note that the distances in 'pos' excludes us (path[0]) and
4742        the next_hop (path[1]), so we need to subtract two
4743        and check next_hop explicitly */
4744     if ( (pos->distance == path_len - 2) &&
4745          (pos->next_hop == next_hop) )
4746     {
4747       int match = GNUNET_YES;
4748
4749       for (unsigned int i=0;i<pos->distance;i++)
4750       {
4751         if (0 !=
4752             GNUNET_memcmp (&pos->path[i],
4753                            &path[i+2]))
4754         {
4755           match = GNUNET_NO;
4756           break;
4757         }
4758       }
4759       if (GNUNET_YES == match)
4760       {
4761         struct GNUNET_TIME_Relative last_timeout;
4762
4763         /* Re-discovered known path, update timeout */
4764         GNUNET_STATISTICS_update (GST_stats,
4765                                   "# Known DV path refreshed",
4766                                   1,
4767                                   GNUNET_NO);
4768         last_timeout = GNUNET_TIME_absolute_get_remaining (pos->timeout);
4769         pos->timeout
4770           = GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
4771         GNUNET_CONTAINER_MDLL_remove (dv,
4772                                       dv->dv_head,
4773                                       dv->dv_tail,
4774                                       pos);
4775         GNUNET_CONTAINER_MDLL_insert (dv,
4776                                       dv->dv_head,
4777                                       dv->dv_tail,
4778                                       pos);
4779         if (last_timeout.rel_value_us <
4780             GNUNET_TIME_relative_subtract (DV_PATH_VALIDITY_TIMEOUT,
4781                                            DV_PATH_DISCOVERY_FREQUENCY).rel_value_us)
4782         {
4783           /* Some peer send DV learn messages too often, we are learning
4784              the same path faster than it would be useful; do not forward! */
4785           return GNUNET_NO;
4786         }
4787         return GNUNET_YES;
4788       }
4789     }
4790   }
4791   /* Count how many shorter paths we have (incl. direct
4792      neighbours) before simply giving up on this one! */
4793   if (shorter_distance >= MAX_DV_PATHS_TO_TARGET)
4794   {
4795     /* We have a shorter path already! */
4796     return GNUNET_NO;
4797   }
4798   /* create new DV path entry */
4799   hop = GNUNET_malloc (sizeof (struct DistanceVectorHop) +
4800                        sizeof (struct GNUNET_PeerIdentity) * (path_len - 2));
4801   hop->next_hop = next_hop;
4802   hop->dv = dv;
4803   hop->path = (const struct GNUNET_PeerIdentity *) &hop[1];
4804   memcpy (&hop[1],
4805           &path[2],
4806           sizeof (struct GNUNET_PeerIdentity) * (path_len - 2));
4807   hop->timeout = GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
4808   hop->distance = path_len - 2;
4809   GNUNET_CONTAINER_MDLL_insert (dv,
4810                                 dv->dv_head,
4811                                 dv->dv_tail,
4812                                 hop);
4813   GNUNET_CONTAINER_MDLL_insert (neighbour,
4814                                 next_hop->dv_head,
4815                                 next_hop->dv_tail,
4816                                 hop);
4817   return GNUNET_YES;
4818 }
4819
4820
4821 /**
4822  * Communicator gave us a DV learn message.  Check the message.
4823  *
4824  * @param cls a `struct CommunicatorMessageContext`
4825  * @param dvl the send message that was sent
4826  * @return #GNUNET_YES if message is well-formed
4827  */
4828 static int
4829 check_dv_learn (void *cls,
4830                 const struct TransportDVLearn *dvl)
4831 {
4832   uint16_t size = ntohs (dvl->header.size);
4833   uint16_t num_hops = ntohs (dvl->num_hops);
4834   const struct DVPathEntryP *hops = (const struct DVPathEntryP *) &dvl[1];
4835
4836   (void) cls;
4837   if (size != sizeof (*dvl) + num_hops * sizeof (struct DVPathEntryP))
4838   {
4839     GNUNET_break_op (0);
4840     return GNUNET_SYSERR;
4841   }
4842   if (num_hops > MAX_DV_HOPS_ALLOWED)
4843   {
4844     GNUNET_break_op (0);
4845     return GNUNET_SYSERR;
4846   }
4847   for (unsigned int i=0;i<num_hops;i++)
4848   {
4849     if (0 == GNUNET_memcmp (&dvl->initiator,
4850                             &hops[i].hop))
4851     {
4852       GNUNET_break_op (0);
4853       return GNUNET_SYSERR;
4854     }
4855     if (0 == GNUNET_memcmp (&GST_my_identity,
4856                             &hops[i].hop))
4857     {
4858       GNUNET_break_op (0);
4859       return GNUNET_SYSERR;
4860     }
4861   }
4862   return GNUNET_YES;
4863 }
4864
4865
4866 /**
4867  * Build and forward a DV learn message to @a next_hop.
4868  *
4869  * @param next_hop peer to send the message to
4870  * @param msg message received
4871  * @param bi_history bitmask specifying hops on path that were bidirectional
4872  * @param nhops length of the @a hops array
4873  * @param hops path the message traversed so far
4874  * @param in_time when did we receive the message, used to calculate network delay
4875  */
4876 static void
4877 forward_dv_learn (const struct GNUNET_PeerIdentity *next_hop,
4878                   const struct TransportDVLearn *msg,
4879                   uint16_t bi_history,
4880                   uint16_t nhops,
4881                   const struct DVPathEntryP *hops,
4882                   struct GNUNET_TIME_Absolute in_time)
4883 {
4884   struct DVPathEntryP *dhops;
4885   struct TransportDVLearn *fwd;
4886   struct GNUNET_TIME_Relative nnd;
4887
4888   /* compute message for forwarding */
4889   GNUNET_assert (nhops < MAX_DV_HOPS_ALLOWED);
4890   fwd = GNUNET_malloc (sizeof (struct TransportDVLearn) +
4891                        (nhops + 1) * sizeof (struct DVPathEntryP));
4892   fwd->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN);
4893   fwd->header.size = htons (sizeof (struct TransportDVLearn) +
4894                             (nhops + 1) * sizeof (struct DVPathEntryP));
4895   fwd->num_hops = htons (nhops + 1);
4896   fwd->bidirectional = htons (bi_history);
4897   nnd = GNUNET_TIME_relative_add (GNUNET_TIME_absolute_get_duration (in_time),
4898                                   GNUNET_TIME_relative_ntoh (msg->non_network_delay));
4899   fwd->non_network_delay = GNUNET_TIME_relative_hton (nnd);
4900   fwd->init_sig = msg->init_sig;
4901   fwd->initiator = msg->initiator;
4902   fwd->challenge = msg->challenge;
4903   dhops = (struct DVPathEntryP *) &fwd[1];
4904   GNUNET_memcpy (dhops,
4905                  hops,
4906                  sizeof (struct DVPathEntryP) * nhops);
4907   dhops[nhops].hop = GST_my_identity;
4908   {
4909     struct DvHopPS dhp = {
4910       .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP),
4911       .purpose.size = htonl (sizeof (dhp)),
4912       .pred = dhops[nhops-1].hop,
4913       .succ = *next_hop,
4914       .challenge = msg->challenge
4915     };
4916
4917     GNUNET_assert (GNUNET_OK ==
4918                    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
4919                                              &dhp.purpose,
4920                                              &dhops[nhops].hop_sig));
4921   }
4922   route_message (next_hop,
4923                  &fwd->header,
4924                  RMO_UNCONFIRMED_ALLOWED);
4925 }
4926
4927
4928 /**
4929  * Check signature of type #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
4930  *
4931  * @param init the signer
4932  * @param challenge the challenge that was signed
4933  * @param init_sig signature presumably by @a init
4934  * @return #GNUNET_OK if the signature is valid
4935  */
4936 static int
4937 validate_dv_initiator_signature (const struct GNUNET_PeerIdentity *init,
4938                                  const struct GNUNET_ShortHashCode *challenge,
4939                                  const struct GNUNET_CRYPTO_EddsaSignature *init_sig)
4940 {
4941   struct DvInitPS ip = {
4942     .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR),
4943     .purpose.size = htonl (sizeof (ip)),
4944     .challenge = *challenge
4945   };
4946
4947   if (GNUNET_OK !=
4948       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR,
4949                                   &ip.purpose,
4950                                   init_sig,
4951                                   &init->public_key))
4952   {
4953     GNUNET_break_op (0);
4954     return GNUNET_SYSERR;
4955   }
4956   return GNUNET_OK;
4957 }
4958
4959
4960 /**
4961  * Communicator gave us a DV learn message.  Process the request.
4962  *
4963  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
4964  * @param dvl the message that was received
4965  */
4966 static void
4967 handle_dv_learn (void *cls,
4968                  const struct TransportDVLearn *dvl)
4969 {
4970   struct CommunicatorMessageContext *cmc = cls;
4971   enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
4972   int bi_hop;
4973   uint16_t nhops;
4974   uint16_t bi_history;
4975   const struct DVPathEntryP *hops;
4976   int do_fwd;
4977   int did_initiator;
4978   struct GNUNET_TIME_Absolute in_time;
4979
4980   nhops = ntohs (dvl->bidirectional); /* 0 = sender is initiator */
4981   bi_history = ntohs (dvl->bidirectional);
4982   hops = (const struct DVPathEntryP *) &dvl[1];
4983   if (0 == nhops)
4984   {
4985     /* sanity check */
4986     if (0 != GNUNET_memcmp (&dvl->initiator,
4987                             &cmc->im.sender))
4988     {
4989       GNUNET_break (0);
4990       finish_cmc_handling (cmc);
4991       return;
4992     }
4993   }
4994   else
4995   {
4996     /* sanity check */
4997     if (0 != GNUNET_memcmp (&hops[nhops - 1].hop,
4998                             &cmc->im.sender))
4999     {
5000       GNUNET_break (0);
5001       finish_cmc_handling (cmc);
5002       return;
5003     }
5004   }
5005
5006   GNUNET_assert (CT_COMMUNICATOR == cmc->tc->type);
5007   cc = cmc->tc->details.communicator.cc;
5008   bi_hop = (GNUNET_TRANSPORT_CC_RELIABLE == cc); // FIXME: add bi-directional flag to cc?
5009   in_time = GNUNET_TIME_absolute_get ();
5010
5011   /* continue communicator here, everything else can happen asynchronous! */
5012   finish_cmc_handling (cmc);
5013
5014   /* OPTIMIZE-FIXME: Technically, we only need to bother checking
5015      the initiator signature if we send the message back to the initiator... */
5016   if (GNUNET_OK !=
5017       validate_dv_initiator_signature (&dvl->initiator,
5018                                        &dvl->challenge,
5019                                        &dvl->init_sig))
5020   {
5021     GNUNET_break_op (0);
5022     return;
5023   }
5024   // FIXME: asynchronously (!) verify hop-by-hop signatures!
5025   // => if signature verification load too high, implement random drop strategy!?
5026
5027   do_fwd = GNUNET_YES;
5028   if (0 == GNUNET_memcmp (&GST_my_identity,
5029                           &dvl->initiator))
5030   {
5031     struct GNUNET_PeerIdentity path[nhops + 1];
5032     struct GNUNET_TIME_Relative host_latency_sum;
5033     struct GNUNET_TIME_Relative latency;
5034     struct GNUNET_TIME_Relative network_latency;
5035
5036     /* We initiated this, learn the forward path! */
5037     path[0] = GST_my_identity;
5038     path[1] = hops[0].hop;
5039     host_latency_sum = GNUNET_TIME_relative_ntoh (dvl->non_network_delay);
5040
5041     // Need also something to lookup initiation time
5042     // to compute RTT! -> add RTT argument here?
5043     latency = GNUNET_TIME_UNIT_FOREVER_REL; // FIXME: initialize properly
5044     // (based on dvl->challenge, we can identify time of origin!)
5045
5046     network_latency = GNUNET_TIME_relative_subtract (latency,
5047                                                      host_latency_sum);
5048     /* assumption: latency on all links is the same */
5049     network_latency = GNUNET_TIME_relative_divide (network_latency,
5050                                                    nhops);
5051
5052     for (unsigned int i=2;i<=nhops;i++)
5053     {
5054       struct GNUNET_TIME_Relative ilat;
5055
5056       /* assumption: linear latency increase per hop */
5057       ilat = GNUNET_TIME_relative_multiply (network_latency,
5058                                             i);
5059       path[i] = hops[i-1].hop;
5060       // FIXME: mark ALL of these as *confirmed* (with what timeout?)
5061       // -- and schedule a job for the confirmation to time out! --
5062       // and possibly do #cores_send_connect_info() if
5063       // the respective neighbour is NOT confirmed yet!
5064       learn_dv_path (path,
5065                      i,
5066                      ilat);
5067     }
5068     /* as we initiated, do not forward again (would be circular!) */
5069     do_fwd = GNUNET_NO;
5070     return;
5071   }
5072   else if (bi_hop)
5073   {
5074     /* last hop was bi-directional, we could learn something here! */
5075     struct GNUNET_PeerIdentity path[nhops + 2];
5076
5077     path[0] = GST_my_identity;
5078     path[1] = hops[nhops - 1].hop; /* direct neighbour == predecessor! */
5079     for (unsigned int i=0;i<nhops;i++)
5080     {
5081       int iret;
5082
5083       if (0 == (bi_history & (1 << i)))
5084         break; /* i-th hop not bi-directional, stop learning! */
5085       if (i == nhops)
5086       {
5087         path[i + 2] = dvl->initiator;
5088       }
5089       else
5090       {
5091         path[i + 2] = hops[nhops - i - 2].hop;
5092       }
5093
5094       iret = learn_dv_path (path,
5095                             i + 2,
5096                             GNUNET_TIME_UNIT_FOREVER_REL);
5097       if (GNUNET_SYSERR == iret)
5098       {
5099         /* path invalid or too long to be interesting for US, thus should also
5100            not be interesting to our neighbours, cut path when forwarding to
5101            'i' hops, except of course for the one that goes back to the
5102            initiator */
5103         GNUNET_STATISTICS_update (GST_stats,
5104                                   "# DV learn not forwarded due invalidity of path",
5105                                   1,
5106                                   GNUNET_NO);
5107         do_fwd = GNUNET_NO;
5108         break;
5109       }
5110       if ( (GNUNET_NO == iret) &&
5111            (nhops == i + 1) )
5112       {
5113         /* we have better paths, and this is the longest target,
5114            so there cannot be anything interesting later */
5115         GNUNET_STATISTICS_update (GST_stats,
5116                                   "# DV learn not forwarded, got better paths",
5117                                   1,
5118                                   GNUNET_NO);
5119         do_fwd = GNUNET_NO;
5120         break;
5121       }
5122     }
5123   }
5124
5125   if (MAX_DV_HOPS_ALLOWED == nhops)
5126   {
5127     /* At limit, we're out of here! */
5128     finish_cmc_handling (cmc);
5129     return;
5130   }
5131
5132   /* Forward to initiator, if path non-trivial and possible */
5133   bi_history = (bi_history << 1) | (bi_hop ? 1 : 0);
5134   did_initiator = GNUNET_NO;
5135   if ( (1 < nhops) &&
5136        (GNUNET_YES ==
5137         GNUNET_CONTAINER_multipeermap_contains (neighbours,
5138                                                 &dvl->initiator)) )
5139   {
5140     /* send back to origin! */
5141     forward_dv_learn (&dvl->initiator,
5142                       dvl,
5143                       bi_history,
5144                       nhops,
5145                       hops,
5146                       in_time);
5147     did_initiator = GNUNET_YES;
5148   }
5149   /* We forward under two conditions: either we still learned something
5150      ourselves (do_fwd), or the path was darn short and thus the initiator is
5151      likely to still be very interested in this (and we did NOT already
5152      send it back to the initiator) */
5153   if ( (do_fwd) ||
5154        ( (nhops < MIN_DV_PATH_LENGTH_FOR_INITIATOR) &&
5155          (GNUNET_NO == did_initiator) ) )
5156   {
5157     /* FIXME: loop over all neighbours, pick those with low
5158        queues AND that are not yet on the path; possibly
5159        adapt threshold to nhops! */
5160 #if FIXME
5161     forward_dv_learn (NULL, // fill in peer from iterator here!
5162                       dvl,
5163                       bi_history,
5164                       nhops,
5165                       hops,
5166                       in_time);
5167 #endif
5168   }
5169 }
5170
5171
5172 /**
5173  * Communicator gave us a DV box.  Check the message.
5174  *
5175  * @param cls a `struct CommunicatorMessageContext`
5176  * @param dvb the send message that was sent
5177  * @return #GNUNET_YES if message is well-formed
5178  */
5179 static int
5180 check_dv_box (void *cls,
5181               const struct TransportDVBox *dvb)
5182 {
5183   uint16_t size = ntohs (dvb->header.size);
5184   uint16_t num_hops = ntohs (dvb->num_hops);
5185   const struct GNUNET_PeerIdentity *hops = (const struct GNUNET_PeerIdentity *) &dvb[1];
5186   const struct GNUNET_MessageHeader *inbox = (const struct GNUNET_MessageHeader *) &hops[num_hops];
5187   uint16_t isize;
5188   uint16_t itype;
5189
5190   (void) cls;
5191   if (size < sizeof (*dvb) + num_hops * sizeof (struct GNUNET_PeerIdentity) + sizeof (struct GNUNET_MessageHeader))
5192   {
5193     GNUNET_break_op (0);
5194     return GNUNET_SYSERR;
5195   }
5196   isize = ntohs (inbox->size);
5197   if (size != sizeof (*dvb) + num_hops * sizeof (struct GNUNET_PeerIdentity) + isize)
5198   {
5199     GNUNET_break_op (0);
5200     return GNUNET_SYSERR;
5201   }
5202   itype = ntohs (inbox->type);
5203   if ( (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX == itype) ||
5204        (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN == itype) )
5205   {
5206     GNUNET_break_op (0);
5207     return GNUNET_SYSERR;
5208   }
5209   if (0 ==
5210       GNUNET_memcmp (&dvb->origin,
5211                      &GST_my_identity))
5212   {
5213     GNUNET_break_op (0);
5214     return GNUNET_SYSERR;
5215   }
5216   return GNUNET_YES;
5217 }
5218
5219
5220 /**
5221  * Create a DV Box message and queue it for transmission to
5222  * @ea next_hop.
5223  *
5224  * @param next_hop peer to receive the message next
5225  * @param total_hops how many hops did the message take so far
5226  * @param num_hops length of the @a hops array
5227  * @param origin origin of the message
5228  * @param hops next peer(s) to the destination, including destination
5229  * @param payload payload of the box
5230  * @param payload_size number of bytes in @a payload
5231  */
5232 static void
5233 forward_dv_box (struct Neighbour *next_hop,
5234                 uint16_t total_hops,
5235                 uint16_t num_hops,
5236                 const struct GNUNET_PeerIdentity *origin,
5237                 const struct GNUNET_PeerIdentity *hops,
5238                 const void *payload,
5239                 uint16_t payload_size)
5240 {
5241   struct TransportDVBox *dvb;
5242   struct GNUNET_PeerIdentity *dhops;
5243
5244   GNUNET_assert (UINT16_MAX <
5245                  sizeof (struct TransportDVBox) +
5246                  sizeof (struct GNUNET_PeerIdentity) * num_hops +
5247                  payload_size);
5248   dvb = GNUNET_malloc (sizeof (struct TransportDVBox) +
5249                        sizeof (struct GNUNET_PeerIdentity) * num_hops +
5250                        payload_size);
5251   dvb->header.size = htons (sizeof (struct TransportDVBox) +
5252                             sizeof (struct GNUNET_PeerIdentity) * num_hops +
5253                             payload_size);
5254   dvb->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX);
5255   dvb->total_hops = htons (total_hops);
5256   dvb->num_hops = htons (num_hops);
5257   dvb->origin = *origin;
5258   dhops = (struct GNUNET_PeerIdentity *) &dvb[1];
5259   memcpy (dhops,
5260           hops,
5261           num_hops * sizeof (struct GNUNET_PeerIdentity));
5262   memcpy (&dhops[num_hops],
5263           payload,
5264           payload_size);
5265   route_message (&next_hop->pid,
5266                  &dvb->header,
5267                  RMO_NONE);
5268 }
5269
5270
5271 /**
5272  * Communicator gave us a DV box.  Process the request.
5273  *
5274  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
5275  * @param dvb the message that was received
5276  */
5277 static void
5278 handle_dv_box (void *cls,
5279                const struct TransportDVBox *dvb)
5280 {
5281   struct CommunicatorMessageContext *cmc = cls;
5282   uint16_t size = ntohs (dvb->header.size) - sizeof (*dvb);
5283   uint16_t num_hops = ntohs (dvb->num_hops);
5284   const struct GNUNET_PeerIdentity *hops = (const struct GNUNET_PeerIdentity *) &dvb[1];
5285   const struct GNUNET_MessageHeader *inbox = (const struct GNUNET_MessageHeader *) &hops[num_hops];
5286
5287   if (num_hops > 0)
5288   {
5289     /* We're trying from the end of the hops array, as we may be
5290        able to find a shortcut unknown to the origin that way */
5291     for (int i=num_hops-1;i>=0;i--)
5292     {
5293       struct Neighbour *n;
5294
5295       if (0 ==
5296           GNUNET_memcmp (&hops[i],
5297                          &GST_my_identity))
5298       {
5299         GNUNET_break_op (0);
5300         finish_cmc_handling (cmc);
5301         return;
5302       }
5303       n = GNUNET_CONTAINER_multipeermap_get (neighbours,
5304                                              &hops[i]);
5305       if (NULL == n)
5306         continue;
5307       forward_dv_box (n,
5308                       ntohs (dvb->total_hops) + 1,
5309                       num_hops - i - 1, /* number of hops left */
5310                       &dvb->origin,
5311                       &hops[i+1], /* remaining hops */
5312                       (const void *) &dvb[1],
5313                       size);
5314       finish_cmc_handling (cmc);
5315       return;
5316     }
5317     /* Woopsie, next hop not in neighbours, drop! */
5318     GNUNET_STATISTICS_update (GST_stats,
5319                               "# DV Boxes dropped: next hop unknown",
5320                               1,
5321                               GNUNET_NO);
5322     finish_cmc_handling (cmc);
5323     return;
5324   }
5325   /* We are the target. Unbox and handle message. */
5326   cmc->im.sender = dvb->origin;
5327   cmc->total_hops = ntohs (dvb->total_hops);
5328   demultiplex_with_cmc (cmc,
5329                         inbox);
5330 }
5331
5332
5333 /**
5334  * Client notified us about transmission from a peer.  Process the request.
5335  *
5336  * @param cls a `struct TransportClient` which sent us the message
5337  * @param obm the send message that was sent
5338  * @return #GNUNET_YES if message is well-formed
5339  */
5340 static int
5341 check_incoming_msg (void *cls,
5342                     const struct GNUNET_TRANSPORT_IncomingMessage *im)
5343 {
5344   struct TransportClient *tc = cls;
5345
5346   if (CT_COMMUNICATOR != tc->type)
5347   {
5348     GNUNET_break (0);
5349     return GNUNET_SYSERR;
5350   }
5351   GNUNET_MQ_check_boxed_message (im);
5352   return GNUNET_OK;
5353 }
5354
5355
5356 /**
5357  * Communicator gave us a transport address validation challenge.  Process the request.
5358  *
5359  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
5360  * @param tvc the message that was received
5361  */
5362 static void
5363 handle_validation_challenge (void *cls,
5364                              const struct TransportValidationChallenge *tvc)
5365 {
5366   struct CommunicatorMessageContext *cmc = cls;
5367   struct TransportValidationResponse *tvr;
5368
5369   if (cmc->total_hops > 0)
5370   {
5371     /* DV routing is not allowed for validation challenges! */
5372     GNUNET_break_op (0);
5373     finish_cmc_handling (cmc);
5374     return;
5375   }
5376   tvr = GNUNET_new (struct TransportValidationResponse);
5377   tvr->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE);
5378   tvr->header.size = htons (sizeof (*tvr));
5379   tvr->challenge = tvc->challenge;
5380   tvr->origin_time = tvc->sender_time;
5381   tvr->validity_duration = cmc->im.expected_address_validity;
5382   {
5383     /* create signature */
5384     struct TransportValidationPS tvp = {
5385       .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE),
5386       .purpose.size = htonl (sizeof (tvp)),
5387       .validity_duration = tvr->validity_duration,
5388       .challenge = tvc->challenge
5389     };
5390
5391     GNUNET_assert (GNUNET_OK ==
5392                    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
5393                                              &tvp.purpose,
5394                                              &tvr->signature));
5395   }
5396   route_message (&cmc->im.sender,
5397                  &tvr->header,
5398                  RMO_ANYTHING_GOES | RMO_REDUNDANT);
5399   finish_cmc_handling (cmc);
5400 }
5401
5402
5403 /**
5404  * Closure for #check_known_challenge.
5405  */
5406 struct CheckKnownChallengeContext
5407 {
5408   /**
5409    * Set to the challenge we are looking for.
5410    */
5411   const struct GNUNET_ShortHashCode *challenge;
5412
5413   /**
5414    * Set to a matching validation state, if one was found.
5415    */
5416   struct ValidationState *vs;
5417 };
5418
5419
5420 /**
5421  * Test if the validation state in @a value matches the
5422  * challenge from @a cls.
5423  *
5424  * @param cls a `struct CheckKnownChallengeContext`
5425  * @param pid unused (must match though)
5426  * @param value a `struct ValidationState`
5427  * @return #GNUNET_OK if not matching, #GNUNET_NO if match found
5428  */
5429 static int
5430 check_known_challenge (void *cls,
5431                        const struct GNUNET_PeerIdentity *pid,
5432                        void *value)
5433 {
5434   struct CheckKnownChallengeContext *ckac = cls;
5435   struct ValidationState *vs = value;
5436
5437   (void) pid;
5438   if (0 != GNUNET_memcmp (&vs->challenge,
5439                           ckac->challenge))
5440     return GNUNET_OK;
5441   ckac->vs = vs;
5442   return GNUNET_NO;
5443 }
5444
5445
5446 /**
5447  * Function called when peerstore is done storing a
5448  * validated address.
5449  *
5450  * @param cls a `struct ValidationState`
5451  * @param success #GNUNET_YES on success
5452  */
5453 static void
5454 peerstore_store_validation_cb (void *cls,
5455                                int success)
5456 {
5457   struct ValidationState *vs = cls;
5458
5459   vs->sc = NULL;
5460   if (GNUNET_YES == success)
5461     return;
5462   GNUNET_STATISTICS_update (GST_stats,
5463                             "# Peerstore failed to store foreign address",
5464                             1,
5465                             GNUNET_NO);
5466 }
5467
5468
5469 /**
5470  * Task run periodically to validate some address based on #validation_heap.
5471  *
5472  * @param cls NULL
5473  */
5474 static void
5475 validation_start_cb (void *cls);
5476
5477
5478 /**
5479  * Set the time for next_challenge of @a vs to @a new_time.
5480  * Updates the heap and if necessary reschedules the job.
5481  *
5482  * @param vs validation state to update
5483  * @param new_time new time for revalidation
5484  */
5485 static void
5486 update_next_challenge_time (struct ValidationState *vs,
5487                             struct GNUNET_TIME_Absolute new_time)
5488 {
5489   struct GNUNET_TIME_Relative delta;
5490
5491   if (new_time.abs_value_us == vs->next_challenge.abs_value_us)
5492     return; /* be lazy */
5493   vs->next_challenge = new_time;
5494   if (NULL == vs->hn)
5495     vs->hn = GNUNET_CONTAINER_heap_insert (validation_heap,
5496                                            vs,
5497                                            new_time.abs_value_us);
5498   else
5499     GNUNET_CONTAINER_heap_update_cost (vs->hn,
5500                                        new_time.abs_value_us);
5501   if ( (vs != GNUNET_CONTAINER_heap_peek (validation_heap)) &&
5502        (NULL != validation_task) )
5503     return;
5504   if (NULL != validation_task)
5505     GNUNET_SCHEDULER_cancel (validation_task);
5506   /* randomize a bit */
5507   delta.rel_value_us = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
5508                                                  MIN_DELAY_ADDRESS_VALIDATION.rel_value_us);
5509   new_time = GNUNET_TIME_absolute_add (new_time,
5510                                        delta);
5511   validation_task = GNUNET_SCHEDULER_add_at (new_time,
5512                                              &validation_start_cb,
5513                                              NULL);
5514 }
5515
5516
5517 /**
5518  * Find the queue matching @a pid and @a address.
5519  *
5520  * @param pid peer the queue must go to
5521  * @param address address the queue must use
5522  * @return NULL if no such queue exists
5523  */
5524 static struct Queue *
5525 find_queue (const struct GNUNET_PeerIdentity *pid,
5526             const char *address)
5527 {
5528   struct Neighbour *n;
5529
5530   n = GNUNET_CONTAINER_multipeermap_get (neighbours,
5531                                          pid);
5532   if (NULL == n)
5533     return NULL;
5534   for (struct Queue *pos = n->queue_head;
5535        NULL != pos;
5536        pos = pos->next_neighbour)
5537   {
5538     if (0 == strcmp (pos->address,
5539                      address))
5540       return pos;
5541   }
5542   return NULL;
5543 }
5544
5545
5546 /**
5547  * Task run periodically to check whether the validity of the given queue has
5548  * run its course. If so, finds either another queue to take over, or clears
5549  * the neighbour's `core_visible` flag. In the latter case, gives DV routes a
5550  * chance to take over, and if that fails, notifies CORE about the disconnect.
5551  *
5552  * @param cls a `struct Queue`
5553  */
5554 static void
5555 core_queue_visibility_check (void *cls)
5556 {
5557   struct Queue *q = cls;
5558
5559   q->visibility_task = NULL;
5560   if (0 != GNUNET_TIME_absolute_get_remaining (q->validated_until).rel_value_us)
5561   {
5562     q->visibility_task
5563       = GNUNET_SCHEDULER_add_at (q->validated_until,
5564                                  &core_queue_visibility_check,
5565                                  q);
5566     return;
5567   }
5568   update_neighbour_core_visibility (q->neighbour);
5569 }
5570
5571
5572 /**
5573  * Check whether the CORE visibility of @a n should change.  Finds either a
5574  * queue to preserve the visibility, or clears the neighbour's `core_visible`
5575  * flag. In the latter case, gives DV routes a chance to take over, and if
5576  * that fails, notifies CORE about the disconnect.  If so, check whether we
5577  * need to notify CORE.
5578  *
5579  * @param n neighbour to perform the check for
5580  */
5581 static void
5582 update_neighbour_core_visibility (struct Neighbour *n)
5583 {
5584   struct DistanceVector *dv;
5585
5586   GNUNET_assert (GNUNET_YES == n->core_visible);
5587   /* Check if _any_ queue of this neighbour is still valid, if so, schedule
5588      the #core_queue_visibility_check() task for that queue */
5589   for (struct Queue *q = n->queue_head;
5590        NULL != q;
5591        q = q->next_neighbour)
5592   {
5593     if (0 != GNUNET_TIME_absolute_get_remaining (q->validated_until).rel_value_us)
5594     {
5595       /* found a valid queue, use this one */
5596       q->visibility_task
5597         = GNUNET_SCHEDULER_add_at (q->validated_until,
5598                                    &core_queue_visibility_check,
5599                                    q);
5600       return;
5601     }
5602   }
5603   n->core_visible = GNUNET_NO;
5604
5605   /* Check if _any_ DV route to this neighbour is currently
5606      valid, if so, do NOT tell core about the loss of direct
5607      connectivity (DV still counts!) */
5608   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes,
5609                                           &n->pid);
5610   if (GNUNET_YES == dv->core_visible)
5611     return;
5612   /* Nothing works anymore, need to tell CORE about the loss of
5613      connectivity! */
5614   cores_send_disconnect_info (&n->pid);
5615 }
5616
5617
5618 /**
5619  * Communicator gave us a transport address validation response.  Process the request.
5620  *
5621  * @param cls a `struct CommunicatorMessageContext` (must call #finish_cmc_handling() when done)
5622  * @param tvr the message that was received
5623  */
5624 static void
5625 handle_validation_response (void *cls,
5626                             const struct TransportValidationResponse *tvr)
5627 {
5628   struct CommunicatorMessageContext *cmc = cls;
5629   struct ValidationState *vs;
5630   struct CheckKnownChallengeContext ckac = {
5631     .challenge = &tvr->challenge,
5632     .vs = NULL
5633   };
5634   struct GNUNET_TIME_Absolute origin_time;
5635   struct Queue *q;
5636   struct DistanceVector *dv;
5637
5638   /* check this is one of our challenges */
5639   (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
5640                                                      &cmc->im.sender,
5641                                                      &check_known_challenge,
5642                                                      &ckac);
5643   if (NULL == (vs = ckac.vs))
5644   {
5645     /* This can happen simply if we 'forgot' the challenge by now,
5646        i.e. because we received the validation response twice */
5647     GNUNET_STATISTICS_update (GST_stats,
5648                               "# Validations dropped, challenge unknown",
5649                               1,
5650                               GNUNET_NO);
5651     finish_cmc_handling (cmc);
5652     return;
5653   }
5654
5655   /* sanity check on origin time */
5656   origin_time = GNUNET_TIME_absolute_ntoh (tvr->origin_time);
5657   if ( (origin_time.abs_value_us < vs->first_challenge_use.abs_value_us) ||
5658        (origin_time.abs_value_us > vs->last_challenge_use.abs_value_us) )
5659   {
5660     GNUNET_break_op (0);
5661     finish_cmc_handling (cmc);
5662     return;
5663   }
5664
5665   {
5666     /* check signature */
5667     struct TransportValidationPS tvp = {
5668       .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE),
5669       .purpose.size = htonl (sizeof (tvp)),
5670       .validity_duration = tvr->validity_duration,
5671       .challenge = tvr->challenge
5672     };
5673
5674     if (GNUNET_OK !=
5675         GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE,
5676                                     &tvp.purpose,
5677                                     &tvr->signature,
5678                                     &cmc->im.sender.public_key))
5679     {
5680       GNUNET_break_op (0);
5681       finish_cmc_handling (cmc);
5682       return;
5683     }
5684   }
5685
5686   /* validity is capped by our willingness to keep track of the
5687      validation entry and the maximum the other peer allows */
5688   vs->valid_until
5689     = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_min (GNUNET_TIME_relative_ntoh (tvr->validity_duration),
5690                                                                   MAX_ADDRESS_VALID_UNTIL));
5691   vs->validated_until
5692     = GNUNET_TIME_absolute_min (vs->valid_until,
5693                                 GNUNET_TIME_relative_to_absolute (ADDRESS_VALIDATION_LIFETIME));
5694   vs->validation_rtt = GNUNET_TIME_absolute_get_duration (origin_time);
5695   vs->challenge_backoff = GNUNET_TIME_UNIT_ZERO;
5696   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
5697                               &vs->challenge,
5698                               sizeof (vs->challenge));
5699   vs->first_challenge_use = GNUNET_TIME_absolute_subtract (vs->validated_until,
5700                                                            GNUNET_TIME_relative_multiply (vs->validation_rtt,
5701                                                                                           VALIDATION_RTT_BUFFER_FACTOR));
5702   vs->last_challenge_use = GNUNET_TIME_UNIT_ZERO_ABS; /* challenge was not yet used */
5703   update_next_challenge_time (vs,
5704                               vs->first_challenge_use);
5705   vs->sc = GNUNET_PEERSTORE_store (peerstore,
5706                                    "transport",
5707                                    &cmc->im.sender,
5708                                    GNUNET_PEERSTORE_TRANSPORT_URLADDRESS_KEY,
5709                                    vs->address,
5710                                    strlen (vs->address) + 1,
5711                                    vs->valid_until,
5712                                    GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
5713                                    &peerstore_store_validation_cb,
5714                                    vs);
5715   finish_cmc_handling (cmc);
5716
5717   /* Finally, we now possibly have a confirmed (!) working queue,
5718      update queue status (if queue still is around) */
5719   q = find_queue (&vs->pid,
5720                   vs->address);
5721   if (NULL == q)
5722   {
5723     GNUNET_STATISTICS_update (GST_stats,
5724                               "# Queues lost at time of successful validation",
5725                               1,
5726                               GNUNET_NO);
5727     return;
5728   }
5729   q->validated_until = vs->validated_until;
5730   q->rtt = vs->validation_rtt;
5731   if (GNUNET_NO != q->neighbour->core_visible)
5732     return; /* nothing changed, we are done here */
5733   q->neighbour->core_visible = GNUNET_YES;
5734   q->visibility_task
5735     = GNUNET_SCHEDULER_add_at (q->validated_until,
5736                                &core_queue_visibility_check,
5737                                q);
5738   /* Check if _any_ DV route to this neighbour is
5739      currently valid, if so, do NOT tell core anything! */
5740   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes,
5741                                           &q->neighbour->pid);
5742   if (GNUNET_YES == dv->core_visible)
5743     return; /* nothing changed, done */
5744   /* We lacked a confirmed connection to the neighbour
5745      before, so tell CORE about it (finally!) */
5746   cores_send_connect_info (&q->neighbour->pid,
5747                            GNUNET_BANDWIDTH_ZERO);
5748 }
5749
5750
5751 /**
5752  * Incoming meessage.  Process the request.
5753  *
5754  * @param im the send message that was received
5755  */
5756 static void
5757 handle_incoming_msg (void *cls,
5758                      const struct GNUNET_TRANSPORT_IncomingMessage *im)
5759 {
5760   struct TransportClient *tc = cls;
5761   struct CommunicatorMessageContext *cmc = GNUNET_new (struct CommunicatorMessageContext);
5762
5763   cmc->tc = tc;
5764   cmc->im = *im;
5765   demultiplex_with_cmc (cmc,
5766                         (const struct GNUNET_MessageHeader *) &im[1]);
5767 }
5768
5769
5770 /**
5771  * Given an inbound message @a msg from a communicator @a cmc,
5772  * demultiplex it based on the type calling the right handler.
5773  *
5774  * @param cmc context for demultiplexing
5775  * @param msg message to demultiplex
5776  */
5777 static void
5778 demultiplex_with_cmc (struct CommunicatorMessageContext *cmc,
5779                       const struct GNUNET_MessageHeader *msg)
5780 {
5781   struct GNUNET_MQ_MessageHandler handlers[] = {
5782     GNUNET_MQ_hd_var_size (fragment_box,
5783                            GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT,
5784                            struct TransportFragmentBox,
5785                            &cmc),
5786     GNUNET_MQ_hd_fixed_size (fragment_ack,
5787                              GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK,
5788                              struct TransportFragmentAckMessage,
5789                              &cmc),
5790     GNUNET_MQ_hd_var_size (reliability_box,
5791                            GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX,
5792                            struct TransportReliabilityBox,
5793                            &cmc),
5794     GNUNET_MQ_hd_fixed_size (reliability_ack,
5795                              GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK,
5796                              struct TransportReliabilityAckMessage,
5797                              &cmc),
5798     GNUNET_MQ_hd_var_size (backchannel_encapsulation,
5799                            GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION,
5800                            struct TransportBackchannelEncapsulationMessage,
5801                            &cmc),
5802     GNUNET_MQ_hd_var_size (dv_learn,
5803                            GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN,
5804                            struct TransportDVLearn,
5805                            &cmc),
5806     GNUNET_MQ_hd_var_size (dv_box,
5807                            GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX,
5808                            struct TransportDVBox,
5809                            &cmc),
5810     GNUNET_MQ_hd_fixed_size (validation_challenge,
5811                              GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE,
5812                              struct TransportValidationChallenge,
5813                              &cmc),
5814     GNUNET_MQ_hd_fixed_size (validation_response,
5815                              GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE,
5816                              struct TransportValidationResponse,
5817                              &cmc),
5818     GNUNET_MQ_handler_end()
5819   };
5820   int ret;
5821
5822   ret = GNUNET_MQ_handle_message (handlers,
5823                                   msg);
5824   if (GNUNET_SYSERR == ret)
5825   {
5826     GNUNET_break (0);
5827     GNUNET_SERVICE_client_drop (cmc->tc->client);
5828     GNUNET_free (cmc);
5829     return;
5830   }
5831   if (GNUNET_NO == ret)
5832   {
5833     /* unencapsulated 'raw' message */
5834     handle_raw_message (&cmc,
5835                         msg);
5836   }
5837 }
5838
5839
5840 /**
5841  * New queue became available.  Check message.
5842  *
5843  * @param cls the client
5844  * @param aqm the send message that was sent
5845  */
5846 static int
5847 check_add_queue_message (void *cls,
5848                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
5849 {
5850   struct TransportClient *tc = cls;
5851
5852   if (CT_COMMUNICATOR != tc->type)
5853   {
5854     GNUNET_break (0);
5855     return GNUNET_SYSERR;
5856   }
5857   GNUNET_MQ_check_zero_termination (aqm);
5858   return GNUNET_OK;
5859 }
5860
5861
5862 /**
5863  * Bandwidth tracker informs us that the delay until we should receive
5864  * more has changed.
5865  *
5866  * @param cls a `struct Queue` for which the delay changed
5867  */
5868 static void
5869 tracker_update_in_cb (void *cls)
5870 {
5871   struct Queue *queue = cls;
5872   struct GNUNET_TIME_Relative in_delay;
5873   unsigned int rsize;
5874
5875   rsize = (0 == queue->mtu) ? IN_PACKET_SIZE_WITHOUT_MTU : queue->mtu;
5876   in_delay = GNUNET_BANDWIDTH_tracker_get_delay (&queue->tracker_in,
5877                                                  rsize);
5878   // FIXME: how exactly do we do inbound flow control?
5879 }
5880
5881
5882 /**
5883  * If necessary, generates the UUID for a @a pm
5884  *
5885  * @param pm pending message to generate UUID for.
5886  */
5887 static void
5888 set_pending_message_uuid (struct PendingMessage *pm)
5889 {
5890   if (pm->msg_uuid_set)
5891     return;
5892   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
5893                               &pm->msg_uuid,
5894                               sizeof (pm->msg_uuid));
5895   pm->msg_uuid_set = GNUNET_YES;
5896 }
5897
5898
5899 /**
5900  * Fragment the given @a pm to the given @a mtu.  Adds
5901  * additional fragments to the neighbour as well. If the
5902  * @a mtu is too small, generates and error for the @a pm
5903  * and returns NULL.
5904  *
5905  * @param pm pending message to fragment for transmission
5906  * @param mtu MTU to apply
5907  * @return new message to transmit
5908  */
5909 static struct PendingMessage *
5910 fragment_message (struct PendingMessage *pm,
5911                   uint16_t mtu)
5912 {
5913   struct PendingMessage *ff;
5914
5915   set_pending_message_uuid (pm);
5916
5917   /* This invariant is established in #handle_add_queue_message() */
5918   GNUNET_assert (mtu > sizeof (struct TransportFragmentBox));
5919
5920   /* select fragment for transmission, descending the tree if it has
5921      been expanded until we are at a leaf or at a fragment that is small enough */
5922   ff = pm;
5923   while ( ( (ff->bytes_msg > mtu) ||
5924             (pm == ff) ) &&
5925           (ff->frag_off == ff->bytes_msg) &&
5926           (NULL != ff->head_frag) )
5927   {
5928     ff = ff->head_frag; /* descent into fragmented fragments */
5929   }
5930
5931   if ( ( (ff->bytes_msg > mtu) ||
5932          (pm == ff) ) &&
5933        (pm->frag_off < pm->bytes_msg) )
5934   {
5935     /* Did not yet calculate all fragments, calculate next fragment */
5936     struct PendingMessage *frag;
5937     struct TransportFragmentBox tfb;
5938     const char *orig;
5939     char *msg;
5940     uint16_t fragmax;
5941     uint16_t fragsize;
5942     uint16_t msize;
5943     uint16_t xoff = 0;
5944
5945     orig = (const char *) &ff[1];
5946     msize = ff->bytes_msg;
5947     if (pm != ff)
5948     {
5949       const struct TransportFragmentBox *tfbo;
5950
5951       tfbo = (const struct TransportFragmentBox *) orig;
5952       orig += sizeof (struct TransportFragmentBox);
5953       msize -= sizeof (struct TransportFragmentBox);
5954       xoff = ntohs (tfbo->frag_off);
5955     }
5956     fragmax = mtu - sizeof (struct TransportFragmentBox);
5957     fragsize = GNUNET_MIN (msize - ff->frag_off,
5958                            fragmax);
5959     frag = GNUNET_malloc (sizeof (struct PendingMessage) +
5960                           sizeof (struct TransportFragmentBox) +
5961                           fragsize);
5962     frag->target = pm->target;
5963     frag->frag_parent = ff;
5964     frag->timeout = pm->timeout;
5965     frag->bytes_msg = sizeof (struct TransportFragmentBox) + fragsize;
5966     frag->pmt = PMT_FRAGMENT_BOX;
5967     msg = (char *) &frag[1];
5968     tfb.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT);
5969     tfb.header.size = htons (sizeof (struct TransportFragmentBox) +
5970                              fragsize);
5971     tfb.frag_uuid = htonl (pm->frag_uuidgen++);
5972     tfb.msg_uuid = pm->msg_uuid;
5973     tfb.frag_off = htons (ff->frag_off + xoff);
5974     tfb.msg_size = htons (pm->bytes_msg);
5975     memcpy (msg,
5976             &tfb,
5977             sizeof (tfb));
5978     memcpy (&msg[sizeof (tfb)],
5979             &orig[ff->frag_off],
5980             fragsize);
5981     GNUNET_CONTAINER_MDLL_insert (frag,
5982                                   ff->head_frag,
5983                                   ff->tail_frag,
5984                                   frag);
5985     ff->frag_off += fragsize;
5986     ff = frag;
5987   }
5988
5989   /* Move head to the tail and return it */
5990   GNUNET_CONTAINER_MDLL_remove (frag,
5991                                 ff->frag_parent->head_frag,
5992                                 ff->frag_parent->tail_frag,
5993                                 ff);
5994   GNUNET_CONTAINER_MDLL_insert_tail (frag,
5995                                      ff->frag_parent->head_frag,
5996                                      ff->frag_parent->tail_frag,
5997                                      ff);
5998   return ff;
5999 }
6000
6001
6002 /**
6003  * Reliability-box the given @a pm. On error (can there be any), NULL
6004  * may be returned, otherwise the "replacement" for @a pm (which
6005  * should then be added to the respective neighbour's queue instead of
6006  * @a pm).  If the @a pm is already fragmented or reliability boxed,
6007  * or itself an ACK, this function simply returns @a pm.
6008  *
6009  * @param pm pending message to box for transmission over unreliabile queue
6010  * @return new message to transmit
6011  */
6012 static struct PendingMessage *
6013 reliability_box_message (struct PendingMessage *pm)
6014 {
6015   struct TransportReliabilityBox rbox;
6016   struct PendingMessage *bpm;
6017   char *msg;
6018
6019   if (PMT_CORE != pm->pmt)
6020     return pm;  /* already fragmented or reliability boxed, or control message: do nothing */
6021   if (NULL != pm->bpm)
6022     return pm->bpm; /* already computed earlier: do nothing */
6023   GNUNET_assert (NULL == pm->head_frag);
6024   if (pm->bytes_msg + sizeof (rbox) > UINT16_MAX)
6025   {
6026     /* failed hard */
6027     GNUNET_break (0);
6028     client_send_response (pm,
6029                           GNUNET_NO,
6030                           0);
6031     return NULL;
6032   }
6033   bpm = GNUNET_malloc (sizeof (struct PendingMessage) +
6034                        sizeof (rbox) +
6035                        pm->bytes_msg);
6036   bpm->target = pm->target;
6037   bpm->frag_parent = pm;
6038   GNUNET_CONTAINER_MDLL_insert (frag,
6039                                 pm->head_frag,
6040                                 pm->tail_frag,
6041                                 bpm);
6042   bpm->timeout = pm->timeout;
6043   bpm->pmt = PMT_RELIABILITY_BOX;
6044   bpm->bytes_msg = pm->bytes_msg + sizeof (rbox);
6045   set_pending_message_uuid (bpm);
6046   rbox.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX);
6047   rbox.header.size = htons (sizeof (rbox) + pm->bytes_msg);
6048   rbox.ack_countdown = htonl (0); // FIXME: implement ACK countdown support
6049   rbox.msg_uuid = pm->msg_uuid;
6050   msg = (char *) &bpm[1];
6051   memcpy (msg,
6052           &rbox,
6053           sizeof (rbox));
6054   memcpy (&msg[sizeof (rbox)],
6055           &pm[1],
6056           pm->bytes_msg);
6057   pm->bpm = bpm;
6058   return bpm;
6059 }
6060
6061
6062 /**
6063  * We believe we are ready to transmit a message on a queue. Double-checks
6064  * with the queue's "tracker_out" and then gives the message to the
6065  * communicator for transmission (updating the tracker, and re-scheduling
6066  * itself if applicable).
6067  *
6068  * @param cls the `struct Queue` to process transmissions for
6069  */
6070 static void
6071 transmit_on_queue (void *cls)
6072 {
6073   struct Queue *queue = cls;
6074   struct Neighbour *n = queue->neighbour;
6075   struct PendingMessage *pm;
6076   struct PendingMessage *s;
6077   uint32_t overhead;
6078
6079   queue->transmit_task = NULL;
6080   if (NULL == (pm = n->pending_msg_head))
6081   {
6082     /* no message pending, nothing to do here! */
6083     return;
6084   }
6085   schedule_transmit_on_queue (queue);
6086   if (NULL != queue->transmit_task)
6087     return; /* do it later */
6088   overhead = 0;
6089   if (GNUNET_TRANSPORT_CC_RELIABLE != queue->tc->details.communicator.cc)
6090     overhead += sizeof (struct TransportReliabilityBox);
6091   s = pm;
6092   if ( ( (0 != queue->mtu) &&
6093          (pm->bytes_msg + overhead > queue->mtu) ) ||
6094        (pm->bytes_msg > UINT16_MAX - sizeof (struct GNUNET_TRANSPORT_SendMessageTo)) ||
6095        (NULL != pm->head_frag /* fragments already exist, should
6096                                  respect that even if MTU is 0 for
6097                                  this queue */) )
6098     s = fragment_message (s,
6099                           (0 == queue->mtu)
6100                           ? UINT16_MAX - sizeof (struct GNUNET_TRANSPORT_SendMessageTo)
6101                           : queue->mtu);
6102   if (NULL == s)
6103   {
6104     /* Fragmentation failed, try next message... */
6105     schedule_transmit_on_queue (queue);
6106     return;
6107   }
6108   if (GNUNET_TRANSPORT_CC_RELIABLE != queue->tc->details.communicator.cc)
6109     s = reliability_box_message (s);
6110   if (NULL == s)
6111   {
6112     /* Reliability boxing failed, try next message... */
6113     schedule_transmit_on_queue (queue);
6114     return;
6115   }
6116
6117   /* Pass 's' for transission to the communicator */
6118   queue_send_msg (queue,
6119                   s,
6120                   &s[1],
6121                   s->bytes_msg);
6122   // FIXME: do something similar to the logic below
6123   // in defragmentation / reliability ACK handling!
6124
6125   /* Check if this transmission somehow conclusively finished handing 'pm'
6126      even without any explicit ACKs */
6127   if ( (PMT_CORE == s->pmt) &&
6128        (GNUNET_TRANSPORT_CC_RELIABLE == queue->tc->details.communicator.cc) )
6129   {
6130     /* Full message sent, and over reliabile channel */
6131     client_send_response (pm,
6132                           GNUNET_YES,
6133                           pm->bytes_msg);
6134   }
6135   else if ( (GNUNET_TRANSPORT_CC_RELIABLE == queue->tc->details.communicator.cc) &&
6136             (PMT_FRAGMENT_BOX == s->pmt) )
6137   {
6138     struct PendingMessage *pos;
6139
6140     /* Fragment sent over reliabile channel */
6141     free_fragment_tree (s);
6142     pos = s->frag_parent;
6143     GNUNET_CONTAINER_MDLL_remove (frag,
6144                                   pos->head_frag,
6145                                   pos->tail_frag,
6146                                   s);
6147     GNUNET_free (s);
6148     /* check if subtree is done */
6149     while ( (NULL == pos->head_frag) &&
6150             (pos->frag_off == pos->bytes_msg) &&
6151             (pos != pm) )
6152     {
6153       s = pos;
6154       pos = s->frag_parent;
6155       GNUNET_CONTAINER_MDLL_remove (frag,
6156                                     pos->head_frag,
6157                                     pos->tail_frag,
6158                                     s);
6159       GNUNET_free (s);
6160     }
6161
6162     /* Was this the last applicable fragmment? */
6163     if ( (NULL == pm->head_frag) &&
6164          (pm->frag_off == pm->bytes_msg) )
6165       client_send_response (pm,
6166                             GNUNET_YES,
6167                             pm->bytes_msg /* FIXME: calculate and add overheads! */);
6168   }
6169   else if (PMT_CORE != pm->pmt)
6170   {
6171     /* This was an acknowledgement of some type, always free */
6172     free_pending_message (pm);
6173   }
6174   else
6175   {
6176     /* message not finished, waiting for acknowledgement */
6177     struct Neighbour *neighbour = pm->target;
6178     /* Update time by which we might retransmit 's' based on queue
6179        characteristics (i.e. RTT); it takes one RTT for the message to
6180        arrive and the ACK to come back in the best case; but the other
6181        side is allowed to delay ACKs by 2 RTTs, so we use 4 RTT before
6182        retransmitting.  Note that in the future this heuristic should
6183        likely be improved further (measure RTT stability, consider
6184        message urgency and size when delaying ACKs, etc.) */
6185     s->next_attempt = GNUNET_TIME_relative_to_absolute
6186       (GNUNET_TIME_relative_multiply (queue->rtt,
6187                                       4));
6188     if (s == pm)
6189     {
6190       struct PendingMessage *pos;
6191
6192       /* re-insert sort in neighbour list */
6193       GNUNET_CONTAINER_MDLL_remove (neighbour,
6194                                     neighbour->pending_msg_head,
6195                                     neighbour->pending_msg_tail,
6196                                     pm);
6197       pos = neighbour->pending_msg_tail;
6198       while ( (NULL != pos) &&
6199               (pm->next_attempt.abs_value_us > pos->next_attempt.abs_value_us) )
6200         pos = pos->prev_neighbour;
6201       GNUNET_CONTAINER_MDLL_insert_after (neighbour,
6202                                           neighbour->pending_msg_head,
6203                                           neighbour->pending_msg_tail,
6204                                           pos,
6205                                           pm);
6206     }
6207     else
6208     {
6209       /* re-insert sort in fragment list */
6210       struct PendingMessage *fp = s->frag_parent;
6211       struct PendingMessage *pos;
6212
6213       GNUNET_CONTAINER_MDLL_remove (frag,
6214                                     fp->head_frag,
6215                                     fp->tail_frag,
6216                                     s);
6217       pos = fp->tail_frag;
6218       while ( (NULL != pos) &&
6219               (s->next_attempt.abs_value_us > pos->next_attempt.abs_value_us) )
6220         pos = pos->prev_frag;
6221       GNUNET_CONTAINER_MDLL_insert_after (frag,
6222                                           fp->head_frag,
6223                                           fp->tail_frag,
6224                                           pos,
6225                                           s);
6226     }
6227   }
6228
6229   /* finally, re-schedule queue transmission task itself */
6230   schedule_transmit_on_queue (queue);
6231 }
6232
6233
6234 /**
6235  * Bandwidth tracker informs us that the delay until we
6236  * can transmit again changed.
6237  *
6238  * @param cls a `struct Queue` for which the delay changed
6239  */
6240 static void
6241 tracker_update_out_cb (void *cls)
6242 {
6243   struct Queue *queue = cls;
6244   struct Neighbour *n = queue->neighbour;
6245
6246   if (NULL == n->pending_msg_head)
6247   {
6248     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
6249                 "Bandwidth allocation updated for empty transmission queue `%s'\n",
6250                 queue->address);
6251     return; /* no message pending, nothing to do here! */
6252   }
6253   GNUNET_SCHEDULER_cancel (queue->transmit_task);
6254   queue->transmit_task = NULL;
6255   schedule_transmit_on_queue (queue);
6256 }
6257
6258
6259 /**
6260  * Bandwidth tracker informs us that excessive outbound bandwidth was
6261  * allocated which is not being used.
6262  *
6263  * @param cls a `struct Queue` for which the excess was noted
6264  */
6265 static void
6266 tracker_excess_out_cb (void *cls)
6267 {
6268   (void) cls;
6269
6270   /* FIXME: trigger excess bandwidth report to core? Right now,
6271      this is done internally within transport_api2_core already,
6272      but we probably want to change the logic and trigger it
6273      from here via a message instead! */
6274   /* TODO: maybe inform someone at this point? */
6275   GNUNET_STATISTICS_update (GST_stats,
6276                             "# Excess outbound bandwidth reported",
6277                             1,
6278                             GNUNET_NO);
6279 }
6280
6281
6282
6283 /**
6284  * Bandwidth tracker informs us that excessive inbound bandwidth was allocated
6285  * which is not being used.
6286  *
6287  * @param cls a `struct Queue` for which the excess was noted
6288  */
6289 static void
6290 tracker_excess_in_cb (void *cls)
6291 {
6292   (void) cls;
6293
6294   /* TODO: maybe inform somone at this point? */
6295   GNUNET_STATISTICS_update (GST_stats,
6296                             "# Excess inbound bandwidth reported",
6297                             1,
6298                             GNUNET_NO);
6299 }
6300
6301
6302 /**
6303  * Queue to a peer went down.  Process the request.
6304  *
6305  * @param cls the client
6306  * @param dqm the send message that was sent
6307  */
6308 static void
6309 handle_del_queue_message (void *cls,
6310                           const struct GNUNET_TRANSPORT_DelQueueMessage *dqm)
6311 {
6312   struct TransportClient *tc = cls;
6313
6314   if (CT_COMMUNICATOR != tc->type)
6315   {
6316     GNUNET_break (0);
6317     GNUNET_SERVICE_client_drop (tc->client);
6318     return;
6319   }
6320   for (struct Queue *queue = tc->details.communicator.queue_head;
6321        NULL != queue;
6322        queue = queue->next_client)
6323   {
6324     struct Neighbour *neighbour = queue->neighbour;
6325
6326     if ( (dqm->qid != queue->qid) ||
6327          (0 != GNUNET_memcmp (&dqm->receiver,
6328                               &neighbour->pid)) )
6329       continue;
6330     free_queue (queue);
6331     GNUNET_SERVICE_client_continue (tc->client);
6332     return;
6333   }
6334   GNUNET_break (0);
6335   GNUNET_SERVICE_client_drop (tc->client);
6336 }
6337
6338
6339 /**
6340  * Message was transmitted.  Process the request.
6341  *
6342  * @param cls the client
6343  * @param sma the send message that was sent
6344  */
6345 static void
6346 handle_send_message_ack (void *cls,
6347                          const struct GNUNET_TRANSPORT_SendMessageToAck *sma)
6348 {
6349   struct TransportClient *tc = cls;
6350   struct QueueEntry *qe;
6351
6352   if (CT_COMMUNICATOR != tc->type)
6353   {
6354     GNUNET_break (0);
6355     GNUNET_SERVICE_client_drop (tc->client);
6356     return;
6357   }
6358
6359   /* find our queue entry matching the ACK */
6360   qe = NULL;
6361   for (struct Queue *queue = tc->details.communicator.queue_head;
6362        NULL != queue;
6363        queue = queue->next_client)
6364   {
6365     if (0 != GNUNET_memcmp (&queue->neighbour->pid,
6366                             &sma->receiver))
6367       continue;
6368     for (struct QueueEntry *qep = queue->queue_head;
6369          NULL != qep;
6370          qep = qep->next)
6371     {
6372       if (qep->mid != sma->mid)
6373         continue;
6374       qe = qep;
6375       break;
6376     }
6377     break;
6378   }
6379   if (NULL == qe)
6380   {
6381     /* this should never happen */
6382     GNUNET_break (0);
6383     GNUNET_SERVICE_client_drop (tc->client);
6384     return;
6385   }
6386   GNUNET_CONTAINER_DLL_remove (qe->queue->queue_head,
6387                                qe->queue->queue_tail,
6388                                qe);
6389   qe->queue->queue_length--;
6390   tc->details.communicator.total_queue_length--;
6391   GNUNET_SERVICE_client_continue (tc->client);
6392
6393   /* if applicable, resume transmissions that waited on ACK */
6394   if (COMMUNICATOR_TOTAL_QUEUE_LIMIT - 1 == tc->details.communicator.total_queue_length)
6395   {
6396     /* Communicator dropped below threshold, resume all queues */
6397     GNUNET_STATISTICS_update (GST_stats,
6398                               "# Transmission throttled due to communicator queue limit",
6399                               -1,
6400                               GNUNET_NO);
6401     for (struct Queue *queue = tc->details.communicator.queue_head;
6402          NULL != queue;
6403          queue = queue->next_client)
6404       schedule_transmit_on_queue (queue);
6405   }
6406   else if (QUEUE_LENGTH_LIMIT - 1 == qe->queue->queue_length)
6407   {
6408     /* queue dropped below threshold; only resume this one queue */
6409     GNUNET_STATISTICS_update (GST_stats,
6410                               "# Transmission throttled due to queue queue limit",
6411                               -1,
6412                               GNUNET_NO);
6413     schedule_transmit_on_queue (qe->queue);
6414   }
6415
6416   /* TODO: we also should react on the status! */
6417   // FIXME: this probably requires queue->pm = s assignment!
6418   // FIXME: react to communicator status about transmission request. We got:
6419   sma->status; // OK success, SYSERR failure
6420
6421   GNUNET_free (qe);
6422 }
6423
6424
6425 /**
6426  * Iterator telling new MONITOR client about all existing
6427  * queues to peers.
6428  *
6429  * @param cls the new `struct TransportClient`
6430  * @param pid a connected peer
6431  * @param value the `struct Neighbour` with more information
6432  * @return #GNUNET_OK (continue to iterate)
6433  */
6434 static int
6435 notify_client_queues (void *cls,
6436                       const struct GNUNET_PeerIdentity *pid,
6437                       void *value)
6438 {
6439   struct TransportClient *tc = cls;
6440   struct Neighbour *neighbour = value;
6441
6442   GNUNET_assert (CT_MONITOR == tc->type);
6443   for (struct Queue *q = neighbour->queue_head;
6444        NULL != q;
6445        q = q->next_neighbour)
6446   {
6447     struct MonitorEvent me = {
6448       .rtt = q->rtt,
6449       .cs = q->cs,
6450       .num_msg_pending = q->num_msg_pending,
6451       .num_bytes_pending = q->num_bytes_pending
6452     };
6453
6454     notify_monitor (tc,
6455                     pid,
6456                     q->address,
6457                     q->nt,
6458                     &me);
6459   }
6460   return GNUNET_OK;
6461 }
6462
6463
6464 /**
6465  * Initialize a monitor client.
6466  *
6467  * @param cls the client
6468  * @param start the start message that was sent
6469  */
6470 static void
6471 handle_monitor_start (void *cls,
6472                       const struct GNUNET_TRANSPORT_MonitorStart *start)
6473 {
6474   struct TransportClient *tc = cls;
6475
6476   if (CT_NONE != tc->type)
6477   {
6478     GNUNET_break (0);
6479     GNUNET_SERVICE_client_drop (tc->client);
6480     return;
6481   }
6482   tc->type = CT_MONITOR;
6483   tc->details.monitor.peer = start->peer;
6484   tc->details.monitor.one_shot = ntohl (start->one_shot);
6485   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
6486                                          &notify_client_queues,
6487                                          tc);
6488   GNUNET_SERVICE_client_mark_monitor (tc->client);
6489   GNUNET_SERVICE_client_continue (tc->client);
6490 }
6491
6492
6493 /**
6494  * Find transport client providing communication service
6495  * for the protocol @a prefix.
6496  *
6497  * @param prefix communicator name
6498  * @return NULL if no such transport client is available
6499  */
6500 static struct TransportClient *
6501 lookup_communicator (const char *prefix)
6502 {
6503   for (struct TransportClient *tc = clients_head;
6504        NULL != tc;
6505        tc = tc->next)
6506   {
6507     if (CT_COMMUNICATOR != tc->type)
6508       continue;
6509     if (0 == strcmp (prefix,
6510                      tc->details.communicator.address_prefix))
6511       return tc;
6512   }
6513   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
6514               "Somone suggested use of communicator for `%s', but we do not have such a communicator!\n",
6515               prefix);
6516   return NULL;
6517 }
6518
6519
6520 /**
6521  * Signature of a function called with a communicator @a address of a peer
6522  * @a pid that an application wants us to connect to.
6523  *
6524  * @param pid target peer
6525  * @param address the address to try
6526  */
6527 static void
6528 suggest_to_connect (const struct GNUNET_PeerIdentity *pid,
6529                     const char *address)
6530 {
6531   static uint32_t idgen;
6532   struct TransportClient *tc;
6533   char *prefix;
6534   struct GNUNET_TRANSPORT_CreateQueue *cqm;
6535   struct GNUNET_MQ_Envelope *env;
6536   size_t alen;
6537
6538   prefix = GNUNET_HELLO_address_to_prefix (address);
6539   if (NULL == prefix)
6540   {
6541     GNUNET_break (0); /* We got an invalid address!? */
6542     return;
6543   }
6544   tc = lookup_communicator (prefix);
6545   if (NULL == tc)
6546   {
6547     GNUNET_STATISTICS_update (GST_stats,
6548                               "# Suggestions ignored due to missing communicator",
6549                               1,
6550                               GNUNET_NO);
6551     return;
6552   }
6553   /* forward suggestion for queue creation to communicator */
6554   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
6555               "Request #%u for `%s' communicator to create queue to `%s'\n",
6556               (unsigned int) idgen,
6557               prefix,
6558               address);
6559   alen = strlen (address) + 1;
6560   env = GNUNET_MQ_msg_extra (cqm,
6561                              alen,
6562                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE);
6563   cqm->request_id = htonl (idgen++);
6564   cqm->receiver = *pid;
6565   memcpy (&cqm[1],
6566           address,
6567           alen);
6568   GNUNET_MQ_send (tc->mq,
6569                   env);
6570 }
6571
6572
6573 /**
6574  * The queue @a q (which matches the peer and address in @a vs) is
6575  * ready for queueing. We should now queue the validation request.
6576  *
6577  * @param q queue to send on
6578  * @param vs state to derive validation challenge from
6579  */
6580 static void
6581 validation_transmit_on_queue (struct Queue *q,
6582                               struct ValidationState *vs)
6583 {
6584   struct TransportValidationChallenge tvc;
6585
6586   vs->last_challenge_use = GNUNET_TIME_absolute_get ();
6587   tvc.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE);
6588   tvc.header.size = htons (sizeof (tvc));
6589   tvc.reserved = htonl (0);
6590   tvc.challenge = vs->challenge;
6591   tvc.sender_time = GNUNET_TIME_absolute_hton (vs->last_challenge_use);
6592   queue_send_msg (q,
6593                   NULL,
6594                   &tvc,
6595                   sizeof (tvc));
6596 }
6597
6598
6599 /**
6600  * Task run periodically to validate some address based on #validation_heap.
6601  *
6602  * @param cls NULL
6603  */
6604 static void
6605 validation_start_cb (void *cls)
6606 {
6607   struct ValidationState *vs;
6608   struct Queue *q;
6609
6610   (void) cls;
6611   validation_task = NULL;
6612   vs = GNUNET_CONTAINER_heap_peek (validation_heap);
6613   /* drop validations past their expiration */
6614   while ( (NULL != vs) &&
6615           (0 == GNUNET_TIME_absolute_get_remaining (vs->valid_until).rel_value_us) )
6616   {
6617     free_validation_state (vs);
6618     vs = GNUNET_CONTAINER_heap_peek (validation_heap);
6619   }
6620   if (NULL == vs)
6621     return; /* woopsie, no more addresses known, should only
6622                happen if we're really a lonely peer */
6623   q = find_queue (&vs->pid,
6624                   vs->address);
6625   if (NULL == q)
6626   {
6627     vs->awaiting_queue = GNUNET_YES;
6628     suggest_to_connect (&vs->pid,
6629                         vs->address);
6630   }
6631   else
6632     validation_transmit_on_queue (q,
6633                                   vs);
6634   /* Finally, reschedule next attempt */
6635   vs->challenge_backoff = GNUNET_TIME_randomized_backoff (vs->challenge_backoff,
6636                                                           MAX_VALIDATION_CHALLENGE_FREQ);
6637   update_next_challenge_time (vs,
6638                               GNUNET_TIME_relative_to_absolute (vs->challenge_backoff));
6639 }
6640
6641
6642 /**
6643  * Closure for #check_connection_quality.
6644  */
6645 struct QueueQualityContext
6646 {
6647   /**
6648    * Set to the @e k'th queue encountered.
6649    */
6650   struct Queue *q;
6651
6652   /**
6653    * Set to the number of quality queues encountered.
6654    */
6655   unsigned int quality_count;
6656
6657   /**
6658    * Set to the total number of queues encountered.
6659    */
6660   unsigned int num_queues;
6661
6662   /**
6663    * Decremented for each queue, for selection of the
6664    * k-th queue in @e q.
6665    */
6666   unsigned int k;
6667
6668 };
6669
6670
6671 /**
6672  * Check whether any queue to the given neighbour is
6673  * of a good "quality" and if so, increment the counter.
6674  * Also counts the total number of queues, and returns
6675  * the k-th queue found.
6676  *
6677  * @param cls a `struct QueueQualityContext *` with counters
6678  * @param pid peer this is about
6679  * @param value a `struct Neighbour`
6680  * @return #GNUNET_OK (continue to iterate)
6681  */
6682 static int
6683 check_connection_quality (void *cls,
6684                           const struct GNUNET_PeerIdentity *pid,
6685                           void *value)
6686 {
6687   struct QueueQualityContext *ctx = cls;
6688   struct Neighbour *n = value;
6689   int do_inc;
6690
6691   (void) pid;
6692   do_inc = GNUNET_NO;
6693   for (struct Queue *q = n->queue_head;
6694        NULL != q;
6695        q = q->next_neighbour)
6696   {
6697     if (0 != q->distance)
6698       continue; /* DV does not count */
6699     ctx->num_queues++;
6700     if (0 == ctx->k--)
6701       ctx->q = q;
6702     /* OPTIMIZE-FIXME: in the future, add reliability / goodput
6703        statistics and consider those as well here? */
6704     if (q->rtt.rel_value_us < DV_QUALITY_RTT_THRESHOLD.rel_value_us)
6705       do_inc = GNUNET_YES;
6706   }
6707   if (GNUNET_YES == do_inc)
6708     ctx->quality_count++;
6709   return GNUNET_OK;
6710 }
6711
6712
6713 /**
6714  * Task run when we CONSIDER initiating a DV learn
6715  * process. We first check that sending out a message is
6716  * even possible (queues exist), then that it is desirable
6717  * (if not, reschedule the task for later), and finally
6718  * we may then begin the job.  If there are too many
6719  * entries in the #dvlearn_map, we purge the oldest entry
6720  * using #lle_tail.
6721  *
6722  * @param cls NULL
6723  */
6724 static void
6725 start_dv_learn (void *cls)
6726 {
6727   struct LearnLaunchEntry *lle;
6728   struct QueueQualityContext qqc;
6729   struct TransportDVLearn dvl;
6730
6731   (void) cls;
6732   dvlearn_task = NULL;
6733   if (0 ==
6734       GNUNET_CONTAINER_multipeermap_size (neighbours))
6735     return; /* lost all connectivity, cannot do learning */
6736   qqc.quality_count = 0;
6737   qqc.num_queues = 0;
6738   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
6739                                          &check_connection_quality,
6740                                          &qqc);
6741   if (qqc.quality_count > DV_LEARN_QUALITY_THRESHOLD)
6742   {
6743     struct GNUNET_TIME_Relative delay;
6744     unsigned int factor;
6745
6746     /* scale our retries by how far we are above the threshold */
6747     factor = qqc.quality_count / DV_LEARN_QUALITY_THRESHOLD;
6748     delay = GNUNET_TIME_relative_multiply (DV_LEARN_BASE_FREQUENCY,
6749                                            factor);
6750     dvlearn_task = GNUNET_SCHEDULER_add_delayed (delay,
6751                                                  &start_dv_learn,
6752                                                  NULL);
6753     return;
6754   }
6755   /* remove old entries in #dvlearn_map if it has grown too big */
6756   while (MAX_DV_LEARN_PENDING >=
6757          GNUNET_CONTAINER_multishortmap_size (dvlearn_map))
6758   {
6759     lle = lle_tail;
6760     GNUNET_assert (GNUNET_YES ==
6761                    GNUNET_CONTAINER_multishortmap_remove (dvlearn_map,
6762                                                           &lle->challenge,
6763                                                           lle));
6764     GNUNET_CONTAINER_DLL_remove (lle_head,
6765                                  lle_tail,
6766                                  lle);
6767     GNUNET_free (lle);
6768   }
6769   /* setup data structure for learning */
6770   lle = GNUNET_new (struct LearnLaunchEntry);
6771   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
6772                               &lle->challenge,
6773                               sizeof (lle->challenge));
6774   GNUNET_CONTAINER_DLL_insert (lle_head,
6775                                lle_tail,
6776                                lle);
6777   GNUNET_break (GNUNET_YES ==
6778                 GNUNET_CONTAINER_multishortmap_put (dvlearn_map,
6779                                                     &lle->challenge,
6780                                                     lle,
6781                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
6782   dvl.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN);
6783   dvl.header.size = htons (sizeof (dvl));
6784   dvl.num_hops = htons (0);
6785   dvl.bidirectional = htons (0);
6786   dvl.non_network_delay = GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_ZERO);
6787   {
6788     struct DvInitPS dvip = {
6789       .purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR),
6790       .purpose.size = htonl (sizeof (dvip)),
6791       .challenge = lle->challenge
6792     };
6793
6794     GNUNET_assert (GNUNET_OK ==
6795                    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
6796                                              &dvip.purpose,
6797                                              &dvl.init_sig));
6798   }
6799   dvl.initiator = GST_my_identity;
6800   dvl.challenge = lle->challenge;
6801
6802   qqc.quality_count = 0;
6803   qqc.k = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
6804                                     qqc.num_queues);
6805   qqc.num_queues = 0;
6806   qqc.q = NULL;
6807   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
6808                                          &check_connection_quality,
6809                                          &qqc);
6810   GNUNET_assert (NULL != qqc.q);
6811
6812   /* Do this as close to transmission time as possible! */
6813   lle->launch_time = GNUNET_TIME_absolute_get ();
6814
6815   queue_send_msg (qqc.q,
6816                   NULL,
6817                   &dvl,
6818                   sizeof (dvl));
6819   /* reschedule this job, randomizing the time it runs (but no
6820      actual backoff!) */
6821   dvlearn_task
6822     = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_randomize (DV_LEARN_BASE_FREQUENCY),
6823                                     &start_dv_learn,
6824                                     NULL);
6825 }
6826
6827
6828 /**
6829  * A new queue has been created, check if any address validation
6830  * requests have been waiting for it.
6831  *
6832  * @param cls a `struct Queue`
6833  * @param pid peer concerned (unused)
6834  * @param value a `struct ValidationState`
6835  * @return #GNUNET_NO if a match was found and we can stop looking
6836  */
6837 static int
6838 check_validation_request_pending (void *cls,
6839                                   const struct GNUNET_PeerIdentity *pid,
6840                                   void *value)
6841 {
6842   struct Queue *q = cls;
6843   struct ValidationState *vs = value;
6844
6845   (void) pid;
6846   if ( (GNUNET_YES == vs->awaiting_queue) &&
6847        (0 == strcmp (vs->address,
6848                      q->address)) )
6849   {
6850     vs->awaiting_queue = GNUNET_NO;
6851     validation_transmit_on_queue (q,
6852                                   vs);
6853     return GNUNET_NO;
6854   }
6855   return GNUNET_OK;
6856 }
6857
6858
6859 /**
6860  * New queue became available.  Process the request.
6861  *
6862  * @param cls the client
6863  * @param aqm the send message that was sent
6864  */
6865 static void
6866 handle_add_queue_message (void *cls,
6867                           const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
6868 {
6869   struct TransportClient *tc = cls;
6870   struct Queue *queue;
6871   struct Neighbour *neighbour;
6872   const char *addr;
6873   uint16_t addr_len;
6874
6875   if (ntohl (aqm->mtu) <= sizeof (struct TransportFragmentBox))
6876   {
6877     /* MTU so small as to be useless for transmissions,
6878        required for #fragment_message()! */
6879     GNUNET_break_op (0);
6880     GNUNET_SERVICE_client_drop (tc->client);
6881     return;
6882   }
6883   neighbour = lookup_neighbour (&aqm->receiver);
6884   if (NULL == neighbour)
6885   {
6886     neighbour = GNUNET_new (struct Neighbour);
6887     neighbour->earliest_timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
6888     neighbour->pid = aqm->receiver;
6889     GNUNET_assert (GNUNET_OK ==
6890                    GNUNET_CONTAINER_multipeermap_put (neighbours,
6891                                                       &neighbour->pid,
6892                                                       neighbour,
6893                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
6894   }
6895   addr_len = ntohs (aqm->header.size) - sizeof (*aqm);
6896   addr = (const char *) &aqm[1];
6897
6898   queue = GNUNET_malloc (sizeof (struct Queue) + addr_len);
6899   queue->tc = tc;
6900   queue->address = (const char *) &queue[1];
6901   queue->rtt = GNUNET_TIME_UNIT_FOREVER_REL;
6902   queue->qid = aqm->qid;
6903   queue->mtu = ntohl (aqm->mtu);
6904   queue->nt = (enum GNUNET_NetworkType) ntohl (aqm->nt);
6905   queue->cs = (enum GNUNET_TRANSPORT_ConnectionStatus) ntohl (aqm->cs);
6906   queue->neighbour = neighbour;
6907   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_in,
6908                                   &tracker_update_in_cb,
6909                                   queue,
6910                                   GNUNET_BANDWIDTH_ZERO,
6911                                   GNUNET_CONSTANTS_MAX_BANDWIDTH_CARRY_S,
6912                                   &tracker_excess_in_cb,
6913                                   queue);
6914   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_out,
6915                                   &tracker_update_out_cb,
6916                                   queue,
6917                                   GNUNET_BANDWIDTH_ZERO,
6918                                   GNUNET_CONSTANTS_MAX_BANDWIDTH_CARRY_S,
6919                                   &tracker_excess_out_cb,
6920                                   queue);
6921   memcpy (&queue[1],
6922           addr,
6923           addr_len);
6924   /* notify monitors about new queue */
6925   {
6926     struct MonitorEvent me = {
6927       .rtt = queue->rtt,
6928       .cs = queue->cs
6929     };
6930
6931     notify_monitors (&neighbour->pid,
6932                      queue->address,
6933                      queue->nt,
6934                      &me);
6935   }
6936   GNUNET_CONTAINER_MDLL_insert (neighbour,
6937                                 neighbour->queue_head,
6938                                 neighbour->queue_tail,
6939                                 queue);
6940   GNUNET_CONTAINER_MDLL_insert (client,
6941                                 tc->details.communicator.queue_head,
6942                                 tc->details.communicator.queue_tail,
6943                                 queue);
6944   /* check if valdiations are waiting for the queue */
6945   (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
6946                                                      &aqm->receiver,
6947                                                      &check_validation_request_pending,
6948                                                      queue);
6949   /* might be our first queue, try launching DV learning */
6950   if (NULL == dvlearn_task)
6951     dvlearn_task = GNUNET_SCHEDULER_add_now (&start_dv_learn,
6952                                              NULL);
6953   GNUNET_SERVICE_client_continue (tc->client);
6954 }
6955
6956
6957 /**
6958  * Communicator tells us that our request to create a queue "worked", that
6959  * is setting up the queue is now in process.
6960  *
6961  * @param cls the `struct TransportClient`
6962  * @param cqr confirmation message
6963  */
6964 static void
6965 handle_queue_create_ok (void *cls,
6966                         const struct GNUNET_TRANSPORT_CreateQueueResponse *cqr)
6967 {
6968   struct TransportClient *tc = cls;
6969
6970   if (CT_COMMUNICATOR != tc->type)
6971   {
6972     GNUNET_break (0);
6973     GNUNET_SERVICE_client_drop (tc->client);
6974     return;
6975   }
6976   GNUNET_STATISTICS_update (GST_stats,
6977                             "# Suggestions succeeded at communicator",
6978                             1,
6979                             GNUNET_NO);
6980   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
6981               "Request #%u for communicator to create queue succeeded\n",
6982               (unsigned int) ntohs (cqr->request_id));
6983   GNUNET_SERVICE_client_continue (tc->client);
6984 }
6985
6986
6987 /**
6988  * Communicator tells us that our request to create a queue failed. This usually
6989  * indicates that the provided address is simply invalid or that the communicator's
6990  * resources are exhausted.
6991  *
6992  * @param cls the `struct TransportClient`
6993  * @param cqr failure message
6994  */
6995 static void
6996 handle_queue_create_fail (void *cls,
6997                           const struct GNUNET_TRANSPORT_CreateQueueResponse *cqr)
6998 {
6999   struct TransportClient *tc = cls;
7000
7001   if (CT_COMMUNICATOR != tc->type)
7002   {
7003     GNUNET_break (0);
7004     GNUNET_SERVICE_client_drop (tc->client);
7005     return;
7006   }
7007   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
7008               "Request #%u for communicator to create queue failed\n",
7009               (unsigned int) ntohs (cqr->request_id));
7010   GNUNET_STATISTICS_update (GST_stats,
7011                             "# Suggestions failed in queue creation at communicator",
7012                             1,
7013                             GNUNET_NO);
7014   GNUNET_SERVICE_client_continue (tc->client);
7015 }
7016
7017
7018 /**
7019  * We have received a `struct ExpressPreferenceMessage` from an application client.
7020  *
7021  * @param cls handle to the client
7022  * @param msg the start message
7023  */
7024 static void
7025 handle_suggest_cancel (void *cls,
7026                        const struct ExpressPreferenceMessage *msg)
7027 {
7028   struct TransportClient *tc = cls;
7029   struct PeerRequest *pr;
7030
7031   if (CT_APPLICATION != tc->type)
7032   {
7033     GNUNET_break (0);
7034     GNUNET_SERVICE_client_drop (tc->client);
7035     return;
7036   }
7037   pr = GNUNET_CONTAINER_multipeermap_get (tc->details.application.requests,
7038                                           &msg->peer);
7039   if (NULL == pr)
7040   {
7041     GNUNET_break (0);
7042     GNUNET_SERVICE_client_drop (tc->client);
7043     return;
7044   }
7045   (void) stop_peer_request (tc,
7046                             &pr->pid,
7047                             pr);
7048   GNUNET_SERVICE_client_continue (tc->client);
7049 }
7050
7051
7052 /**
7053  * Check #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_CONSIDER_VERIFY
7054  * messages. We do nothing here, real verification is done later.
7055  *
7056  * @param cls a `struct TransportClient *`
7057  * @param msg message to verify
7058  * @return #GNUNET_OK
7059  */
7060 static int
7061 check_address_consider_verify (void *cls,
7062                                const struct GNUNET_TRANSPORT_AddressToVerify *hdr)
7063 {
7064   (void) cls;
7065   (void) hdr;
7066   return GNUNET_OK;
7067 }
7068
7069
7070 /**
7071  * Closure for #check_known_address.
7072  */
7073 struct CheckKnownAddressContext
7074 {
7075   /**
7076    * Set to the address we are looking for.
7077    */
7078   const char *address;
7079
7080   /**
7081    * Set to a matching validation state, if one was found.
7082    */
7083   struct ValidationState *vs;
7084 };
7085
7086
7087 /**
7088  * Test if the validation state in @a value matches the
7089  * address from @a cls.
7090  *
7091  * @param cls a `struct CheckKnownAddressContext`
7092  * @param pid unused (must match though)
7093  * @param value a `struct ValidationState`
7094  * @return #GNUNET_OK if not matching, #GNUNET_NO if match found
7095  */
7096 static int
7097 check_known_address (void *cls,
7098                      const struct GNUNET_PeerIdentity *pid,
7099                      void *value)
7100 {
7101   struct CheckKnownAddressContext *ckac = cls;
7102   struct ValidationState *vs = value;
7103
7104   (void) pid;
7105   if (0 != strcmp (vs->address,
7106                    ckac->address))
7107     return GNUNET_OK;
7108   ckac->vs = vs;
7109   return GNUNET_NO;
7110 }
7111
7112
7113 /**
7114  * Start address validation.
7115  *
7116  * @param pid peer the @a address is for
7117  * @param address an address to reach @a pid (presumably)
7118  * @param expiration when did @a pid claim @a address will become invalid
7119  */
7120 static void
7121 start_address_validation (const struct GNUNET_PeerIdentity *pid,
7122                           const char *address,
7123                           struct GNUNET_TIME_Absolute expiration)
7124 {
7125   struct GNUNET_TIME_Absolute now;
7126   struct ValidationState *vs;
7127   struct CheckKnownAddressContext ckac = {
7128     .address = address,
7129     .vs = NULL
7130   };
7131
7132   if (0 == GNUNET_TIME_absolute_get_remaining (expiration).rel_value_us)
7133     return; /* expired */
7134   (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
7135                                                      pid,
7136                                                      &check_known_address,
7137                                                      &ckac);
7138   if (NULL != (vs = ckac.vs))
7139   {
7140     /* if 'vs' is not currently valid, we need to speed up retrying the validation */
7141     if (vs->validated_until.abs_value_us < vs->next_challenge.abs_value_us)
7142     {
7143       /* reduce backoff as we got a fresh advertisement */
7144       vs->challenge_backoff = GNUNET_TIME_relative_min (FAST_VALIDATION_CHALLENGE_FREQ,
7145                                                         GNUNET_TIME_relative_divide (vs->challenge_backoff,
7146                                                                                      2));
7147       update_next_challenge_time (vs,
7148                                   GNUNET_TIME_relative_to_absolute (vs->challenge_backoff));
7149     }
7150     return;
7151   }
7152   now = GNUNET_TIME_absolute_get();
7153   vs = GNUNET_new (struct ValidationState);
7154   vs->pid = *pid;
7155   vs->valid_until = expiration;
7156   vs->first_challenge_use = now;
7157   vs->validation_rtt = GNUNET_TIME_UNIT_FOREVER_REL;
7158   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
7159                               &vs->challenge,
7160                               sizeof (vs->challenge));
7161   vs->address = GNUNET_strdup (address);
7162   GNUNET_assert (GNUNET_YES ==
7163                  GNUNET_CONTAINER_multipeermap_put (validation_map,
7164                                                     &vs->pid,
7165                                                     vs,
7166                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
7167   update_next_challenge_time (vs,
7168                               now);
7169 }
7170
7171
7172 /**
7173  * Function called by PEERSTORE for each matching record.
7174  *
7175  * @param cls closure
7176  * @param record peerstore record information
7177  * @param emsg error message, or NULL if no errors
7178  */
7179 static void
7180 handle_hello (void *cls,
7181               const struct GNUNET_PEERSTORE_Record *record,
7182               const char *emsg)
7183 {
7184   struct PeerRequest *pr = cls;
7185   const char *val;
7186
7187   if (NULL != emsg)
7188   {
7189     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
7190                 "Got failure from PEERSTORE: %s\n",
7191                 emsg);
7192     return;
7193   }
7194   val = record->value;
7195   if ( (0 == record->value_size) ||
7196        ('\0' != val[record->value_size - 1]) )
7197   {
7198     GNUNET_break (0);
7199     return;
7200   }
7201   start_address_validation (&pr->pid,
7202                             (const char *) record->value,
7203                             record->expiry);
7204 }
7205
7206
7207 /**
7208  * We have received a `struct ExpressPreferenceMessage` from an application client.
7209  *
7210  * @param cls handle to the client
7211  * @param msg the start message
7212  */
7213 static void
7214 handle_suggest (void *cls,
7215                 const struct ExpressPreferenceMessage *msg)
7216 {
7217   struct TransportClient *tc = cls;
7218   struct PeerRequest *pr;
7219
7220   if (CT_NONE == tc->type)
7221   {
7222     tc->type = CT_APPLICATION;
7223     tc->details.application.requests
7224       = GNUNET_CONTAINER_multipeermap_create (16,
7225                                               GNUNET_YES);
7226   }
7227   if (CT_APPLICATION != tc->type)
7228   {
7229     GNUNET_break (0);
7230     GNUNET_SERVICE_client_drop (tc->client);
7231     return;
7232   }
7233   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
7234               "Client suggested we talk to %s with preference %d at rate %u\n",
7235               GNUNET_i2s (&msg->peer),
7236               (int) ntohl (msg->pk),
7237               (int) ntohl (msg->bw.value__));
7238   pr = GNUNET_new (struct PeerRequest);
7239   pr->tc = tc;
7240   pr->pid = msg->peer;
7241   pr->bw = msg->bw;
7242   pr->pk = (enum GNUNET_MQ_PreferenceKind) ntohl (msg->pk);
7243   if (GNUNET_YES !=
7244       GNUNET_CONTAINER_multipeermap_put (tc->details.application.requests,
7245                                          &pr->pid,
7246                                          pr,
7247                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
7248   {
7249     GNUNET_break (0);
7250     GNUNET_free (pr);
7251     GNUNET_SERVICE_client_drop (tc->client);
7252     return;
7253   }
7254   pr->wc = GNUNET_PEERSTORE_watch (peerstore,
7255                                    "transport",
7256                                    &pr->pid,
7257                                    GNUNET_PEERSTORE_TRANSPORT_URLADDRESS_KEY,
7258                                    &handle_hello,
7259                                    pr);
7260   GNUNET_SERVICE_client_continue (tc->client);
7261 }
7262
7263
7264 /**
7265  * Given another peers address, consider checking it for validity
7266  * and then adding it to the Peerstore.
7267  *
7268  * @param cls a `struct TransportClient`
7269  * @param hdr message containing the raw address data and
7270  *        signature in the body, see #GNUNET_HELLO_extract_address()
7271  */
7272 static void
7273 handle_address_consider_verify (void *cls,
7274                                 const struct GNUNET_TRANSPORT_AddressToVerify *hdr)
7275 {
7276   struct TransportClient *tc = cls;
7277   char *address;
7278   enum GNUNET_NetworkType nt;
7279   struct GNUNET_TIME_Absolute expiration;
7280
7281   (void) cls;
7282   // OPTIMIZE-FIXME: checking that we know this address already should
7283   //        be done BEFORE checking the signature => HELLO API change!
7284   // OPTIMIZE-FIXME: pre-check: rate-limit signature verification / validation?!
7285   address = GNUNET_HELLO_extract_address (&hdr[1],
7286                                           ntohs (hdr->header.size) - sizeof (*hdr),
7287                                           &hdr->peer,
7288                                           &nt,
7289                                           &expiration);
7290   if (NULL == address)
7291   {
7292     GNUNET_break_op (0);
7293     return;
7294   }
7295   start_address_validation (&hdr->peer,
7296                             address,
7297                             expiration);
7298   GNUNET_free (address);
7299   GNUNET_SERVICE_client_continue (tc->client);
7300 }
7301
7302
7303 /**
7304  * Check #GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_HELLO_VALIDATION
7305  * messages.
7306  *
7307  * @param cls a `struct TransportClient *`
7308  * @param m message to verify
7309  * @return #GNUNET_OK on success
7310  */
7311 static int
7312 check_request_hello_validation (void *cls,
7313                                 const struct RequestHelloValidationMessage *m)
7314 {
7315   (void) cls;
7316   GNUNET_MQ_check_zero_termination (m);
7317   return GNUNET_OK;
7318 }
7319
7320
7321 /**
7322  * A client encountered an address of another peer. Consider validating it,
7323  * and if validation succeeds, persist it to PEERSTORE.
7324  *
7325  * @param cls a `struct TransportClient *`
7326  * @param m message to verify
7327  */
7328 static void
7329 handle_request_hello_validation (void *cls,
7330                                  const struct RequestHelloValidationMessage *m)
7331 {
7332   struct TransportClient *tc = cls;
7333
7334   start_address_validation (&m->peer,
7335                             (const char *) &m[1],
7336                             GNUNET_TIME_absolute_ntoh (m->expiration));
7337   GNUNET_SERVICE_client_continue (tc->client);
7338 }
7339
7340
7341 /**
7342  * Free neighbour entry.
7343  *
7344  * @param cls NULL
7345  * @param pid unused
7346  * @param value a `struct Neighbour`
7347  * @return #GNUNET_OK (always)
7348  */
7349 static int
7350 free_neighbour_cb (void *cls,
7351                    const struct GNUNET_PeerIdentity *pid,
7352                    void *value)
7353 {
7354   struct Neighbour *neighbour = value;
7355
7356   (void) cls;
7357   (void) pid;
7358   GNUNET_break (0); // should this ever happen?
7359   free_neighbour (neighbour);
7360
7361   return GNUNET_OK;
7362 }
7363
7364
7365 /**
7366  * Free DV route entry.
7367  *
7368  * @param cls NULL
7369  * @param pid unused
7370  * @param value a `struct DistanceVector`
7371  * @return #GNUNET_OK (always)
7372  */
7373 static int
7374 free_dv_routes_cb (void *cls,
7375                    const struct GNUNET_PeerIdentity *pid,
7376                    void *value)
7377 {
7378   struct DistanceVector *dv = value;
7379
7380   (void) cls;
7381   (void) pid;
7382   free_dv_route (dv);
7383
7384   return GNUNET_OK;
7385 }
7386
7387
7388 /**
7389  * Free ephemeral entry.
7390  *
7391  * @param cls NULL
7392  * @param pid unused
7393  * @param value a `struct EphemeralCacheEntry`
7394  * @return #GNUNET_OK (always)
7395  */
7396 static int
7397 free_ephemeral_cb (void *cls,
7398                    const struct GNUNET_PeerIdentity *pid,
7399                    void *value)
7400 {
7401   struct EphemeralCacheEntry *ece = value;
7402
7403   (void) cls;
7404   (void) pid;
7405   free_ephemeral (ece);
7406   return GNUNET_OK;
7407 }
7408
7409
7410 /**
7411  * Free validation state.
7412  *
7413  * @param cls NULL
7414  * @param pid unused
7415  * @param value a `struct ValidationState`
7416  * @return #GNUNET_OK (always)
7417  */
7418 static int
7419 free_validation_state_cb (void *cls,
7420                           const struct GNUNET_PeerIdentity *pid,
7421                           void *value)
7422 {
7423   struct ValidationState *vs = value;
7424
7425   (void) cls;
7426   (void) pid;
7427   free_validation_state (vs);
7428   return GNUNET_OK;
7429 }
7430
7431
7432 /**
7433  * Function called when the service shuts down.  Unloads our plugins
7434  * and cancels pending validations.
7435  *
7436  * @param cls closure, unused
7437  */
7438 static void
7439 do_shutdown (void *cls)
7440 {
7441   struct LearnLaunchEntry *lle;
7442   (void) cls;
7443
7444   if (NULL != ephemeral_task)
7445   {
7446     GNUNET_SCHEDULER_cancel (ephemeral_task);
7447     ephemeral_task = NULL;
7448   }
7449   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
7450                                          &free_neighbour_cb,
7451                                          NULL);
7452   if (NULL != peerstore)
7453   {
7454     GNUNET_PEERSTORE_disconnect (peerstore,
7455                                  GNUNET_NO);
7456     peerstore = NULL;
7457   }
7458   if (NULL != GST_stats)
7459   {
7460     GNUNET_STATISTICS_destroy (GST_stats,
7461                                GNUNET_NO);
7462     GST_stats = NULL;
7463   }
7464   if (NULL != GST_my_private_key)
7465   {
7466     GNUNET_free (GST_my_private_key);
7467     GST_my_private_key = NULL;
7468   }
7469   GNUNET_CONTAINER_multipeermap_destroy (neighbours);
7470   neighbours = NULL;
7471   GNUNET_CONTAINER_multipeermap_iterate (validation_map,
7472                                          &free_validation_state_cb,
7473                                          NULL);
7474   GNUNET_CONTAINER_multipeermap_destroy (validation_map);
7475   validation_map = NULL;
7476   while (NULL != (lle = lle_head))
7477   {
7478     GNUNET_CONTAINER_DLL_remove (lle_head,
7479                                  lle_tail,
7480                                  lle);
7481     GNUNET_free (lle);
7482   }
7483   GNUNET_CONTAINER_multishortmap_destroy (dvlearn_map);
7484   dvlearn_map = NULL;
7485   GNUNET_CONTAINER_heap_destroy (validation_heap);
7486   validation_heap = NULL;
7487   GNUNET_CONTAINER_multipeermap_iterate (dv_routes,
7488                                          &free_dv_routes_cb,
7489                                          NULL);
7490   GNUNET_CONTAINER_multipeermap_destroy (dv_routes);
7491   dv_routes = NULL;
7492   GNUNET_CONTAINER_multipeermap_iterate (ephemeral_map,
7493                                          &free_ephemeral_cb,
7494                                          NULL);
7495   GNUNET_CONTAINER_multipeermap_destroy (ephemeral_map);
7496   ephemeral_map = NULL;
7497   GNUNET_CONTAINER_heap_destroy (ephemeral_heap);
7498   ephemeral_heap = NULL;
7499 }
7500
7501
7502 /**
7503  * Initiate transport service.
7504  *
7505  * @param cls closure
7506  * @param c configuration to use
7507  * @param service the initialized service
7508  */
7509 static void
7510 run (void *cls,
7511      const struct GNUNET_CONFIGURATION_Handle *c,
7512      struct GNUNET_SERVICE_Handle *service)
7513 {
7514   (void) cls;
7515   (void) service;
7516   /* setup globals */
7517   GST_cfg = c;
7518   neighbours = GNUNET_CONTAINER_multipeermap_create (1024,
7519                                                      GNUNET_YES);
7520   dv_routes = GNUNET_CONTAINER_multipeermap_create (1024,
7521                                                     GNUNET_YES);
7522   ephemeral_map = GNUNET_CONTAINER_multipeermap_create (32,
7523                                                         GNUNET_YES);
7524   ephemeral_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
7525   dvlearn_map = GNUNET_CONTAINER_multishortmap_create (2 * MAX_DV_LEARN_PENDING,
7526                                                        GNUNET_YES);
7527   validation_map = GNUNET_CONTAINER_multipeermap_create (1024,
7528                                                          GNUNET_YES);
7529   validation_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
7530   GST_my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (GST_cfg);
7531   if (NULL == GST_my_private_key)
7532   {
7533     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
7534                 _("Transport service is lacking key configuration settings. Exiting.\n"));
7535     GNUNET_SCHEDULER_shutdown ();
7536     return;
7537   }
7538   GNUNET_CRYPTO_eddsa_key_get_public (GST_my_private_key,
7539                                       &GST_my_identity.public_key);
7540   GNUNET_log(GNUNET_ERROR_TYPE_INFO,
7541              "My identity is `%s'\n",
7542              GNUNET_i2s_full (&GST_my_identity));
7543   GST_stats = GNUNET_STATISTICS_create ("transport",
7544                                         GST_cfg);
7545   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
7546                                  NULL);
7547   peerstore = GNUNET_PEERSTORE_connect (GST_cfg);
7548   if (NULL == peerstore)
7549   {
7550     GNUNET_break (0);
7551     GNUNET_SCHEDULER_shutdown ();
7552     return;
7553   }
7554 }
7555
7556
7557 /**
7558  * Define "main" method using service macro.
7559  */
7560 GNUNET_SERVICE_MAIN
7561 ("transport",
7562  GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN,
7563  &run,
7564  &client_connect_cb,
7565  &client_disconnect_cb,
7566  NULL,
7567  /* communication with applications */
7568  GNUNET_MQ_hd_fixed_size (suggest,
7569                           GNUNET_MESSAGE_TYPE_TRANSPORT_SUGGEST,
7570                           struct ExpressPreferenceMessage,
7571                           NULL),
7572  GNUNET_MQ_hd_fixed_size (suggest_cancel,
7573                           GNUNET_MESSAGE_TYPE_TRANSPORT_SUGGEST_CANCEL,
7574                           struct ExpressPreferenceMessage,
7575                           NULL),
7576  GNUNET_MQ_hd_var_size (request_hello_validation,
7577                         GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_HELLO_VALIDATION,
7578                         struct RequestHelloValidationMessage,
7579                         NULL),
7580  /* communication with core */
7581  GNUNET_MQ_hd_fixed_size (client_start,
7582                           GNUNET_MESSAGE_TYPE_TRANSPORT_START,
7583                           struct StartMessage,
7584                           NULL),
7585  GNUNET_MQ_hd_var_size (client_send,
7586                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND,
7587                         struct OutboundMessage,
7588                         NULL),
7589  /* communication with communicators */
7590  GNUNET_MQ_hd_var_size (communicator_available,
7591                         GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
7592                         struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
7593                         NULL),
7594  GNUNET_MQ_hd_var_size (communicator_backchannel,
7595                         GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL,
7596                         struct GNUNET_TRANSPORT_CommunicatorBackchannel,
7597                         NULL),
7598  GNUNET_MQ_hd_var_size (add_address,
7599                         GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
7600                         struct GNUNET_TRANSPORT_AddAddressMessage,
7601                         NULL),
7602  GNUNET_MQ_hd_fixed_size (del_address,
7603                           GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
7604                           struct GNUNET_TRANSPORT_DelAddressMessage,
7605                           NULL),
7606  GNUNET_MQ_hd_var_size (incoming_msg,
7607                         GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
7608                         struct GNUNET_TRANSPORT_IncomingMessage,
7609                         NULL),
7610  GNUNET_MQ_hd_fixed_size (queue_create_ok,
7611                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK,
7612                           struct GNUNET_TRANSPORT_CreateQueueResponse,
7613                           NULL),
7614  GNUNET_MQ_hd_fixed_size (queue_create_fail,
7615                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL,
7616                           struct GNUNET_TRANSPORT_CreateQueueResponse,
7617                           NULL),
7618  GNUNET_MQ_hd_var_size (add_queue_message,
7619                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
7620                         struct GNUNET_TRANSPORT_AddQueueMessage,
7621                         NULL),
7622  GNUNET_MQ_hd_var_size (address_consider_verify,
7623                         GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_CONSIDER_VERIFY,
7624                         struct GNUNET_TRANSPORT_AddressToVerify,
7625                         NULL),
7626  GNUNET_MQ_hd_fixed_size (del_queue_message,
7627                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
7628                           struct GNUNET_TRANSPORT_DelQueueMessage,
7629                           NULL),
7630  GNUNET_MQ_hd_fixed_size (send_message_ack,
7631                           GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
7632                           struct GNUNET_TRANSPORT_SendMessageToAck,
7633                           NULL),
7634  /* communication with monitors */
7635  GNUNET_MQ_hd_fixed_size (monitor_start,
7636                           GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_START,
7637                           struct GNUNET_TRANSPORT_MonitorStart,
7638                           NULL),
7639  GNUNET_MQ_handler_end ());
7640
7641
7642 /* end of file gnunet-service-transport.c */