fix for mantis 2355
[oweals/gnunet.git] / src / transport / gnunet-service-transport_neighbours.c
1 /*
2      This file is part of GNUnet.
3      (C) 2010,2011,2012 Christian Grothoff (and other contributing authors)
4
5      GNUnet is free software; you can redistribute it and/or modify
6      it under the terms of the GNU General Public License as published
7      by the Free Software Foundation; either version 3, or (at your
8      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      General Public License for more details.
14
15      You should have received a copy of the GNU General Public License
16      along with GNUnet; see the file COPYING.  If not, write to the
17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18      Boston, MA 02111-1307, USA.
19 */
20
21 /**
22  * @file transport/gnunet-service-transport_neighbours.c
23  * @brief neighbour management
24  * @author Christian Grothoff
25  *
26  * TODO:
27  * - "address_change_cb" is NEVER invoked; when should we call this one exactly?
28  * - TEST, TEST, TEST...
29  */
30 #include "platform.h"
31 #include "gnunet_ats_service.h"
32 #include "gnunet-service-transport_neighbours.h"
33 #include "gnunet-service-transport_plugins.h"
34 #include "gnunet-service-transport_validation.h"
35 #include "gnunet-service-transport_clients.h"
36 #include "gnunet-service-transport.h"
37 #include "gnunet_peerinfo_service.h"
38 #include "gnunet-service-transport_blacklist.h"
39 #include "gnunet_constants.h"
40 #include "transport.h"
41
42
43 /**
44  * Size of the neighbour hash map.
45  */
46 #define NEIGHBOUR_TABLE_SIZE 256
47
48 /**
49  * Time we give plugin to transmit DISCONNECT message before the
50  * neighbour entry self-destructs.
51  */
52 #define DISCONNECT_SENT_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 100)
53
54 /**
55  * How often must a peer violate bandwidth quotas before we start
56  * to simply drop its messages?
57  */
58 #define QUOTA_VIOLATION_DROP_THRESHOLD 10
59
60 /**
61  * How often do we send KEEPALIVE messages to each of our neighbours and measure
62  * the latency with this neighbour?
63  * (idle timeout is 5 minutes or 300 seconds, so with 30s interval we
64  * send 10 keepalives in each interval, so 10 messages would need to be
65  * lost in a row for a disconnect).
66  */
67 #define KEEPALIVE_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
68
69 /**
70  * How long are we willing to wait for a response from ATS before timing out?
71  */
72 #define ATS_RESPONSE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 500)
73
74 /**
75  * How long are we willing to wait for an ACK from the other peer before
76  * giving up on our connect operation?
77  */
78 #define SETUP_CONNECTION_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 15)
79
80 /**
81  * How long are we willing to wait for a successful reconnect if 
82  * an existing connection went down?  Much shorter than the
83  * usual SETUP_CONNECTION_TIMEOUT as we do not inform the
84  * higher layers about the disconnect during this period.
85  */
86 #define FAST_RECONNECT_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
87
88 /**
89  * How long are we willing to wait for a response from the blacklist
90  * subsystem before timing out?
91  */
92 #define BLACKLIST_RESPONSE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 500)
93
94
95 GNUNET_NETWORK_STRUCT_BEGIN
96
97 /**
98  * Message a peer sends to another to indicate that it intends to
99  * setup a connection/session for data exchange.  A 'SESSION_CONNECT'
100  * should be answered with a 'SESSION_CONNECT_ACK' with the same body
101  * to confirm.  A 'SESSION_CONNECT_ACK' should then be followed with
102  * a 'SESSION_ACK'.  Once the 'SESSION_ACK' is received, both peers 
103  * should be connected.
104  */
105 struct SessionConnectMessage
106 {
107   /**
108    * Header of type 'GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT'
109    * or 'GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT_ACK'
110    */
111   struct GNUNET_MessageHeader header;
112
113   /**
114    * Always zero.
115    */
116   uint32_t reserved GNUNET_PACKED;
117
118   /**
119    * Absolute time at the sender.  Only the most recent connect
120    * message implies which session is preferred by the sender.
121    */
122   struct GNUNET_TIME_AbsoluteNBO timestamp;
123
124 };
125
126
127 /**
128  * Message we send to the other peer to notify him that we intentionally
129  * are disconnecting (to reduce timeouts).  This is just a friendly 
130  * notification, peers must not rely on always receiving disconnect
131  * messages.
132  */
133 struct SessionDisconnectMessage
134 {
135   /**
136    * Header of type 'GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT'
137    */
138   struct GNUNET_MessageHeader header;
139
140   /**
141    * Always zero.
142    */
143   uint32_t reserved GNUNET_PACKED;
144
145   /**
146    * Purpose of the signature.  Extends over the timestamp.
147    * Purpose should be GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DISCONNECT.
148    */
149   struct GNUNET_CRYPTO_RsaSignaturePurpose purpose;
150
151   /**
152    * Absolute time at the sender.  Only the most recent connect
153    * message implies which session is preferred by the sender.
154    */
155   struct GNUNET_TIME_AbsoluteNBO timestamp;
156
157   /**
158    * Public key of the sender.
159    */
160   struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded public_key;
161
162   /**
163    * Signature of the peer that sends us the disconnect.  Only
164    * valid if the timestamp is AFTER the timestamp from the
165    * corresponding 'CONNECT' message.
166    */
167   struct GNUNET_CRYPTO_RsaSignature signature;
168
169 };
170
171 GNUNET_NETWORK_STRUCT_END
172
173
174 /**
175  * For each neighbour we keep a list of messages
176  * that we still want to transmit to the neighbour.
177  */
178 struct MessageQueue
179 {
180
181   /**
182    * This is a doubly linked list.
183    */
184   struct MessageQueue *next;
185
186   /**
187    * This is a doubly linked list.
188    */
189   struct MessageQueue *prev;
190
191   /**
192    * Function to call once we're done.
193    */
194   GST_NeighbourSendContinuation cont;
195
196   /**
197    * Closure for 'cont'
198    */
199   void *cont_cls;
200
201   /**
202    * The message(s) we want to transmit, GNUNET_MessageHeader(s)
203    * stuck together in memory.  Allocated at the end of this struct.
204    */
205   const char *message_buf;
206
207   /**
208    * Size of the message buf
209    */
210   size_t message_buf_size;
211
212   /**
213    * At what time should we fail?
214    */
215   struct GNUNET_TIME_Absolute timeout;
216
217 };
218
219
220 /**
221  * Possible state of a neighbour.  Initially, we are S_NOT_CONNECTED.
222  *
223  * Then, there are two main paths. If we receive a CONNECT message, we
224  * first run a check against the blacklist and ask ATS for a
225  * suggestion.  (S_CONNECT_RECV_ATS).  If the blacklist comes back
226  * positive, we give the address to ATS.  If ATS makes a suggestion,
227  * we ALSO give that suggestion to the blacklist
228  * (S_CONNECT_RECV_BLACKLIST).  Once the blacklist approves the
229  * address we got from ATS, we send our CONNECT_ACK and go to
230  * S_CONNECT_RECV_ACK.  If we receive a SESSION_ACK, we go to
231  * S_CONNECTED (and notify everyone about the new connection).  If the
232  * operation times out, we go to S_DISCONNECT.
233  *
234  * The other case is where we transmit a CONNECT message first.  We
235  * start with S_INIT_ATS.  If we get an address, we enter
236  * S_INIT_BLACKLIST and check the blacklist.  If the blacklist is OK
237  * with the connection, we actually send the CONNECT message and go to
238  * state S_CONNECT_SENT.  Once we receive a CONNECT_ACK, we go to
239  * S_CONNECTED (and notify everyone about the new connection and send
240  * back a SESSION_ACK).  If the operation times out, we go to
241  * S_DISCONNECT.
242  *
243  * If the session is in trouble (i.e. transport-level disconnect or
244  * timeout), we go to S_RECONNECT_ATS where we ask ATS for a new
245  * address (we don't notify anyone about the disconnect yet).  Once we
246  * have a new address, we go to S_RECONNECT_BLACKLIST to check the new
247  * address against the blacklist.  If the blacklist approves, we enter
248  * S_RECONNECT_SENT and send a CONNECT message.  If we receive a
249  * CONNECT_ACK, we go to S_CONNECTED and nobody noticed that we had
250  * trouble; we also send a SESSION_ACK at this time just in case.  If
251  * the operation times out, we go to S_DISCONNECT (and notify everyone
252  * about the lost connection).
253  *
254  * If ATS decides to switch addresses while we have a normal
255  * connection, we go to S_CONNECTED_SWITCHING_BLACKLIST to check the
256  * new address against the blacklist.  If the blacklist approves, we
257  * go to S_CONNECTED_SWITCHING_CONNECT_SENT and send a
258  * SESSION_CONNECT.  If we get a SESSION_ACK back, we switch the
259  * primary connection to the suggested alternative from ATS, go back
260  * to S_CONNECTED and send a SESSION_ACK to the other peer just to be
261  * sure.  If the operation times out (or the blacklist disapproves),
262  * we go to S_CONNECTED (and notify ATS that the given alternative
263  * address is "invalid").
264  *
265  * Once a session is in S_DISCONNECT, it is cleaned up and then goes
266  * to (S_DISCONNECT_FINISHED).  If we receive an explicit disconnect
267  * request, we can go from any state to S_DISCONNECT, possibly after
268  * generating disconnect notifications.
269  *
270  * Note that it is quite possible that while we are in any of these
271  * states, we could receive a 'CONNECT' request from the other peer.
272  * We then enter a 'weird' state where we pursue our own primary state
273  * machine (as described above), but with the 'send_connect_ack' flag
274  * set to 1.  If our state machine allows us to send a 'CONNECT_ACK'
275  * (because we have an acceptable address), we send the 'CONNECT_ACK'
276  * and set the 'send_connect_ack' to 2.  If we then receive a
277  * 'SESSION_ACK', we go to 'S_CONNECTED' (and reset 'send_connect_ack'
278  * to 0).
279  * 
280  */ 
281 enum State
282 {
283   /**
284    * fresh peer or completely disconnected
285    */
286   S_NOT_CONNECTED,
287
288   /**
289    * Asked to initiate connection, trying to get address from ATS
290    */
291   S_INIT_ATS,
292
293   /**
294    * Asked to initiate connection, trying to get address approved
295    * by blacklist.
296    */
297   S_INIT_BLACKLIST,
298
299   /**
300    * Sent CONNECT message to other peer, waiting for CONNECT_ACK
301    */
302   S_CONNECT_SENT,
303
304   /**
305    * Received a CONNECT, asking ATS about address suggestions.
306    */
307   S_CONNECT_RECV_ATS,
308
309   /**
310    * Received CONNECT from other peer, got an address, checking with blacklist.
311    */
312   S_CONNECT_RECV_BLACKLIST,
313
314   /**
315    * CONNECT request from other peer was SESSION_ACK'ed, waiting for
316    * SESSION_ACK.
317    */
318   S_CONNECT_RECV_ACK,
319
320   /**
321    * Got our CONNECT_ACK/SESSION_ACK, connection is up.
322    */
323   S_CONNECTED,
324
325   /**
326    * Connection got into trouble, rest of the system still believes
327    * it to be up, but we're getting a new address from ATS.
328    */
329   S_RECONNECT_ATS,
330
331   /**
332    * Connection got into trouble, rest of the system still believes
333    * it to be up; we are checking the new address against the blacklist.
334    */
335   S_RECONNECT_BLACKLIST,
336
337   /**
338    * Sent CONNECT over new address (either by ATS telling us to switch
339    * addresses or from RECONNECT_ATS); if this fails, we need to tell
340    * the rest of the system about a disconnect.
341    */
342   S_RECONNECT_SENT,
343
344   /**
345    * We have some primary connection, but ATS suggested we switch
346    * to some alternative; we're now checking the alternative against
347    * the blacklist.
348    */
349   S_CONNECTED_SWITCHING_BLACKLIST,
350
351   /** 
352    * We have some primary connection, but ATS suggested we switch
353    * to some alternative; we now sent a CONNECT message for the
354    * alternative session to the other peer and waiting for a
355    * CONNECT_ACK to make this our primary connection.
356    */
357   S_CONNECTED_SWITCHING_CONNECT_SENT,
358
359   /**
360    * Disconnect in progress (we're sending the DISCONNECT message to the
361    * other peer; after that is finished, the state will be cleaned up).
362    */
363   S_DISCONNECT,
364
365   /**
366    * We're finished with the disconnect; clean up state now!
367    */
368   S_DISCONNECT_FINISHED
369 };
370
371
372 /**
373  * A possible address we could use to communicate with a neighbour.
374  */
375 struct NeighbourAddress
376 {
377
378   /**
379    * Active session for this address.
380    */
381   struct Session *session;
382
383   /**
384    * Network-level address information.
385    */
386   struct GNUNET_HELLO_Address *address;
387
388   /**
389    * Timestamp of the 'SESSION_CONNECT' message we sent to the other
390    * peer for this address.  Use to check that the ACK is in response
391    * to our most recent 'CONNECT'.
392    */
393   struct GNUNET_TIME_Absolute connect_timestamp;
394
395   /**
396    * Inbound bandwidth from ATS for this address.
397    */
398   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in;
399
400   /**
401    * Outbound bandwidth from ATS for this address.
402    */
403   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out;
404
405   /**
406    * Did we tell ATS that this is our 'active' address?
407    */
408   int ats_active;
409   
410 };
411
412
413 /**
414  * Entry in neighbours.
415  */
416 struct NeighbourMapEntry
417 {
418
419   /**
420    * Head of list of messages we would like to send to this peer;
421    * must contain at most one message per client.
422    */
423   struct MessageQueue *messages_head;
424
425   /**
426    * Tail of list of messages we would like to send to this peer; must
427    * contain at most one message per client.
428    */
429   struct MessageQueue *messages_tail;
430
431   /**
432    * Are we currently trying to send a message? If so, which one?
433    */
434   struct MessageQueue *is_active;
435
436   /**
437    * Primary address we currently use to communicate with the neighbour.
438    */
439   struct NeighbourAddress primary_address;
440
441   /**
442    * Alternative address currently under consideration for communicating
443    * with the neighbour.
444    */
445   struct NeighbourAddress alternative_address;
446
447   /**
448    * Identity of this neighbour.
449    */
450   struct GNUNET_PeerIdentity id;
451
452   /**
453    * Main task that drives this peer (timeouts, keepalives, etc.).
454    * Always runs the 'master_task'.
455    */
456   GNUNET_SCHEDULER_TaskIdentifier task;
457
458   /**
459    * At what time should we sent the next keep-alive message?
460    */
461   struct GNUNET_TIME_Absolute keep_alive_time;
462
463   /**
464    * At what time did we sent the last keep-alive message?  Used 
465    * to calculate round-trip time ("latency").
466    */
467   struct GNUNET_TIME_Absolute last_keep_alive_time;
468
469   /**
470    * Timestamp we should include in our next CONNECT_ACK message.
471    * (only valid if 'send_connect_ack' is GNUNET_YES).  Used to build
472    * our CONNECT_ACK message.
473    */
474   struct GNUNET_TIME_Absolute connect_ack_timestamp;
475
476   /**
477    * Time where we should cut the connection (timeout) if we don't
478    * make progress in the state machine (or get a KEEPALIVE_RESPONSE
479    * if we are in S_CONNECTED).
480    */
481   struct GNUNET_TIME_Absolute timeout;
482
483   /**
484    * Latest calculated latency value
485    */
486   struct GNUNET_TIME_Relative latency;
487
488   /**
489    * Tracker for inbound bandwidth.
490    */
491   struct GNUNET_BANDWIDTH_Tracker in_tracker;
492
493   /**
494    * How often has the other peer (recently) violated the inbound
495    * traffic limit?  Incremented by 10 per violation, decremented by 1
496    * per non-violation (for each time interval).
497    */
498   unsigned int quota_violation_count;
499
500   /**
501    * The current state of the peer.
502    */
503   enum State state;
504
505   /**
506    * Did we sent an KEEP_ALIVE message and are we expecting a response?
507    */
508   int expect_latency_response;
509
510   /**
511    * Flag to set if we still need to send a CONNECT_ACK message to the other peer
512    * (once we have an address to use and the peer has been allowed by our
513    * blacklist).  Set to 1 if we need to send a CONNECT_ACK.  Set to 2 if we
514    * did send a CONNECT_ACK and should go to 'S_CONNECTED' upon receiving
515    * a 'SESSION_ACK' (regardless of what our own state machine might say).
516    */
517   int send_connect_ack;
518
519 };
520
521
522 /**
523  * Context for blacklist checks and the 'handle_test_blacklist_cont'
524  * function.  Stores information about ongoing blacklist checks.
525  */
526 struct BlackListCheckContext
527 {
528   
529   /**
530    * We keep blacklist checks in a DLL.
531    */
532   struct BlackListCheckContext *next;
533
534   /**
535    * We keep blacklist checks in a DLL.
536    */
537   struct BlackListCheckContext *prev;
538
539   /**
540    * Address that is being checked.
541    */
542   struct NeighbourAddress na;
543   
544   /**
545    * ATS information about the address.
546    */
547   struct GNUNET_ATS_Information *ats;
548
549   /**
550    * Handle to the ongoing blacklist check.
551    */
552   struct GST_BlacklistCheck *bc;
553
554   /**
555    * Size of the 'ats' array.
556    */
557   uint32_t ats_count;
558
559 };
560
561
562 /**
563  * Hash map from peer identities to the respective 'struct NeighbourMapEntry'.
564  */
565 static struct GNUNET_CONTAINER_MultiHashMap *neighbours;
566
567 /**
568  * We keep blacklist checks in a DLL so that we can find
569  * the 'sessions' in their 'struct NeighbourAddress' if
570  * a session goes down.
571  */
572 static struct BlackListCheckContext *bc_head;
573
574 /**
575  * We keep blacklist checks in a DLL.
576  */
577 static struct BlackListCheckContext *bc_tail;
578
579 /**
580  * Closure for connect_notify_cb, disconnect_notify_cb and address_change_cb
581  */
582 static void *callback_cls;
583
584 /**
585  * Function to call when we connected to a neighbour.
586  */
587 static GNUNET_TRANSPORT_NotifyConnect connect_notify_cb;
588
589 /**
590  * Function to call when we disconnected from a neighbour.
591  */
592 static GNUNET_TRANSPORT_NotifyDisconnect disconnect_notify_cb;
593
594 /**
595  * Function to call when we changed an active address of a neighbour.
596  */
597 static GNUNET_TRANSPORT_PeerIterateCallback address_change_cb;
598
599 /**
600  * counter for connected neighbours
601  */
602 static unsigned int neighbours_connected;
603
604 /**
605  * Number of bytes we have currently queued for transmission.
606  */
607 static unsigned long long bytes_in_send_queue;
608
609
610 /**
611  * Lookup a neighbour entry in the neighbours hash map.
612  *
613  * @param pid identity of the peer to look up
614  * @return the entry, NULL if there is no existing record
615  */
616 static struct NeighbourMapEntry *
617 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
618 {
619   if (NULL == neighbours)
620     return NULL;
621   return GNUNET_CONTAINER_multihashmap_get (neighbours, &pid->hashPubKey);
622 }
623
624
625 /**
626  * Test if we're connected to the given peer.
627  *
628  * @param n neighbour entry of peer to test
629  * @return GNUNET_YES if we are connected, GNUNET_NO if not
630  */
631 static int
632 test_connected (struct NeighbourMapEntry *n)
633 {
634   if (NULL == n)
635     return GNUNET_NO;
636   switch (n->state)
637   {
638   case S_NOT_CONNECTED:
639   case S_INIT_ATS:
640   case S_INIT_BLACKLIST:
641   case S_CONNECT_SENT:
642   case S_CONNECT_RECV_ATS:
643   case S_CONNECT_RECV_BLACKLIST:
644   case S_CONNECT_RECV_ACK:
645     return GNUNET_NO;
646   case S_CONNECTED:
647   case S_RECONNECT_ATS:
648   case S_RECONNECT_BLACKLIST:
649   case S_RECONNECT_SENT:
650   case S_CONNECTED_SWITCHING_BLACKLIST:
651   case S_CONNECTED_SWITCHING_CONNECT_SENT:
652     return GNUNET_YES;
653   case S_DISCONNECT:
654   case S_DISCONNECT_FINISHED:
655     return GNUNET_NO;
656   default:
657     GNUNET_break (0);
658     break;
659   }
660   return GNUNET_SYSERR;
661 }
662
663
664 /**
665  * Send information about a new outbound quota to our clients.
666  *
667  * @param target affected peer
668  * @param quota new quota
669  */
670 static void
671 send_outbound_quota (const struct GNUNET_PeerIdentity *target,
672                      struct GNUNET_BANDWIDTH_Value32NBO quota)
673 {
674   struct QuotaSetMessage q_msg;
675
676   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
677               "Sending outbound quota of %u Bps for peer `%s' to all clients\n",
678               ntohl (quota.value__), GNUNET_i2s (target));
679   q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
680   q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
681   q_msg.quota = quota;
682   q_msg.peer = (*target);
683   GST_clients_broadcast (&q_msg.header, GNUNET_NO);
684 }
685
686
687 /**
688  * We don't need a given neighbour address any more.
689  * Release its resources and give appropriate notifications
690  * to ATS and other subsystems.
691  *
692  * @param na address we are done with; 'na' itself must NOT be 'free'd, only the contents!
693  */
694 static void
695 free_address (struct NeighbourAddress *na)
696 {
697   if (GNUNET_YES == na->ats_active)
698   {
699     GST_validation_set_address_use (na->address, na->session, GNUNET_NO, __LINE__);
700     GNUNET_ATS_address_in_use (GST_ats, na->address, na->session, GNUNET_NO);
701   }
702   na->ats_active = GNUNET_NO;
703   if (NULL != na->address)
704   {
705     GNUNET_HELLO_address_free (na->address);
706     na->address = NULL;
707   }
708   na->session = NULL;
709 }
710
711
712 /**
713  * Initialize the 'struct NeighbourAddress'.
714  *
715  * @param na neighbour address to initialize
716  * @param address address of the other peer, NULL if other peer
717  *                       connected to us
718  * @param session session to use (or NULL, in which case an
719  *        address must be setup)
720  * @param bandwidth_in inbound quota to be used when connection is up
721  * @param bandwidth_out outbound quota to be used when connection is up
722  * @param is_active GNUNET_YES to mark this as the active address with ATS
723  */
724 static void
725 set_address (struct NeighbourAddress *na,
726              const struct GNUNET_HELLO_Address *address,
727              struct Session *session,
728              struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
729              struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
730              int is_active)
731 {
732   struct GNUNET_TRANSPORT_PluginFunctions *papi;
733
734   if (NULL == (papi = GST_plugins_find (address->transport_name)))  
735   {
736     GNUNET_break (0);
737     return;
738   }
739   if (session == na->session)
740   {
741     na->bandwidth_in = bandwidth_in;
742     na->bandwidth_out = bandwidth_out;
743     if (is_active != na->ats_active)
744     {
745       na->ats_active = is_active;
746       GNUNET_ATS_address_in_use (GST_ats, na->address, na->session, is_active);
747       GST_validation_set_address_use (na->address, na->session, is_active,  __LINE__);
748     }
749     if (GNUNET_YES == is_active)
750     {
751       /* FIXME: is this the right place to set quotas? */
752       GST_neighbours_set_incoming_quota (&address->peer, bandwidth_in);
753       send_outbound_quota (&address->peer, bandwidth_out);
754     }    
755     return;
756   }
757   free_address (na);
758   if (NULL == session)
759     session = papi->get_session (papi->cls, address);    
760   if (NULL == session)
761   {
762     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
763                 "Failed to obtain new session for peer `%s' and  address '%s'\n",
764                 GNUNET_i2s (&address->peer), GST_plugins_a2s (address));    
765     GNUNET_ATS_address_destroyed (GST_ats, address, NULL);
766     return;
767   }
768   na->address = GNUNET_HELLO_address_copy (address);
769   na->bandwidth_in = bandwidth_in;
770   na->bandwidth_out = bandwidth_out;
771   na->session = session;
772   na->ats_active = is_active;
773   if (GNUNET_YES == is_active)
774   {
775     /* Telling ATS about new session */
776     GNUNET_ATS_address_update (GST_ats, na->address, na->session, NULL, 0);
777     GNUNET_ATS_address_in_use (GST_ats, na->address, na->session, GNUNET_YES);
778     GST_validation_set_address_use (na->address, na->session, GNUNET_YES,  __LINE__);
779
780     /* FIXME: is this the right place to set quotas? */
781     GST_neighbours_set_incoming_quota (&address->peer, bandwidth_in);
782     send_outbound_quota (&address->peer, bandwidth_out);
783   }
784 }
785
786
787 /**
788  * Free a neighbour map entry.
789  *
790  * @param n entry to free
791  */
792 static void
793 free_neighbour (struct NeighbourMapEntry *n)
794 {
795   struct MessageQueue *mq;
796   struct GNUNET_TRANSPORT_PluginFunctions *papi;
797
798   n->is_active = NULL; /* always free'd by its own continuation! */
799
800   /* fail messages currently in the queue */
801   while (NULL != (mq = n->messages_head))
802   {
803     GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
804     if (NULL != mq->cont)
805       mq->cont (mq->cont_cls, GNUNET_SYSERR);
806     GNUNET_free (mq);
807   }
808   /* It is too late to send other peer disconnect notifications, but at
809      least internally we need to get clean... */
810   if (GNUNET_YES == test_connected (n))
811   {
812     GNUNET_STATISTICS_set (GST_stats, 
813                            gettext_noop ("# peers connected"), 
814                            --neighbours_connected,
815                            GNUNET_NO);
816     disconnect_notify_cb (callback_cls, &n->id);
817   }
818
819   /* FIXME-PLUGIN-API: This does not seem to guarantee that all
820      transport sessions eventually get killed due to inactivity; they
821      MUST have their own timeout logic (but at least TCP doesn't have
822      one yet).  Are we sure that EVERY 'session' of a plugin is
823      actually cleaned up this way!?  Note that if we are switching
824      between two TCP sessions to the same peer, the existing plugin
825      API gives us not even the means to selectively kill only one of
826      them! Killing all sessions like this seems to be very, very
827      wrong. */
828   if ( (NULL != n->primary_address.address) &&
829        (NULL != (papi = GST_plugins_find (n->primary_address.address->transport_name))) )
830     papi->disconnect (papi->cls, &n->id);
831
832   n->state = S_DISCONNECT_FINISHED;
833
834   GNUNET_assert (GNUNET_YES ==
835                  GNUNET_CONTAINER_multihashmap_remove (neighbours,
836                                                        &n->id.hashPubKey, n));
837
838   /* cut transport-level connection */
839   free_address (&n->primary_address);
840   free_address (&n->alternative_address);
841
842   // FIXME-ATS-API: we might want to be more specific about
843   // which states we do this from in the future (ATS should
844   // have given us a 'suggest_address' handle, and if we have
845   // such a handle, we should cancel the operation here!
846   GNUNET_ATS_suggest_address_cancel (GST_ats, &n->id);
847
848   if (GNUNET_SCHEDULER_NO_TASK != n->task)
849   {
850     GNUNET_SCHEDULER_cancel (n->task);
851     n->task = GNUNET_SCHEDULER_NO_TASK;
852   }
853   /* free rest of memory */
854   GNUNET_free (n);
855 }
856
857
858 /**
859  * Transmit a message using the current session of the given
860  * neighbour.
861  *
862  * @param n entry for the recipient
863  * @param msgbuf buffer to transmit
864  * @param msgbuf_size number of bytes in buffer
865  * @param priority transmission priority
866  * @param timeout transmission timeout
867  * @param cont continuation to call when finished (can be NULL)
868  * @param cont_cls closure for cont
869  */
870 static void
871 send_with_session (struct NeighbourMapEntry *n,
872                    const char *msgbuf, size_t msgbuf_size,
873                    uint32_t priority,
874                    struct GNUNET_TIME_Relative timeout,
875                    GNUNET_TRANSPORT_TransmitContinuation cont,
876                    void *cont_cls)
877 {
878   struct GNUNET_TRANSPORT_PluginFunctions *papi;
879
880   GNUNET_assert (n->primary_address.session != NULL);
881   if ( ( (NULL == (papi = GST_plugins_find (n->primary_address.address->transport_name))) ||
882          (-1 == papi->send (papi->cls,
883                             n->primary_address.session,
884                             msgbuf, msgbuf_size,
885                             priority,
886                             timeout,
887                             cont, cont_cls))) &&
888        (NULL != cont) )
889     cont (cont_cls, &n->id, GNUNET_SYSERR);
890   GNUNET_break (NULL != papi);
891 }
892
893
894 /**
895  * Master task run for every neighbour.  Performs all of the time-related
896  * activities (keep alive, send next message, disconnect if idle, finish
897  * clean up after disconnect).
898  *
899  * @param cls the 'struct NeighbourMapEntry' for which we are running
900  * @param tc scheduler context (unused)
901  */
902 static void
903 master_task (void *cls,
904              const struct GNUNET_SCHEDULER_TaskContext *tc);
905
906
907 /**
908  * Function called when the 'DISCONNECT' message has been sent by the
909  * plugin.  Frees the neighbour --- if the entry still exists.
910  *
911  * @param cls NULL
912  * @param target identity of the neighbour that was disconnected
913  * @param result GNUNET_OK if the disconnect got out successfully
914  */
915 static void
916 send_disconnect_cont (void *cls, const struct GNUNET_PeerIdentity *target,
917                       int result)
918 {
919   struct NeighbourMapEntry *n;
920
921   n = lookup_neighbour (target);
922   if (NULL == n)
923     return; /* already gone */
924   if (S_DISCONNECT != n->state)
925     return; /* have created a fresh entry since */
926   n->state = S_DISCONNECT;
927   GNUNET_SCHEDULER_cancel (n->task);
928   n->task = GNUNET_SCHEDULER_add_now (&master_task, n);
929 }
930
931
932 /**
933  * Transmit a DISCONNECT message to the other peer.
934  *
935  * @param n neighbour to send DISCONNECT message.
936  */
937 static void
938 send_disconnect (struct NeighbourMapEntry *n)
939 {
940   struct SessionDisconnectMessage disconnect_msg;
941
942   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
943               "Sending DISCONNECT message to peer `%4s'\n",
944               GNUNET_i2s (&n->id));
945   disconnect_msg.header.size = htons (sizeof (struct SessionDisconnectMessage));
946   disconnect_msg.header.type =
947       htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT);
948   disconnect_msg.reserved = htonl (0);
949   disconnect_msg.purpose.size =
950       htonl (sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) +
951              sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) +
952              sizeof (struct GNUNET_TIME_AbsoluteNBO));
953   disconnect_msg.purpose.purpose =
954       htonl (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT);
955   disconnect_msg.timestamp =
956       GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
957   disconnect_msg.public_key = GST_my_public_key;
958   GNUNET_assert (GNUNET_OK ==
959                  GNUNET_CRYPTO_rsa_sign (GST_my_private_key,
960                                          &disconnect_msg.purpose,
961                                          &disconnect_msg.signature));
962
963   send_with_session (n,
964                      (const char *) &disconnect_msg, sizeof (disconnect_msg),
965                      UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
966                      &send_disconnect_cont, NULL);
967   GNUNET_STATISTICS_update (GST_stats,
968                             gettext_noop
969                             ("# DISCONNECT messages sent"), 1,
970                             GNUNET_NO);
971 }
972
973
974 /**
975  * Disconnect from the given neighbour, clean up the record.
976  *
977  * @param n neighbour to disconnect from
978  */
979 static void
980 disconnect_neighbour (struct NeighbourMapEntry *n)
981 {
982   /* depending on state, notify neighbour and/or upper layers of this peer 
983      about disconnect */
984   switch (n->state)
985   {
986   case S_NOT_CONNECTED:
987   case S_INIT_ATS:
988   case S_INIT_BLACKLIST:
989     /* other peer is completely unaware of us, no need to send DISCONNECT */
990     n->state = S_DISCONNECT_FINISHED;
991     free_neighbour (n);
992     return;
993   case S_CONNECT_SENT:
994     send_disconnect (n); 
995     n->state = S_DISCONNECT;
996     break;   
997   case S_CONNECT_RECV_ATS:
998   case S_CONNECT_RECV_BLACKLIST:
999     /* we never ACK'ed the other peer's request, no need to send DISCONNECT */
1000     n->state = S_DISCONNECT_FINISHED;
1001     free_neighbour (n);
1002     return;
1003   case S_CONNECT_RECV_ACK:
1004     /* we DID ACK the other peer's request, must send DISCONNECT */
1005     send_disconnect (n); 
1006     n->state = S_DISCONNECT;
1007     break;   
1008   case S_CONNECTED:
1009   case S_RECONNECT_ATS:
1010   case S_RECONNECT_BLACKLIST:
1011   case S_RECONNECT_SENT:
1012   case S_CONNECTED_SWITCHING_BLACKLIST:
1013   case S_CONNECTED_SWITCHING_CONNECT_SENT:
1014     /* we are currently connected, need to send disconnect and do
1015        internal notifications and update statistics */
1016     send_disconnect (n);
1017     GNUNET_STATISTICS_set (GST_stats, 
1018                            gettext_noop ("# peers connected"), 
1019                            --neighbours_connected,
1020                            GNUNET_NO);
1021     disconnect_notify_cb (callback_cls, &n->id);
1022     n->state = S_DISCONNECT;
1023     break;
1024   case S_DISCONNECT:
1025     /* already disconnected, ignore */
1026     break;
1027   case S_DISCONNECT_FINISHED:
1028     /* already cleaned up, how did we get here!? */
1029     GNUNET_assert (0);
1030     break;
1031   default:
1032     GNUNET_break (0);
1033     break;
1034   }
1035   /* schedule timeout to clean up */
1036   if (GNUNET_SCHEDULER_NO_TASK != n->task)
1037     GNUNET_SCHEDULER_cancel (n->task);
1038   n->task = GNUNET_SCHEDULER_add_delayed (DISCONNECT_SENT_TIMEOUT,
1039                                           &master_task, n);
1040 }
1041
1042
1043 /**
1044  * We're done with our transmission attempt, continue processing.
1045  *
1046  * @param cls the 'struct MessageQueue' of the message
1047  * @param receiver intended receiver
1048  * @param success whether it worked or not
1049  */
1050 static void
1051 transmit_send_continuation (void *cls,
1052                             const struct GNUNET_PeerIdentity *receiver,
1053                             int success)
1054 {
1055   struct MessageQueue *mq = cls;
1056   struct NeighbourMapEntry *n;
1057
1058   n = lookup_neighbour (receiver);
1059   if (NULL == n)
1060   {
1061     GNUNET_break (0);
1062     return;
1063   }
1064
1065   if (n->is_active == mq)
1066   {
1067     /* this is still "our" neighbour, remove us from its queue
1068        and allow it to send the next message now */
1069     n->is_active = NULL;
1070     GNUNET_SCHEDULER_cancel (n->task);
1071     n->task = GNUNET_SCHEDULER_add_now (&master_task, n);    
1072   }
1073   GNUNET_assert (bytes_in_send_queue >= mq->message_buf_size);
1074   bytes_in_send_queue -= mq->message_buf_size;
1075   GNUNET_STATISTICS_set (GST_stats,
1076                         gettext_noop
1077                          ("# bytes in message queue for other peers"),
1078                          bytes_in_send_queue, GNUNET_NO);
1079   if (GNUNET_OK == success)
1080     GNUNET_STATISTICS_update (GST_stats,
1081                               gettext_noop
1082                               ("# messages transmitted to other peers"),
1083                               1, GNUNET_NO);
1084   else
1085     GNUNET_STATISTICS_update (GST_stats,
1086                               gettext_noop
1087                               ("# transmission failures for messages to other peers"),
1088                               1, GNUNET_NO);
1089   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
1090               "Sending message to `%s' of type %u was a %s\n",
1091               GNUNET_i2s (receiver),
1092               ntohs (((struct GNUNET_MessageHeader *) mq->message_buf)->type),
1093               (success == GNUNET_OK) ? "success" : "FAILURE");
1094   if (NULL != mq->cont)
1095     mq->cont (mq->cont_cls, success);
1096   GNUNET_free (mq);
1097 }
1098
1099
1100 /**
1101  * Check the message list for the given neighbour and if we can
1102  * send a message, do so.  This function should only be called
1103  * if the connection is at least generally ready for transmission.
1104  * While we will only send one message at a time, no bandwidth
1105  * quota management is performed here.  If a message was given to
1106  * the plugin, the continuation will automatically re-schedule
1107  * the 'master' task once the next message might be transmitted.
1108  *
1109  * @param n target peer for which to transmit
1110  */
1111 static void
1112 try_transmission_to_peer (struct NeighbourMapEntry *n)
1113 {
1114   struct MessageQueue *mq;
1115   struct GNUNET_TIME_Relative timeout;
1116
1117   if (NULL == n->primary_address.address)
1118   {
1119     /* no address, why are we here? */
1120     GNUNET_break (0);
1121     return;
1122   }
1123   if ((0 == n->primary_address.address->address_length) && 
1124       (NULL == n->primary_address.session))
1125   {
1126     /* no address, why are we here? */
1127     GNUNET_break (0);
1128     return;
1129   }
1130   if (NULL != n->is_active)
1131   {
1132     /* transmission already pending */
1133     return;                     
1134   }
1135
1136   /* timeout messages from the queue that are past their due date */
1137   while (NULL != (mq = n->messages_head))
1138   {
1139     timeout = GNUNET_TIME_absolute_get_remaining (mq->timeout);
1140     if (timeout.rel_value > 0)
1141       break;
1142     GNUNET_STATISTICS_update (GST_stats,
1143                               gettext_noop
1144                               ("# messages timed out while in transport queue"),
1145                               1, GNUNET_NO);
1146     GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
1147     n->is_active = mq;
1148     transmit_send_continuation (mq, &n->id, GNUNET_SYSERR);     /* timeout */
1149   }
1150   if (NULL == mq)
1151     return;                     /* no more messages */
1152   GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
1153   n->is_active = mq;
1154   send_with_session (n,
1155                      mq->message_buf, mq->message_buf_size,
1156                      0 /* priority */, timeout,
1157                      &transmit_send_continuation, mq);
1158 }
1159
1160
1161 /**
1162  * Send keepalive message to the neighbour.  Must only be called
1163  * if we are on 'connected' state.  Will internally determine
1164  * if a keepalive is truly needed (so can always be called).
1165  *
1166  * @param n neighbour that went idle and needs a keepalive
1167  */
1168 static void
1169 send_keepalive (struct NeighbourMapEntry *n)
1170 {
1171   struct GNUNET_MessageHeader m;
1172
1173   GNUNET_assert (S_CONNECTED == n->state);
1174   if (GNUNET_TIME_absolute_get_remaining (n->keep_alive_time).rel_value > 0)
1175     return; /* no keepalive needed at this time */
1176   m.size = htons (sizeof (struct GNUNET_MessageHeader));
1177   m.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_KEEPALIVE);
1178   send_with_session (n,
1179                      (const void *) &m, sizeof (m),
1180                      UINT32_MAX /* priority */,
1181                      KEEPALIVE_FREQUENCY,
1182                      NULL, NULL);
1183   GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# keepalives sent"), 1,
1184                             GNUNET_NO);
1185   n->expect_latency_response = GNUNET_YES;
1186   n->last_keep_alive_time = GNUNET_TIME_absolute_get ();
1187   n->keep_alive_time = GNUNET_TIME_relative_to_absolute (KEEPALIVE_FREQUENCY);
1188 }
1189
1190
1191 /**
1192  * Keep the connection to the given neighbour alive longer,
1193  * we received a KEEPALIVE (or equivalent); send a response.
1194  *
1195  * @param neighbour neighbour to keep alive (by sending keep alive response)
1196  */
1197 void
1198 GST_neighbours_keepalive (const struct GNUNET_PeerIdentity *neighbour)
1199 {
1200   struct NeighbourMapEntry *n;
1201   struct GNUNET_MessageHeader m;
1202
1203   if (NULL == (n = lookup_neighbour (neighbour)))
1204   {
1205     GNUNET_STATISTICS_update (GST_stats,
1206                               gettext_noop
1207                               ("# KEEPALIVE messages discarded (peer unknown)"),
1208                               1, GNUNET_NO);
1209     return;
1210   }
1211   if (NULL == n->primary_address.session)
1212   {
1213     GNUNET_STATISTICS_update (GST_stats,
1214                               gettext_noop
1215                               ("# KEEPALIVE messages discarded (no session)"),
1216                               1, GNUNET_NO);
1217     return;
1218   }
1219   /* send reply to allow neighbour to measure latency */
1220   m.size = htons (sizeof (struct GNUNET_MessageHeader));
1221   m.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_KEEPALIVE_RESPONSE);
1222   send_with_session(n,
1223                     (const void *) &m, sizeof (m),
1224                     UINT32_MAX /* priority */,
1225                     KEEPALIVE_FREQUENCY,
1226                     NULL, NULL);
1227 }
1228
1229
1230 /**
1231  * We received a KEEP_ALIVE_RESPONSE message and use this to calculate
1232  * latency to this peer.  Pass the updated information (existing ats
1233  * plus calculated latency) to ATS.
1234  *
1235  * @param neighbour neighbour to keep alive
1236  * @param ats performance data
1237  * @param ats_count number of entries in ats
1238  */
1239 void
1240 GST_neighbours_keepalive_response (const struct GNUNET_PeerIdentity *neighbour,
1241                                    const struct GNUNET_ATS_Information *ats,
1242                                    uint32_t ats_count)
1243 {
1244   struct NeighbourMapEntry *n;
1245   uint32_t latency;
1246   struct GNUNET_ATS_Information ats_new[ats_count + 1];
1247
1248   if (NULL == (n = lookup_neighbour (neighbour)))
1249   {
1250     GNUNET_STATISTICS_update (GST_stats,
1251                               gettext_noop
1252                               ("# KEEPALIVE_RESPONSE messages discarded (not connected)"),
1253                               1, GNUNET_NO);
1254     return;
1255   }
1256   if ( (S_CONNECTED != n->state) ||
1257        (GNUNET_YES != n->expect_latency_response) )
1258   {
1259     GNUNET_STATISTICS_update (GST_stats,
1260                               gettext_noop
1261                               ("# KEEPALIVE_RESPONSE messages discarded (not expected)"),
1262                               1, GNUNET_NO);
1263     return;
1264   }
1265   n->expect_latency_response = GNUNET_NO;
1266   n->latency = GNUNET_TIME_absolute_get_duration (n->last_keep_alive_time);
1267   n->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
1268   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
1269               "Latency for peer `%s' is %llu ms\n",
1270               GNUNET_i2s (&n->id), n->latency.rel_value);
1271   memcpy (ats_new, ats, sizeof (struct GNUNET_ATS_Information) * ats_count);
1272   /* append latency */
1273   ats_new[ats_count].type = htonl (GNUNET_ATS_QUALITY_NET_DELAY);
1274   if (n->latency.rel_value > UINT32_MAX)
1275     latency = UINT32_MAX;
1276   else
1277     latency = n->latency.rel_value;
1278   ats_new[ats_count].value = htonl (latency);
1279   GNUNET_ATS_address_update (GST_ats, 
1280                              n->primary_address.address, 
1281                              n->primary_address.session, ats_new,
1282                              ats_count + 1);
1283 }
1284
1285
1286 /**
1287  * We have received a message from the given sender.  How long should
1288  * we delay before receiving more?  (Also used to keep the peer marked
1289  * as live).
1290  *
1291  * @param sender sender of the message
1292  * @param size size of the message
1293  * @param do_forward set to GNUNET_YES if the message should be forwarded to clients
1294  *                   GNUNET_NO if the neighbour is not connected or violates the quota,
1295  *                   GNUNET_SYSERR if the connection is not fully up yet
1296  * @return how long to wait before reading more from this sender
1297  */
1298 struct GNUNET_TIME_Relative
1299 GST_neighbours_calculate_receive_delay (const struct GNUNET_PeerIdentity
1300                                         *sender, ssize_t size, int *do_forward)
1301 {
1302   struct NeighbourMapEntry *n;
1303   struct GNUNET_TIME_Relative ret;
1304   
1305   if (NULL == neighbours)
1306   {
1307     *do_forward = GNUNET_NO;
1308     return GNUNET_TIME_UNIT_FOREVER_REL; /* This can happen during shutdown */
1309   }
1310   if (NULL == (n = lookup_neighbour (sender)))
1311   {
1312     GST_neighbours_try_connect (sender);
1313     if (NULL == (n = lookup_neighbour (sender)))
1314     {
1315       GNUNET_STATISTICS_update (GST_stats,
1316                                 gettext_noop
1317                                 ("# messages discarded due to lack of neighbour record"),
1318                                 1, GNUNET_NO);
1319       *do_forward = GNUNET_NO;
1320       return GNUNET_TIME_UNIT_ZERO;
1321     }
1322   }
1323   if (! test_connected (n))
1324   {
1325     *do_forward = GNUNET_SYSERR;
1326     return GNUNET_TIME_UNIT_ZERO;
1327   }
1328   if (GNUNET_YES == GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, size))
1329   {
1330     n->quota_violation_count++;
1331     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1332                 "Bandwidth quota (%u b/s) violation detected (total of %u).\n",
1333                 n->in_tracker.available_bytes_per_s__,
1334                 n->quota_violation_count);
1335     /* Discount 32k per violation */
1336     GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, -32 * 1024);
1337   }
1338   else
1339   {
1340     if (n->quota_violation_count > 0)
1341     {
1342       /* try to add 32k back */
1343       GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, 32 * 1024);
1344       n->quota_violation_count--;
1345     }
1346   }
1347   if (n->quota_violation_count > QUOTA_VIOLATION_DROP_THRESHOLD)
1348   {
1349     GNUNET_STATISTICS_update (GST_stats,
1350                               gettext_noop
1351                               ("# bandwidth quota violations by other peers"),
1352                               1, GNUNET_NO);
1353     *do_forward = GNUNET_NO;
1354     return GNUNET_CONSTANTS_QUOTA_VIOLATION_TIMEOUT;
1355   }
1356   *do_forward = GNUNET_YES;
1357   ret = GNUNET_BANDWIDTH_tracker_get_delay (&n->in_tracker, 32 * 1024);
1358   if (ret.rel_value > 0)
1359   {
1360     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1361                 "Throttling read (%llu bytes excess at %u b/s), waiting %llu ms before reading more.\n",
1362                 (unsigned long long) n->in_tracker.
1363                 consumption_since_last_update__,
1364                 (unsigned int) n->in_tracker.available_bytes_per_s__,
1365                 (unsigned long long) ret.rel_value);
1366     GNUNET_STATISTICS_update (GST_stats,
1367                               gettext_noop ("# ms throttling suggested"),
1368                               (int64_t) ret.rel_value, GNUNET_NO);
1369   }
1370   return ret;
1371 }
1372
1373
1374 /**
1375  * Transmit a message to the given target using the active connection.
1376  *
1377  * @param target destination
1378  * @param msg message to send
1379  * @param msg_size number of bytes in msg
1380  * @param timeout when to fail with timeout
1381  * @param cont function to call when done
1382  * @param cont_cls closure for 'cont'
1383  */
1384 void
1385 GST_neighbours_send (const struct GNUNET_PeerIdentity *target, const void *msg,
1386                      size_t msg_size, struct GNUNET_TIME_Relative timeout,
1387                      GST_NeighbourSendContinuation cont, void *cont_cls)
1388 {
1389   struct NeighbourMapEntry *n;
1390   struct MessageQueue *mq;
1391
1392   /* All ove these cases should never happen; they are all API violations.
1393      But we check anyway, just to be sure. */
1394   if (NULL == (n = lookup_neighbour (target)))
1395   {
1396     GNUNET_break (0);
1397     if (NULL != cont)
1398       cont (cont_cls, GNUNET_SYSERR);
1399     return;
1400   }
1401   if (GNUNET_YES != test_connected (n))
1402   {
1403     GNUNET_break (0);
1404     if (NULL != cont)
1405       cont (cont_cls, GNUNET_SYSERR);
1406     return;
1407   }
1408   if ((NULL == n->primary_address.session) && (NULL == n->primary_address.address))
1409   {
1410     GNUNET_break (0);
1411     if (NULL != cont)
1412       cont (cont_cls, GNUNET_SYSERR);
1413     return;
1414   }
1415
1416   bytes_in_send_queue += msg_size;
1417   GNUNET_STATISTICS_set (GST_stats,
1418                          gettext_noop
1419                          ("# bytes in message queue for other peers"),
1420                          bytes_in_send_queue, GNUNET_NO);
1421   mq = GNUNET_malloc (sizeof (struct MessageQueue) + msg_size);
1422   mq->cont = cont;
1423   mq->cont_cls = cont_cls;
1424   memcpy (&mq[1], msg, msg_size);
1425   mq->message_buf = (const char *) &mq[1];
1426   mq->message_buf_size = msg_size;
1427   mq->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1428   GNUNET_CONTAINER_DLL_insert_tail (n->messages_head, n->messages_tail, mq);
1429   if (NULL != n->is_active)
1430     return;
1431   GNUNET_SCHEDULER_cancel (n->task);
1432   n->task = GNUNET_SCHEDULER_add_now (&master_task, n);
1433 }
1434
1435
1436 /**
1437  * Send a SESSION_CONNECT message via the given address.
1438  *
1439  * @param na address to use
1440  */
1441 static void
1442 send_session_connect (struct NeighbourAddress *na)
1443 {
1444   struct GNUNET_TRANSPORT_PluginFunctions *papi;
1445   struct SessionConnectMessage connect_msg;
1446   
1447   if (NULL == (papi = GST_plugins_find (na->address->transport_name)))  
1448   {
1449     GNUNET_break (0);
1450     return;
1451   }
1452   if (NULL == na->session)
1453     na->session = papi->get_session (papi->cls, na->address);    
1454   if (NULL == na->session)
1455   {
1456     GNUNET_break (0);
1457     return;
1458   }
1459   na->connect_timestamp = GNUNET_TIME_absolute_get ();
1460   connect_msg.header.size = htons (sizeof (struct SessionConnectMessage));
1461   connect_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT);
1462   connect_msg.reserved = htonl (0);
1463   connect_msg.timestamp = GNUNET_TIME_absolute_hton (na->connect_timestamp);
1464   (void) papi->send (papi->cls,
1465                      na->session,
1466                      (const char *) &connect_msg, sizeof (struct SessionConnectMessage),
1467                      UINT_MAX,
1468                      GNUNET_TIME_UNIT_FOREVER_REL,
1469                      NULL, NULL);
1470 }
1471
1472
1473 /**
1474  * Send a SESSION_CONNECT_ACK message via the given address.
1475  *
1476  * @param address address to use
1477  * @param session session to use
1478  * @param timestamp timestamp to use for the ACK message
1479  */
1480 static void
1481 send_session_connect_ack_message (const struct GNUNET_HELLO_Address *address,
1482                                   struct Session *session,
1483                                   struct GNUNET_TIME_Absolute timestamp)
1484 {
1485   struct GNUNET_TRANSPORT_PluginFunctions *papi;
1486   struct SessionConnectMessage connect_msg;
1487   
1488   if (NULL == (papi = GST_plugins_find (address->transport_name)))  
1489   {
1490     GNUNET_break (0);
1491     return;
1492   }
1493   if (NULL == session)
1494     session = papi->get_session (papi->cls, address);    
1495   if (NULL == session)
1496   {
1497     GNUNET_break (0);
1498     return;
1499   }
1500   connect_msg.header.size = htons (sizeof (struct SessionConnectMessage));
1501   connect_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT_ACK);
1502   connect_msg.reserved = htonl (0);
1503   connect_msg.timestamp = GNUNET_TIME_absolute_hton (timestamp);
1504   (void) papi->send (papi->cls,
1505                      session,
1506                      (const char *) &connect_msg, sizeof (struct SessionConnectMessage),
1507                      UINT_MAX,
1508                      GNUNET_TIME_UNIT_FOREVER_REL,
1509                      NULL, NULL);
1510 }
1511
1512
1513 /**
1514  * Create a fresh entry in the neighbour map for the given peer
1515  *
1516  * @param peer peer to create an entry for
1517  * @return new neighbour map entry
1518  */
1519 static struct NeighbourMapEntry *
1520 setup_neighbour (const struct GNUNET_PeerIdentity *peer)
1521 {
1522   struct NeighbourMapEntry *n;
1523
1524   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1525               "Creating new neighbour entry for `%s'\n", 
1526               GNUNET_i2s (peer));
1527   n = GNUNET_malloc (sizeof (struct NeighbourMapEntry));
1528   n->id = *peer;
1529   n->state = S_NOT_CONNECTED;
1530   n->latency = GNUNET_TIME_UNIT_FOREVER_REL;
1531   GNUNET_BANDWIDTH_tracker_init (&n->in_tracker,
1532                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
1533                                  MAX_BANDWIDTH_CARRY_S);
1534   n->task = GNUNET_SCHEDULER_add_now (&master_task, n);
1535   GNUNET_assert (GNUNET_OK ==
1536                  GNUNET_CONTAINER_multihashmap_put (neighbours,
1537                                                     &n->id.hashPubKey, n,
1538                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1539   return n;
1540 }
1541
1542
1543 /**
1544  * Check if the two given addresses are the same.
1545  * Actually only checks if the sessions are non-NULL
1546  * (which they should be) and then if they are identical;
1547  * the actual addresses don't matter if the session
1548  * pointers match anyway, and we must have session pointers
1549  * at this time.
1550  *
1551  * @param a1 first address to compare
1552  * @param a2 other address to compare
1553  * @return GNUNET_NO if the addresses do not match, GNUNET_YES if they do match
1554  */
1555 static int
1556 address_matches (const struct NeighbourAddress *a1,
1557                  const struct NeighbourAddress *a2)
1558 {
1559   if ( (NULL == a1->session) ||
1560        (NULL == a2->session) )
1561   {
1562     GNUNET_break (0);
1563     return 0;
1564   }
1565   return (a1->session == a2->session) ? GNUNET_YES : GNUNET_NO;
1566 }
1567
1568
1569 /**
1570  * Try to create a connection to the given target (eventually).
1571  *
1572  * @param target peer to try to connect to
1573  */
1574 void
1575 GST_neighbours_try_connect (const struct GNUNET_PeerIdentity *target)
1576 {
1577   struct NeighbourMapEntry *n;
1578
1579   if (NULL == neighbours)  
1580     return; /* during shutdown, do nothing */
1581   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
1582               "Asked to connect to peer `%s'\n",
1583               GNUNET_i2s (target));
1584   if (0 ==
1585       memcmp (target, &GST_my_identity, sizeof (struct GNUNET_PeerIdentity)))
1586   {
1587     /* refuse to connect to myself */
1588     /* FIXME: can this happen? Is this not an API violation? */
1589     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1590                 "Refusing to try to connect to myself.\n");
1591     return;
1592   }
1593   n = lookup_neighbour (target);
1594   if (NULL != n)
1595   {
1596     switch (n->state)
1597     {
1598     case S_NOT_CONNECTED:
1599       /* this should not be possible */
1600       GNUNET_break (0);
1601       free_neighbour (n);
1602       break;
1603     case S_INIT_ATS:
1604     case S_INIT_BLACKLIST:
1605     case S_CONNECT_SENT:
1606     case S_CONNECT_RECV_ATS:
1607     case S_CONNECT_RECV_BLACKLIST:
1608     case S_CONNECT_RECV_ACK:
1609       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1610                   "Ignoring request to try to connect to `%s', already trying!\n",
1611                   GNUNET_i2s (target));
1612       return; /* already trying */
1613     case S_CONNECTED:      
1614     case S_RECONNECT_ATS:
1615     case S_RECONNECT_BLACKLIST:
1616     case S_RECONNECT_SENT:
1617     case S_CONNECTED_SWITCHING_BLACKLIST:
1618     case S_CONNECTED_SWITCHING_CONNECT_SENT:
1619       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1620                   "Ignoring request to try to connect, already connected to `%s'!\n",
1621                   GNUNET_i2s (target));
1622       return; /* already connected */
1623     case S_DISCONNECT:
1624       /* get rid of remains, ready to re-try immediately */
1625       free_neighbour (n);
1626       break;
1627     case S_DISCONNECT_FINISHED:
1628       /* should not be possible */      
1629       GNUNET_assert (0); 
1630     default:
1631       GNUNET_break (0);
1632       free_neighbour (n);
1633       break;
1634     }
1635   }
1636   n = setup_neighbour (target);  
1637   n->state = S_INIT_ATS; 
1638   n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
1639   GNUNET_ATS_suggest_address (GST_ats, target);
1640 }
1641
1642
1643 /**
1644  * Function called with the result of a blacklist check.
1645  *
1646  * @param cls closure with the 'struct BlackListCheckContext'
1647  * @param peer peer this check affects
1648  * @param result GNUNET_OK if the address is allowed
1649  */
1650 static void
1651 handle_test_blacklist_cont (void *cls,
1652                             const struct GNUNET_PeerIdentity *peer,
1653                             int result)
1654 {
1655   struct BlackListCheckContext *bcc = cls;
1656   struct NeighbourMapEntry *n;
1657
1658   bcc->bc = NULL;
1659   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1660               "Connection to new address of peer `%s' based on blacklist is `%s'\n",
1661               GNUNET_i2s (peer),
1662               (GNUNET_OK == result) ? "allowed" : "FORBIDDEN");
1663   if (GNUNET_OK == result)
1664   {
1665     /* valid new address, let ATS know! */
1666     GNUNET_ATS_address_update (GST_ats, 
1667                                bcc->na.address, 
1668                                bcc->na.session, 
1669                                bcc->ats, bcc->ats_count);
1670   }
1671   if (NULL == (n = lookup_neighbour (peer)))
1672     goto cleanup; /* nobody left to care about new address */
1673   switch (n->state)
1674   {
1675   case S_NOT_CONNECTED:
1676     /* this should not be possible */
1677     GNUNET_break (0);
1678     free_neighbour (n);
1679     break;
1680   case S_INIT_ATS:
1681     /* still waiting on ATS suggestion */
1682     break;
1683   case S_INIT_BLACKLIST:
1684     /* check if the address the blacklist was fine with matches
1685        ATS suggestion, if so, we can move on! */
1686     if (GNUNET_YES != address_matches (&bcc->na, &n->primary_address))
1687       break; /* result for an address we currently don't care about */
1688     if (GNUNET_OK == result)
1689     {
1690       n->timeout = GNUNET_TIME_relative_to_absolute (SETUP_CONNECTION_TIMEOUT);
1691       n->state = S_CONNECT_SENT;
1692       send_session_connect (&n->primary_address);
1693       if (1 == n->send_connect_ack) 
1694       {
1695         n->send_connect_ack = 2;
1696         send_session_connect_ack_message (bcc->na.address,
1697                                           bcc->na.session,
1698                                           n->connect_ack_timestamp);
1699       }
1700     }
1701     else
1702     {
1703       GNUNET_ATS_address_destroyed (GST_ats,
1704                                     bcc->na.address,
1705                                     NULL);
1706       n->state = S_INIT_ATS;
1707       n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
1708       // FIXME: do we need to ask ATS again for suggestions?
1709       GNUNET_ATS_suggest_address (GST_ats, &n->id);
1710     }
1711     break;
1712   case S_CONNECT_SENT:
1713     /* waiting on CONNECT_ACK, don't care about blacklist */
1714     break; 
1715   case S_CONNECT_RECV_ATS:
1716     /* still waiting on ATS suggestion, don't care about blacklist */
1717     break; 
1718   case S_CONNECT_RECV_BLACKLIST:
1719     if (GNUNET_YES != address_matches (&bcc->na, &n->primary_address))
1720       break; /* result for an address we currently don't care about */
1721     if (GNUNET_OK == result)
1722     {
1723       n->timeout = GNUNET_TIME_relative_to_absolute (SETUP_CONNECTION_TIMEOUT);
1724       n->state = S_CONNECT_RECV_ACK;
1725       send_session_connect_ack_message (bcc->na.address,
1726                                         bcc->na.session,
1727                                         n->connect_ack_timestamp);
1728       if (1 == n->send_connect_ack) 
1729         n->send_connect_ack = 2;
1730     }
1731     else
1732     {
1733       GNUNET_ATS_address_destroyed (GST_ats,
1734                                     bcc->na.address,
1735                                     NULL);
1736       n->state = S_INIT_ATS;
1737       n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
1738       // FIXME: do we need to ask ATS again for suggestions?
1739       GNUNET_ATS_suggest_address (GST_ats, &n->id);
1740     }
1741     break;
1742   case S_CONNECT_RECV_ACK:
1743     /* waiting on SESSION_ACK, don't care about blacklist */
1744     break; 
1745   case S_CONNECTED:
1746     /* already connected, don't care about blacklist */
1747     break;
1748   case S_RECONNECT_ATS:
1749     /* still waiting on ATS suggestion, don't care about blacklist */
1750     break;     
1751   case S_RECONNECT_BLACKLIST:
1752     if (GNUNET_YES != address_matches (&bcc->na, &n->primary_address))
1753       break; /* result for an address we currently don't care about */
1754     if (GNUNET_OK == result)
1755     {
1756       send_session_connect (&n->primary_address);
1757       n->timeout = GNUNET_TIME_relative_to_absolute (FAST_RECONNECT_TIMEOUT);
1758       n->state = S_RECONNECT_SENT;
1759       if (1 == n->send_connect_ack) 
1760       {
1761         n->send_connect_ack = 2;
1762         send_session_connect_ack_message (bcc->na.address,
1763                                           bcc->na.session,
1764                                           n->connect_ack_timestamp);
1765       }
1766     }
1767     else
1768     {
1769       GNUNET_ATS_address_destroyed (GST_ats,
1770                                     bcc->na.address,
1771                                     NULL);
1772       n->state = S_RECONNECT_ATS;
1773       n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
1774       // FIXME: do we need to ask ATS again for suggestions?
1775       GNUNET_ATS_suggest_address (GST_ats, &n->id);
1776     }
1777     break;
1778   case S_RECONNECT_SENT:
1779     /* waiting on CONNECT_ACK, don't care about blacklist */
1780     break;     
1781   case S_CONNECTED_SWITCHING_BLACKLIST:
1782     if (GNUNET_YES != address_matches (&bcc->na, &n->alternative_address))
1783       break; /* result for an address we currently don't care about */
1784     if (GNUNET_OK == result)
1785     {
1786       send_session_connect (&n->alternative_address);
1787       n->state = S_CONNECTED_SWITCHING_CONNECT_SENT;
1788     }
1789     else
1790     {
1791       GNUNET_ATS_address_destroyed (GST_ats,
1792                                     bcc->na.address,
1793                                     NULL);
1794       free_address (&n->alternative_address);
1795       n->state = S_CONNECTED;
1796     }
1797     break;
1798   case S_CONNECTED_SWITCHING_CONNECT_SENT:
1799     /* waiting on CONNECT_ACK, don't care about blacklist */
1800     break;     
1801   case S_DISCONNECT:
1802     /* Nothing to do here, ATS will already do what can be done */
1803     break;
1804   case S_DISCONNECT_FINISHED:
1805     /* should not be possible */
1806     GNUNET_assert (0);
1807     break;
1808   default:
1809     GNUNET_break (0);
1810     free_neighbour (n);
1811     break;
1812   }
1813  cleanup:
1814   GNUNET_HELLO_address_free (bcc->na.address);
1815   GNUNET_CONTAINER_DLL_remove (bc_head,
1816                                bc_tail,
1817                                bcc);
1818   GNUNET_free (bcc);
1819 }
1820
1821
1822 /**
1823  * We want to know if connecting to a particular peer via
1824  * a particular address is allowed.  Check it!
1825  *
1826  * @param peer identity of the peer to switch the address for
1827  * @param ts time at which the check was initiated
1828  * @param address address of the other peer, NULL if other peer
1829  *                       connected to us
1830  * @param session session to use (or NULL)
1831  * @param ats performance data
1832  * @param ats_count number of entries in ats (excluding 0-termination)
1833  */
1834 static void
1835 check_blacklist (const struct GNUNET_PeerIdentity *peer,
1836                  struct GNUNET_TIME_Absolute ts,
1837                  const struct GNUNET_HELLO_Address *address,
1838                  struct Session *session,
1839                  const struct GNUNET_ATS_Information *ats,
1840                  uint32_t ats_count)
1841 {
1842   struct BlackListCheckContext *bcc;
1843   struct GST_BlacklistCheck *bc;
1844
1845   bcc =
1846       GNUNET_malloc (sizeof (struct BlackListCheckContext) +
1847                      sizeof (struct GNUNET_ATS_Information) * ats_count);
1848   bcc->ats_count = ats_count;
1849   bcc->na.address = GNUNET_HELLO_address_copy (address);
1850   bcc->na.session = session;
1851   bcc->na.connect_timestamp = ts;
1852   bcc->ats = (struct GNUNET_ATS_Information *) &bcc[1];
1853   memcpy (bcc->ats, ats, sizeof (struct GNUNET_ATS_Information) * ats_count);
1854   GNUNET_CONTAINER_DLL_insert (bc_head,
1855                                bc_tail,
1856                                bcc);
1857   if (NULL != (bc = GST_blacklist_test_allowed (peer, 
1858                                                 address->transport_name,
1859                                                 &handle_test_blacklist_cont, bcc)))
1860     bcc->bc = bc; 
1861   /* if NULL == bc, 'cont' was already called and 'bcc' already free'd, so
1862      we must only store 'bc' if 'bc' is non-NULL... */
1863 }
1864
1865
1866 /**
1867  * We received a 'SESSION_CONNECT' message from the other peer.
1868  * Consider switching to it.
1869  *
1870  * @param message possibly a 'struct SessionConnectMessage' (check format)
1871  * @param peer identity of the peer to switch the address for
1872  * @param address address of the other peer, NULL if other peer
1873  *                       connected to us
1874  * @param session session to use (or NULL)
1875  * @param ats performance data
1876  * @param ats_count number of entries in ats (excluding 0-termination)
1877  */
1878 void
1879 GST_neighbours_handle_connect (const struct GNUNET_MessageHeader *message,
1880                                const struct GNUNET_PeerIdentity *peer,
1881                                const struct GNUNET_HELLO_Address *address,
1882                                struct Session *session,
1883                                const struct GNUNET_ATS_Information *ats,
1884                                uint32_t ats_count)
1885 {
1886   const struct SessionConnectMessage *scm;
1887   struct NeighbourMapEntry *n;
1888   struct GNUNET_TIME_Absolute ts;
1889
1890   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1891               "Received CONNECT message from peer `%s'\n", 
1892               GNUNET_i2s (peer));
1893   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
1894   {
1895     GNUNET_break_op (0);
1896     return;
1897   }
1898   if (NULL == neighbours)
1899     return; /* we're shutting down */
1900   scm = (const struct SessionConnectMessage *) message;
1901   GNUNET_break_op (0 == ntohl (scm->reserved));
1902   ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
1903   n = lookup_neighbour (peer);
1904   if (NULL == n)
1905     n = setup_neighbour (peer);
1906   n->send_connect_ack = 1;
1907   n->connect_ack_timestamp = ts;
1908   switch (n->state)
1909   {  
1910   case S_NOT_CONNECTED:
1911     n->state = S_CONNECT_RECV_ATS;
1912     n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
1913     GNUNET_ATS_suggest_address (GST_ats, peer);
1914     check_blacklist (peer, ts, address, session, ats, ats_count);
1915     break;
1916   case S_INIT_ATS:
1917   case S_INIT_BLACKLIST:
1918   case S_CONNECT_SENT:
1919   case S_CONNECT_RECV_ATS:
1920   case S_CONNECT_RECV_BLACKLIST:
1921   case S_CONNECT_RECV_ACK:
1922     /* It can never hurt to have an alternative address in the above cases, 
1923        see if it is allowed */
1924     check_blacklist (peer, ts, address, session, ats, ats_count);
1925     break;
1926   case S_CONNECTED:
1927     /* we are already connected and can thus send the ACK immediately;
1928        still, it can never hurt to have an alternative address, so also
1929        tell ATS  about it */
1930     GNUNET_assert (NULL != n->primary_address.address);
1931     GNUNET_assert (NULL != n->primary_address.session);
1932     n->send_connect_ack = 0;
1933     send_session_connect_ack_message (n->primary_address.address,
1934                                       n->primary_address.session, ts);
1935     check_blacklist (peer, ts, address, session, ats, ats_count);
1936     break;
1937   case S_RECONNECT_ATS:
1938   case S_RECONNECT_BLACKLIST:
1939   case S_RECONNECT_SENT:
1940     /* It can never hurt to have an alternative address in the above cases, 
1941        see if it is allowed */
1942     check_blacklist (peer, ts, address, session, ats, ats_count);
1943     break;
1944   case S_CONNECTED_SWITCHING_BLACKLIST:
1945   case S_CONNECTED_SWITCHING_CONNECT_SENT:
1946     /* we are already connected and can thus send the ACK immediately;
1947        still, it can never hurt to have an alternative address, so also
1948        tell ATS  about it */
1949     GNUNET_assert (NULL != n->primary_address.address);
1950     GNUNET_assert (NULL != n->primary_address.session);
1951     n->send_connect_ack = 0;
1952     send_session_connect_ack_message (n->primary_address.address,
1953                                       n->primary_address.session, ts);
1954     check_blacklist (peer, ts, address, session, ats, ats_count);
1955     break;
1956   case S_DISCONNECT:
1957     /* get rid of remains, ready to re-try */
1958     free_neighbour (n);
1959     n = setup_neighbour (peer);
1960     n->state = S_CONNECT_RECV_ATS;
1961     GNUNET_ATS_suggest_address (GST_ats, peer);
1962     check_blacklist (peer, ts, address, session, ats, ats_count);
1963     break;
1964   case S_DISCONNECT_FINISHED:
1965     /* should not be possible */
1966     GNUNET_assert (0);
1967     break;
1968   default:
1969     GNUNET_break (0);
1970     free_neighbour (n);
1971     break;
1972   }
1973 }
1974
1975
1976 /**
1977  * For an existing neighbour record, set the active connection to
1978  * use the given address.  
1979  *
1980  * @param peer identity of the peer to switch the address for
1981  * @param address address of the other peer, NULL if other peer
1982  *                       connected to us
1983  * @param session session to use (or NULL)
1984  * @param ats performance data
1985  * @param ats_count number of entries in ats
1986  * @param bandwidth_in inbound quota to be used when connection is up
1987  * @param bandwidth_out outbound quota to be used when connection is up
1988  */
1989 void
1990 GST_neighbours_switch_to_address (const struct GNUNET_PeerIdentity *peer,
1991                                   const struct GNUNET_HELLO_Address *address,
1992                                   struct Session *session,
1993                                   const struct GNUNET_ATS_Information *ats,
1994                                   uint32_t ats_count,
1995                                   struct GNUNET_BANDWIDTH_Value32NBO
1996                                   bandwidth_in,
1997                                   struct GNUNET_BANDWIDTH_Value32NBO
1998                                   bandwidth_out)
1999 {
2000   struct NeighbourMapEntry *n;
2001   struct GNUNET_TRANSPORT_PluginFunctions *papi;
2002
2003   GNUNET_assert (address->transport_name != NULL);
2004   if (NULL == (n = lookup_neighbour (peer)))
2005     return;
2006
2007   /* Obtain an session for this address from plugin */
2008   if (NULL == (papi = GST_plugins_find (address->transport_name)))
2009   {
2010     /* we don't have the plugin for this address */
2011     GNUNET_ATS_address_destroyed (GST_ats, address, NULL);
2012     return;
2013   }
2014   if ((NULL == session) && (0 == address->address_length))
2015   {
2016     GNUNET_break (0);
2017     if (strlen (address->transport_name) > 0)
2018       GNUNET_ATS_address_destroyed (GST_ats, address, session);
2019     return;
2020   }
2021   if (NULL == session)
2022     session = papi->get_session (papi->cls, address);
2023   if (NULL == session)
2024   {
2025     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2026                 "Failed to obtain new session for peer `%s' and  address '%s'\n",
2027                 GNUNET_i2s (&address->peer), GST_plugins_a2s (address));    
2028     GNUNET_ATS_address_destroyed (GST_ats, address, NULL);
2029     return;
2030   }
2031   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2032               "ATS tells us to switch to address '%s' for peer `%s'\n",
2033               (address->address_length != 0) ? GST_plugins_a2s (address): "<inbound>",
2034               GNUNET_i2s (peer));
2035   switch (n->state)
2036   {
2037   case S_NOT_CONNECTED:
2038     GNUNET_break (0);
2039     free_neighbour (n);
2040     return;
2041   case S_INIT_ATS:
2042     set_address (&n->primary_address,
2043                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2044     n->state = S_INIT_BLACKLIST;
2045     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2046     check_blacklist (&n->id,
2047                      n->connect_ack_timestamp,
2048                      address, session, ats, ats_count);    
2049     break;
2050   case S_INIT_BLACKLIST:
2051     /* ATS suggests a different address, switch again */
2052     set_address (&n->primary_address,
2053                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2054     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2055     check_blacklist (&n->id,
2056                      n->connect_ack_timestamp,
2057                      address, session, ats, ats_count);    
2058     break;
2059   case S_CONNECT_SENT:
2060     /* ATS suggests a different address, switch again */
2061     set_address (&n->primary_address,
2062                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2063     n->state = S_INIT_BLACKLIST;
2064     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2065     check_blacklist (&n->id,
2066                      n->connect_ack_timestamp,
2067                      address, session, ats, ats_count);    
2068     break;
2069   case S_CONNECT_RECV_ATS:
2070     set_address (&n->primary_address,
2071                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2072     n->state = S_CONNECT_RECV_BLACKLIST;
2073     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2074     check_blacklist (&n->id,
2075                      n->connect_ack_timestamp,
2076                      address, session, ats, ats_count);    
2077     break;
2078   case S_CONNECT_RECV_BLACKLIST:
2079     /* ATS asks us to switch while we were trying to connect; switch to new
2080        address and check blacklist again */
2081     set_address (&n->primary_address,
2082                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2083     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2084     check_blacklist (&n->id,
2085                      n->connect_ack_timestamp,
2086                      address, session, ats, ats_count);    
2087     break;
2088   case S_CONNECTED:
2089     GNUNET_assert (NULL != n->primary_address.address);
2090     GNUNET_assert (NULL != n->primary_address.session);
2091     if (n->primary_address.session == session)
2092     {
2093       /* not an address change, just a quota change */
2094       set_address (&n->primary_address,
2095                    address, session, bandwidth_in, bandwidth_out, GNUNET_YES);
2096       break;
2097     }
2098     /* ATS asks us to switch a life connection; see if we can get
2099        a CONNECT_ACK on it before we actually do this! */
2100     set_address (&n->alternative_address,
2101                  address, session, bandwidth_in, bandwidth_out, GNUNET_YES);
2102     n->state = S_CONNECTED_SWITCHING_BLACKLIST;
2103     check_blacklist (&n->id,
2104                      GNUNET_TIME_absolute_get (),
2105                      address, session, ats, ats_count);
2106     break;
2107   case S_RECONNECT_ATS:
2108     set_address (&n->primary_address,
2109                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2110     n->state = S_RECONNECT_BLACKLIST;
2111     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2112     check_blacklist (&n->id,
2113                      n->connect_ack_timestamp,
2114                      address, session, ats, ats_count);    
2115     break;
2116   case S_RECONNECT_BLACKLIST:
2117     /* ATS asks us to switch while we were trying to reconnect; switch to new
2118        address and check blacklist again */
2119     set_address (&n->primary_address,
2120                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2121     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2122     check_blacklist (&n->id,
2123                      n->connect_ack_timestamp,
2124                      address, session, ats, ats_count);    
2125     break;
2126   case S_RECONNECT_SENT:
2127     /* ATS asks us to switch while we were trying to reconnect; switch to new
2128        address and check blacklist again */
2129     set_address (&n->primary_address,
2130                  address, session, bandwidth_in, bandwidth_out, GNUNET_NO);
2131     n->state = S_RECONNECT_BLACKLIST;
2132     n->timeout = GNUNET_TIME_relative_to_absolute (BLACKLIST_RESPONSE_TIMEOUT);
2133     check_blacklist (&n->id,
2134                      n->connect_ack_timestamp,
2135                      address, session, ats, ats_count); 
2136     break;
2137   case S_CONNECTED_SWITCHING_BLACKLIST:
2138     if (n->primary_address.session == session)
2139     {
2140       /* ATS switches back to still-active session */
2141       free_address (&n->alternative_address);
2142       n->state = S_CONNECTED;
2143       break;
2144     }
2145     /* ATS asks us to switch a life connection, update blacklist check */
2146     set_address (&n->alternative_address,
2147                  address, session, bandwidth_in, bandwidth_out, GNUNET_YES);
2148     check_blacklist (&n->id,
2149                      GNUNET_TIME_absolute_get (),
2150                      address, session, ats, ats_count);
2151     break;
2152   case S_CONNECTED_SWITCHING_CONNECT_SENT:
2153     if (n->primary_address.session == session)
2154     {
2155       /* ATS switches back to still-active session */
2156       free_address (&n->alternative_address);
2157       n->state = S_CONNECTED;
2158       break;
2159     }
2160     /* ATS asks us to switch a life connection, update blacklist check */
2161     set_address (&n->alternative_address,
2162                  address, session, bandwidth_in, bandwidth_out, GNUNET_YES);
2163     n->state = S_CONNECTED_SWITCHING_BLACKLIST;
2164     check_blacklist (&n->id,
2165                      GNUNET_TIME_absolute_get (),
2166                      address, session, ats, ats_count);
2167     break;
2168   case S_DISCONNECT:
2169     /* not going to switch addresses while disconnecting */
2170     return;
2171   case S_DISCONNECT_FINISHED:
2172     GNUNET_assert (0);
2173     break;
2174   default:
2175     GNUNET_break (0);
2176     break;
2177   }
2178 }
2179
2180
2181 /**
2182  * Master task run for every neighbour.  Performs all of the time-related
2183  * activities (keep alive, send next message, disconnect if idle, finish
2184  * clean up after disconnect).
2185  *
2186  * @param cls the 'struct NeighbourMapEntry' for which we are running
2187  * @param tc scheduler context (unused)
2188  */
2189 static void
2190 master_task (void *cls,
2191              const struct GNUNET_SCHEDULER_TaskContext *tc)
2192 {
2193   struct NeighbourMapEntry *n = cls;
2194   struct GNUNET_TIME_Relative delay;
2195
2196   n->task = GNUNET_SCHEDULER_NO_TASK;
2197   delay = GNUNET_TIME_absolute_get_remaining (n->timeout);  
2198   switch (n->state)
2199   {
2200   case S_NOT_CONNECTED:
2201     /* invalid state for master task, clean up */
2202     GNUNET_break (0);
2203     n->state = S_DISCONNECT_FINISHED;
2204     free_neighbour (n);
2205     return;
2206   case S_INIT_ATS:
2207     if (0 == delay.rel_value)
2208     {
2209       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2210                   "Connection to `%s' timed out waiting for ATS to provide address\n",
2211                   GNUNET_i2s (&n->id));
2212       n->state = S_DISCONNECT_FINISHED;
2213       free_neighbour (n);
2214       return;
2215     }
2216     break;
2217   case S_INIT_BLACKLIST:
2218     if (0 == delay.rel_value)
2219     {
2220       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2221                   "Connection to `%s' timed out waiting for BLACKLIST to approve address\n",
2222                   GNUNET_i2s (&n->id));
2223       n->state = S_DISCONNECT_FINISHED;
2224       free_neighbour (n);
2225       return;
2226     }
2227     break;
2228   case S_CONNECT_SENT:
2229     if (0 == delay.rel_value)
2230     {
2231       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2232                   "Connection to `%s' timed out waiting for other peer to send CONNECT_ACK\n",
2233                   GNUNET_i2s (&n->id));
2234       disconnect_neighbour (n);
2235       return;
2236     }
2237     break;
2238   case S_CONNECT_RECV_ATS:
2239     if (0 == delay.rel_value)
2240     {
2241       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2242                   "Connection to `%s' timed out waiting ATS to provide address to use for CONNECT_ACK\n",
2243                   GNUNET_i2s (&n->id));
2244       n->state = S_DISCONNECT_FINISHED;
2245       free_neighbour (n);
2246       return;
2247     }
2248     break;
2249   case S_CONNECT_RECV_BLACKLIST:
2250     if (0 == delay.rel_value)
2251     {
2252       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2253                   "Connection to `%s' timed out waiting BLACKLIST to approve address to use for CONNECT_ACK\n",
2254                   GNUNET_i2s (&n->id));
2255       n->state = S_DISCONNECT_FINISHED;
2256       free_neighbour (n);
2257       return;
2258     }
2259     break;
2260   case S_CONNECT_RECV_ACK:
2261     if (0 == delay.rel_value)
2262     {
2263       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2264                   "Connection to `%s' timed out waiting for other peer to send SESSION_ACK\n",
2265                   GNUNET_i2s (&n->id));
2266       disconnect_neighbour (n);
2267       return;
2268     }
2269     break;
2270   case S_CONNECTED:
2271     if (0 == delay.rel_value)
2272     {
2273       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2274                   "Connection to `%s' timed out, missing KEEPALIVE_RESPONSEs\n",
2275                   GNUNET_i2s (&n->id));
2276       disconnect_neighbour (n);
2277       return;
2278     }
2279     try_transmission_to_peer (n);
2280     send_keepalive (n);
2281     break;
2282   case S_RECONNECT_ATS:
2283     if (0 == delay.rel_value)
2284     {
2285       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2286                   "Connection to `%s' timed out, waiting for ATS replacement address\n",
2287                   GNUNET_i2s (&n->id));
2288       disconnect_neighbour (n);
2289       return;
2290     }
2291     break;
2292   case S_RECONNECT_BLACKLIST:
2293     if (0 == delay.rel_value)
2294     {
2295       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2296                   "Connection to `%s' timed out, waiting for BLACKLIST to approve replacement address\n",
2297                   GNUNET_i2s (&n->id));
2298       disconnect_neighbour (n);
2299       return;
2300     }
2301     break;
2302   case S_RECONNECT_SENT:
2303     if (0 == delay.rel_value)
2304     {
2305       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2306                   "Connection to `%s' timed out, waiting for other peer to CONNECT_ACK replacement address\n",
2307                   GNUNET_i2s (&n->id));
2308       disconnect_neighbour (n);
2309       return;
2310     }
2311     break;
2312   case S_CONNECTED_SWITCHING_BLACKLIST:
2313     if (0 == delay.rel_value)
2314     {
2315       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2316                   "Connection to `%s' timed out, missing KEEPALIVE_RESPONSEs\n",
2317                   GNUNET_i2s (&n->id));
2318       disconnect_neighbour (n);
2319       return;
2320     }
2321     try_transmission_to_peer (n);
2322     send_keepalive (n);
2323     break;
2324   case S_CONNECTED_SWITCHING_CONNECT_SENT:
2325     if (0 == delay.rel_value)
2326     {
2327       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2328                   "Connection to `%s' timed out, missing KEEPALIVE_RESPONSEs (after trying to CONNECT on alternative address)\n",
2329                   GNUNET_i2s (&n->id));
2330       disconnect_neighbour (n);
2331       return;
2332     }
2333     try_transmission_to_peer (n);
2334     send_keepalive (n);
2335     break;
2336   case S_DISCONNECT:
2337     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2338                 "Cleaning up connection to `%s' after sending DISCONNECT\n",
2339                 GNUNET_i2s (&n->id));
2340     n->state = S_DISCONNECT_FINISHED;
2341     free_neighbour (n);
2342     return;
2343   case S_DISCONNECT_FINISHED:
2344     /* how did we get here!? */
2345     GNUNET_assert (0);
2346     break;
2347   default:
2348     GNUNET_break (0);
2349     break;  
2350   }
2351   delay = GNUNET_TIME_relative_min (GNUNET_TIME_absolute_get_remaining (n->keep_alive_time),
2352                                     delay);
2353   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == n->task);
2354   n->task = GNUNET_SCHEDULER_add_delayed (delay,
2355                                           &master_task,
2356                                           n);
2357 }
2358
2359
2360 /**
2361  * Send a SESSION_ACK message to the neighbour to confirm that we
2362  * got his CONNECT_ACK.
2363  *
2364  * @param n neighbour to send the SESSION_ACK to
2365  */
2366 static void
2367 send_session_ack_message (struct NeighbourMapEntry *n)
2368 {
2369   struct GNUNET_MessageHeader msg;
2370
2371   msg.size = htons (sizeof (struct GNUNET_MessageHeader));
2372   msg.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_ACK);
2373   (void) send_with_session(n,
2374                            (const char *) &msg, sizeof (struct GNUNET_MessageHeader),
2375                            UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
2376                            NULL, NULL);
2377 }
2378
2379
2380 /**
2381  * We received a 'SESSION_CONNECT_ACK' message from the other peer.
2382  * Consider switching to it.
2383  *
2384  * @param message possibly a 'struct SessionConnectMessage' (check format)
2385  * @param peer identity of the peer to switch the address for
2386  * @param address address of the other peer, NULL if other peer
2387  *                       connected to us
2388  * @param session session to use (or NULL)
2389  * @param ats performance data
2390  * @param ats_count number of entries in ats
2391  */
2392 void
2393 GST_neighbours_handle_connect_ack (const struct GNUNET_MessageHeader *message,
2394                                    const struct GNUNET_PeerIdentity *peer,
2395                                    const struct GNUNET_HELLO_Address *address,
2396                                    struct Session *session,
2397                                    const struct GNUNET_ATS_Information *ats,
2398                                    uint32_t ats_count)
2399 {
2400   const struct SessionConnectMessage *scm;
2401   struct GNUNET_TIME_Absolute ts;
2402   struct NeighbourMapEntry *n;
2403
2404   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2405               "Received CONNECT_ACK message from peer `%s'\n",
2406               GNUNET_i2s (peer));
2407   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
2408   {
2409     GNUNET_break_op (0);
2410     return;
2411   }
2412   scm = (const struct SessionConnectMessage *) message;
2413   GNUNET_break_op (ntohl (scm->reserved) == 0);
2414   if (NULL == (n = lookup_neighbour (peer)))
2415   {
2416     GNUNET_STATISTICS_update (GST_stats,
2417                               gettext_noop
2418                               ("# unexpected CONNECT_ACK messages (no peer)"),
2419                               1, GNUNET_NO);
2420     return;
2421   }
2422   ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
2423   switch (n->state)
2424   {
2425   case S_NOT_CONNECTED:
2426     GNUNET_break (0);
2427     free_neighbour (n);
2428     return;
2429   case S_INIT_ATS:
2430   case S_INIT_BLACKLIST:
2431     GNUNET_STATISTICS_update (GST_stats,
2432                               gettext_noop
2433                               ("# unexpected CONNECT_ACK messages (not ready)"),
2434                               1, GNUNET_NO);
2435     break;    
2436   case S_CONNECT_SENT:
2437     if (ts.abs_value != n->primary_address.connect_timestamp.abs_value)
2438       break; /* ACK does not match our original CONNECT message */
2439     n->state = S_CONNECTED;
2440     n->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2441     GNUNET_STATISTICS_set (GST_stats, 
2442                            gettext_noop ("# peers connected"), 
2443                            ++neighbours_connected,
2444                            GNUNET_NO);
2445     connect_notify_cb (callback_cls, &n->id, ats, ats_count);
2446     set_address (&n->primary_address,
2447                  n->primary_address.address,
2448                  n->primary_address.session,
2449                  n->primary_address.bandwidth_in,
2450                  n->primary_address.bandwidth_out,
2451                  GNUNET_YES);
2452     send_session_ack_message (n);
2453     break;
2454   case S_CONNECT_RECV_ATS:
2455   case S_CONNECT_RECV_BLACKLIST:
2456   case S_CONNECT_RECV_ACK:
2457     GNUNET_STATISTICS_update (GST_stats,
2458                               gettext_noop
2459                               ("# unexpected CONNECT_ACK messages (not ready)"),
2460                               1, GNUNET_NO);
2461     break;
2462   case S_CONNECTED:
2463     /* duplicate CONNECT_ACK, let's answer by duplciate SESSION_ACK just in case */
2464     send_session_ack_message (n);
2465     break;
2466   case S_RECONNECT_ATS:
2467   case S_RECONNECT_BLACKLIST:
2468     /* we didn't expect any CONNECT_ACK, as we are waiting for ATS
2469        to give us a new address... */
2470     GNUNET_STATISTICS_update (GST_stats,
2471                               gettext_noop
2472                               ("# unexpected CONNECT_ACK messages (waiting on ATS)"),
2473                               1, GNUNET_NO);
2474     break;
2475   case S_RECONNECT_SENT:
2476     /* new address worked; go back to connected! */
2477     n->state = S_CONNECTED;
2478     send_session_ack_message (n);
2479     break;
2480   case S_CONNECTED_SWITCHING_BLACKLIST:
2481     /* duplicate CONNECT_ACK, let's answer by duplciate SESSION_ACK just in case */
2482     send_session_ack_message (n);
2483     break;
2484   case S_CONNECTED_SWITCHING_CONNECT_SENT:
2485     /* new address worked; adopt it and go back to connected! */
2486     n->state = S_CONNECTED;
2487     n->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2488     GNUNET_assert (GNUNET_NO == n->alternative_address.ats_active);
2489     set_address (&n->primary_address,
2490                  n->alternative_address.address,
2491                  n->alternative_address.session,
2492                  n->alternative_address.bandwidth_in,
2493                  n->alternative_address.bandwidth_out,
2494                  GNUNET_YES);
2495     free_address (&n->alternative_address);
2496     send_session_ack_message (n);
2497     break;    
2498   case S_DISCONNECT:
2499     GNUNET_STATISTICS_update (GST_stats,
2500                               gettext_noop
2501                               ("# unexpected CONNECT_ACK messages (disconnecting)"),
2502                               1, GNUNET_NO);
2503     break;
2504   case S_DISCONNECT_FINISHED:
2505     GNUNET_assert (0);
2506     break;
2507   default:
2508     GNUNET_break (0);
2509     break;   
2510   }
2511 }
2512
2513
2514 /**
2515  * A session was terminated. Take note; if needed, try to get
2516  * an alternative address from ATS.
2517  *
2518  * @param peer identity of the peer where the session died
2519  * @param session session that is gone
2520  */
2521 void
2522 GST_neighbours_session_terminated (const struct GNUNET_PeerIdentity *peer,
2523                                    struct Session *session)
2524 {
2525   struct NeighbourMapEntry *n;
2526   struct BlackListCheckContext *bcc;
2527   struct BlackListCheckContext *bcc_next;
2528
2529   /* make sure to cancel all ongoing blacklist checks involving 'session' */
2530   bcc_next = bc_head;
2531   while (NULL != (bcc = bcc_next))
2532   {
2533     bcc_next = bcc->next;
2534     if (bcc->na.session == session)
2535     {
2536       GST_blacklist_test_cancel (bcc->bc);
2537       GNUNET_HELLO_address_free (bcc->na.address);
2538       GNUNET_CONTAINER_DLL_remove (bc_head,
2539                                    bc_tail,
2540                                    bcc);
2541       GNUNET_free (bcc);
2542     }
2543   }
2544   if (NULL == (n = lookup_neighbour (peer)))
2545     return; /* can't affect us */
2546   if (session != n->primary_address.session)
2547   {
2548     if (session == n->alternative_address.session)
2549     {
2550       free_address (&n->alternative_address);
2551       if ( (S_CONNECTED_SWITCHING_BLACKLIST == n->state) ||
2552            (S_CONNECTED_SWITCHING_CONNECT_SENT == n->state) )
2553         n->state = S_CONNECTED;
2554       else
2555         GNUNET_break (0);
2556     }
2557     return; /* doesn't affect us further */
2558   }
2559
2560   n->expect_latency_response = GNUNET_NO;
2561
2562   switch (n->state)
2563   {
2564   case S_NOT_CONNECTED:
2565     GNUNET_break (0);
2566     free_neighbour (n);
2567     return;
2568   case S_INIT_ATS:
2569     GNUNET_break (0);
2570     free_neighbour (n);
2571     return;
2572   case S_INIT_BLACKLIST:
2573   case S_CONNECT_SENT:
2574     free_address (&n->primary_address);
2575     n->state = S_INIT_ATS;
2576     n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
2577     // FIXME: need to ask ATS for suggestions again?
2578     GNUNET_ATS_suggest_address (GST_ats, &n->id);
2579     break;
2580   case S_CONNECT_RECV_ATS:    
2581   case S_CONNECT_RECV_BLACKLIST:
2582   case S_CONNECT_RECV_ACK:
2583     /* error on inbound session; free neighbour entirely */
2584     free_address (&n->primary_address);
2585     free_neighbour (n);
2586     return;
2587   case S_CONNECTED:
2588     free_address (&n->primary_address);
2589     n->state = S_RECONNECT_ATS;
2590     n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
2591     /* FIXME: is this ATS call needed? */
2592     GNUNET_ATS_suggest_address (GST_ats, &n->id);
2593     break;
2594   case S_RECONNECT_ATS:
2595     /* we don't have an address, how can it go down? */
2596     GNUNET_break (0);
2597     break;
2598   case S_RECONNECT_BLACKLIST:
2599   case S_RECONNECT_SENT:
2600     n->state = S_RECONNECT_ATS;
2601     n->timeout = GNUNET_TIME_relative_to_absolute (ATS_RESPONSE_TIMEOUT);
2602     // FIXME: need to ask ATS for suggestions again?
2603     GNUNET_ATS_suggest_address (GST_ats, &n->id);
2604     break;
2605   case S_CONNECTED_SWITCHING_BLACKLIST:
2606     /* primary went down while we were checking secondary against
2607        blacklist, adopt secondary as primary */       
2608     free_address (&n->primary_address);
2609     n->primary_address = n->alternative_address;
2610     memset (&n->alternative_address, 0, sizeof (struct NeighbourAddress));
2611     n->timeout = GNUNET_TIME_relative_to_absolute (FAST_RECONNECT_TIMEOUT);
2612     n->state = S_RECONNECT_BLACKLIST;
2613     break;
2614   case S_CONNECTED_SWITCHING_CONNECT_SENT:
2615     /* primary went down while we were waiting for CONNECT_ACK on secondary;
2616        secondary as primary */       
2617     free_address (&n->primary_address);
2618     n->primary_address = n->alternative_address;
2619     memset (&n->alternative_address, 0, sizeof (struct NeighbourAddress));
2620     n->timeout = GNUNET_TIME_relative_to_absolute (FAST_RECONNECT_TIMEOUT);
2621     n->state = S_RECONNECT_SENT;
2622     break;
2623   case S_DISCONNECT:
2624     free_address (&n->primary_address);
2625     break;
2626   case S_DISCONNECT_FINISHED:
2627     GNUNET_assert (0);
2628     break;
2629   default:
2630     GNUNET_break (0);
2631     break;
2632   }
2633   if (GNUNET_SCHEDULER_NO_TASK != n->task)
2634     GNUNET_SCHEDULER_cancel (n->task);
2635   n->task = GNUNET_SCHEDULER_add_now (&master_task, n);
2636 }
2637
2638
2639 /**
2640  * We received a 'SESSION_ACK' message from the other peer.
2641  * If we sent a 'CONNECT_ACK' last, this means we are now
2642  * connected.  Otherwise, do nothing.
2643  *
2644  * @param message possibly a 'struct SessionConnectMessage' (check format)
2645  * @param peer identity of the peer to switch the address for
2646  * @param address address of the other peer, NULL if other peer
2647  *                       connected to us
2648  * @param session session to use (or NULL)
2649  * @param ats performance data
2650  * @param ats_count number of entries in ats
2651  */
2652 void
2653 GST_neighbours_handle_session_ack (const struct GNUNET_MessageHeader *message,
2654                                    const struct GNUNET_PeerIdentity *peer,
2655                                    const struct GNUNET_HELLO_Address *address,
2656                                    struct Session *session,
2657                                    const struct GNUNET_ATS_Information *ats,
2658                                    uint32_t ats_count)
2659 {
2660   struct NeighbourMapEntry *n;
2661
2662   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
2663               "Received SESSION_ACK message from peer `%s'\n",
2664               GNUNET_i2s (peer));
2665   if (ntohs (message->size) != sizeof (struct GNUNET_MessageHeader))
2666   {
2667     GNUNET_break_op (0);
2668     return;
2669   }
2670   if (NULL == (n = lookup_neighbour (peer)))
2671     return;
2672   /* check if we are in a plausible state for having sent
2673      a CONNECT_ACK.  If not, return, otherwise break */
2674   if ( ( (S_CONNECT_RECV_ACK != n->state) &&
2675          (S_CONNECT_SENT != n->state) ) ||
2676        (2 != n->send_connect_ack) )
2677   {
2678     GNUNET_STATISTICS_update (GST_stats,
2679                               gettext_noop ("# unexpected SESSION ACK messages"), 1,
2680                               GNUNET_NO);
2681     return;
2682   }
2683   n->state = S_CONNECTED;
2684   n->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2685   GNUNET_STATISTICS_set (GST_stats, 
2686                          gettext_noop ("# peers connected"), 
2687                          ++neighbours_connected,
2688                          GNUNET_NO);
2689   connect_notify_cb (callback_cls, &n->id, ats, ats_count);
2690   set_address (&n->primary_address,
2691                n->primary_address.address,
2692                n->primary_address.session,
2693                n->primary_address.bandwidth_in,
2694                n->primary_address.bandwidth_out,
2695                GNUNET_YES);
2696 }
2697
2698
2699 /**
2700  * Test if we're connected to the given peer.
2701  *
2702  * @param target peer to test
2703  * @return GNUNET_YES if we are connected, GNUNET_NO if not
2704  */
2705 int
2706 GST_neighbours_test_connected (const struct GNUNET_PeerIdentity *target)
2707 {
2708   return test_connected (lookup_neighbour (target));
2709 }
2710
2711
2712 /**
2713  * Change the incoming quota for the given peer.
2714  *
2715  * @param neighbour identity of peer to change qutoa for
2716  * @param quota new quota
2717  */
2718 void
2719 GST_neighbours_set_incoming_quota (const struct GNUNET_PeerIdentity *neighbour,
2720                                    struct GNUNET_BANDWIDTH_Value32NBO quota)
2721 {
2722   struct NeighbourMapEntry *n;
2723
2724   if (NULL == (n = lookup_neighbour (neighbour)))
2725   {
2726     GNUNET_STATISTICS_update (GST_stats,
2727                               gettext_noop
2728                               ("# SET QUOTA messages ignored (no such peer)"),
2729                               1, GNUNET_NO);
2730     return;
2731   }
2732   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2733               "Setting inbound quota of %u Bps for peer `%s' to all clients\n",
2734               ntohl (quota.value__), GNUNET_i2s (&n->id));
2735   GNUNET_BANDWIDTH_tracker_update_quota (&n->in_tracker, quota);
2736   if (0 != ntohl (quota.value__))
2737     return;
2738   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting peer `%4s' due to `%s'\n",
2739               GNUNET_i2s (&n->id), "SET_QUOTA");
2740   if (GNUNET_YES == test_connected (n))
2741     GNUNET_STATISTICS_update (GST_stats,
2742                               gettext_noop ("# disconnects due to quota of 0"),
2743                               1, GNUNET_NO);
2744   disconnect_neighbour (n);
2745 }
2746
2747
2748 /**
2749  * We received a disconnect message from the given peer,
2750  * validate and process.
2751  *
2752  * @param peer sender of the message
2753  * @param msg the disconnect message
2754  */
2755 void
2756 GST_neighbours_handle_disconnect_message (const struct GNUNET_PeerIdentity
2757                                           *peer,
2758                                           const struct GNUNET_MessageHeader
2759                                           *msg)
2760 {
2761   struct NeighbourMapEntry *n;
2762   const struct SessionDisconnectMessage *sdm;
2763   GNUNET_HashCode hc;
2764
2765   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2766               "Received DISCONNECT message from peer `%s'\n",
2767               GNUNET_i2s (peer));
2768   if (ntohs (msg->size) != sizeof (struct SessionDisconnectMessage))
2769   {
2770     // GNUNET_break_op (0);
2771     GNUNET_STATISTICS_update (GST_stats,
2772                               gettext_noop
2773                               ("# disconnect messages ignored (old format)"), 1,
2774                               GNUNET_NO);
2775     return;
2776   }
2777   sdm = (const struct SessionDisconnectMessage *) msg;
2778   if (NULL == (n = lookup_neighbour (peer)))
2779     return;                     /* gone already */
2780   if (GNUNET_TIME_absolute_ntoh (sdm->timestamp).abs_value <= n->connect_ack_timestamp.abs_value)
2781   {
2782     GNUNET_STATISTICS_update (GST_stats,
2783                               gettext_noop
2784                               ("# disconnect messages ignored (timestamp)"), 1,
2785                               GNUNET_NO);
2786     return;
2787   }
2788   GNUNET_CRYPTO_hash (&sdm->public_key,
2789                       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
2790                       &hc);
2791   if (0 != memcmp (peer, &hc, sizeof (struct GNUNET_PeerIdentity)))
2792   {
2793     GNUNET_break_op (0);
2794     return;
2795   }
2796   if (ntohl (sdm->purpose.size) !=
2797       sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) +
2798       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) +
2799       sizeof (struct GNUNET_TIME_AbsoluteNBO))
2800   {
2801     GNUNET_break_op (0);
2802     return;
2803   }
2804   if (GNUNET_OK !=
2805       GNUNET_CRYPTO_rsa_verify
2806       (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT, &sdm->purpose,
2807        &sdm->signature, &sdm->public_key))
2808   {
2809     GNUNET_break_op (0);
2810     return;
2811   }
2812   if (GNUNET_YES == test_connected (n))
2813     GNUNET_STATISTICS_update (GST_stats,
2814                               gettext_noop
2815                               ("# other peer asked to disconnect from us"), 1,
2816                               GNUNET_NO);
2817   disconnect_neighbour (n);
2818 }
2819
2820
2821 /**
2822  * Closure for the neighbours_iterate function.
2823  */
2824 struct IteratorContext
2825 {
2826   /**
2827    * Function to call on each connected neighbour.
2828    */
2829   GST_NeighbourIterator cb;
2830
2831   /**
2832    * Closure for 'cb'.
2833    */
2834   void *cb_cls;
2835 };
2836
2837
2838 /**
2839  * Call the callback from the closure for each connected neighbour.
2840  *
2841  * @param cls the 'struct IteratorContext'
2842  * @param key the hash of the public key of the neighbour
2843  * @param value the 'struct NeighbourMapEntry'
2844  * @return GNUNET_OK (continue to iterate)
2845  */
2846 static int
2847 neighbours_iterate (void *cls, const GNUNET_HashCode * key, void *value)
2848 {
2849   struct IteratorContext *ic = cls;
2850   struct NeighbourMapEntry *n = value;
2851
2852   if (GNUNET_YES == test_connected (n))
2853     ic->cb (ic->cb_cls, &n->id, NULL, 0, n->primary_address.address);
2854   return GNUNET_OK;
2855 }
2856
2857
2858 /**
2859  * Iterate over all connected neighbours.
2860  *
2861  * @param cb function to call
2862  * @param cb_cls closure for cb
2863  */
2864 void
2865 GST_neighbours_iterate (GST_NeighbourIterator cb, void *cb_cls)
2866 {
2867   struct IteratorContext ic;
2868
2869   if (NULL == neighbours)  
2870     return; /* can happen during shutdown */
2871   ic.cb = cb;
2872   ic.cb_cls = cb_cls;
2873   GNUNET_CONTAINER_multihashmap_iterate (neighbours, &neighbours_iterate, &ic);
2874 }
2875
2876
2877 /**
2878  * If we have an active connection to the given target, it must be shutdown.
2879  *
2880  * @param target peer to disconnect from
2881  */
2882 void
2883 GST_neighbours_force_disconnect (const struct GNUNET_PeerIdentity *target)
2884 {
2885   struct NeighbourMapEntry *n;
2886
2887   if (NULL == (n = lookup_neighbour (target)))
2888     return;  /* not active */
2889   if (GNUNET_YES == test_connected (n))
2890     GNUNET_STATISTICS_update (GST_stats,
2891                               gettext_noop
2892                               ("# disconnected from peer upon explicit request"), 1,
2893                               GNUNET_NO);
2894   disconnect_neighbour (n);
2895 }
2896
2897
2898 /**
2899  * Obtain current latency information for the given neighbour.
2900  *
2901  * @param peer to get the latency for
2902  * @return observed latency of the address, FOREVER if the 
2903  *         the connection is not up
2904  */
2905 struct GNUNET_TIME_Relative
2906 GST_neighbour_get_latency (const struct GNUNET_PeerIdentity *peer)
2907 {
2908   struct NeighbourMapEntry *n;
2909
2910   n = lookup_neighbour (peer);
2911   if (NULL == n) 
2912     return GNUNET_TIME_UNIT_FOREVER_REL;
2913   switch (n->state)
2914   {
2915   case S_CONNECTED:
2916   case S_RECONNECT_SENT:
2917   case S_RECONNECT_ATS:
2918     return n->latency;
2919   case S_NOT_CONNECTED:
2920   case S_INIT_BLACKLIST:
2921   case S_INIT_ATS:
2922   case S_CONNECT_SENT:
2923   case S_CONNECT_RECV_BLACKLIST:
2924   case S_DISCONNECT:
2925   case S_DISCONNECT_FINISHED:
2926     return GNUNET_TIME_UNIT_FOREVER_REL;
2927   default:
2928     GNUNET_break (0);
2929     break;
2930   }
2931   return GNUNET_TIME_UNIT_FOREVER_REL;   
2932 }
2933
2934
2935 /**
2936  * Obtain current address information for the given neighbour.
2937  *
2938  * @param peer
2939  * @return address currently used
2940  */
2941 struct GNUNET_HELLO_Address *
2942 GST_neighbour_get_current_address (const struct GNUNET_PeerIdentity *peer)
2943 {
2944   struct NeighbourMapEntry *n;
2945
2946   n = lookup_neighbour (peer);
2947   if (NULL == n)
2948     return NULL;
2949   return n->primary_address.address;
2950 }
2951
2952
2953 /**
2954  * Initialize the neighbours subsystem.
2955  *
2956  * @param cls closure for callbacks
2957  * @param connect_cb function to call if we connect to a peer
2958  * @param disconnect_cb function to call if we disconnect from a peer
2959  * @param peer_address_cb function to call if we change an active address
2960  *                   of a neighbour
2961  */
2962 void
2963 GST_neighbours_start (void *cls,
2964                       GNUNET_TRANSPORT_NotifyConnect connect_cb,
2965                       GNUNET_TRANSPORT_NotifyDisconnect disconnect_cb,
2966                       GNUNET_TRANSPORT_PeerIterateCallback peer_address_cb)
2967 {
2968   callback_cls = cls;
2969   connect_notify_cb = connect_cb;
2970   disconnect_notify_cb = disconnect_cb;
2971   address_change_cb = peer_address_cb;
2972   neighbours = GNUNET_CONTAINER_multihashmap_create (NEIGHBOUR_TABLE_SIZE);
2973 }
2974
2975
2976 /**
2977  * Disconnect from the given neighbour.
2978  *
2979  * @param cls unused
2980  * @param key hash of neighbour's public key (not used)
2981  * @param value the 'struct NeighbourMapEntry' of the neighbour
2982  * @return GNUNET_OK (continue to iterate)
2983  */
2984 static int
2985 disconnect_all_neighbours (void *cls, const GNUNET_HashCode * key, void *value)
2986 {
2987   struct NeighbourMapEntry *n = value;
2988
2989   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
2990               "Disconnecting peer `%4s', %s\n",
2991               GNUNET_i2s (&n->id), "SHUTDOWN_TASK");
2992   free_neighbour (n);
2993   return GNUNET_OK;
2994 }
2995
2996
2997 /**
2998  * Cleanup the neighbours subsystem.
2999  */
3000 void
3001 GST_neighbours_stop ()
3002 {
3003   if (NULL == neighbours)
3004     return;
3005   GNUNET_CONTAINER_multihashmap_iterate (neighbours, 
3006                                          &disconnect_all_neighbours,
3007                                          NULL);
3008   GNUNET_CONTAINER_multihashmap_destroy (neighbours);
3009   neighbours = NULL;
3010   callback_cls = NULL;
3011   connect_notify_cb = NULL;
3012   disconnect_notify_cb = NULL;
3013   address_change_cb = NULL;
3014 }
3015
3016
3017 /* end of file gnunet-service-transport_neighbours.c */