do not call send_disconnect with NULL
[oweals/gnunet.git] / src / transport / gnunet-service-transport_neighbours.c
1 /*
2      This file is part of GNUnet.
3      (C) 2010,2011 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 #include "platform.h"
27 #include "gnunet_ats_service.h"
28 #include "gnunet-service-transport_neighbours.h"
29 #include "gnunet-service-transport_plugins.h"
30 #include "gnunet-service-transport_validation.h"
31 #include "gnunet-service-transport_clients.h"
32 #include "gnunet-service-transport.h"
33 #include "gnunet_peerinfo_service.h"
34 #include "gnunet-service-transport_blacklist.h"
35 #include "gnunet_constants.h"
36 #include "transport.h"
37
38
39 /**
40  * Size of the neighbour hash map.
41  */
42 #define NEIGHBOUR_TABLE_SIZE 256
43
44 /**
45  * How often must a peer violate bandwidth quotas before we start
46  * to simply drop its messages?
47  */
48 #define QUOTA_VIOLATION_DROP_THRESHOLD 10
49
50 /**
51  * How often do we send KEEPALIVE messages to each of our neighbours?
52  * (idle timeout is 5 minutes or 300 seconds, so with 90s interval we
53  * send 3 keepalives in each interval, so 3 messages would need to be
54  * lost in a row for a disconnect).
55  */
56 #define KEEPALIVE_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 90)
57
58
59 #define ATS_RESPONSE_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 3)
60
61
62 #define SETUP_CONNECTION_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 15)
63
64
65 /**
66  * Entry in neighbours.
67  */
68 struct NeighbourMapEntry;
69
70 /**
71  * Message a peer sends to another to indicate its
72  * preference for communicating via a particular
73  * session (and the desire to establish a real
74  * connection).
75  */
76 struct SessionConnectMessage
77 {
78   /**
79    * Header of type 'GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT'
80    */
81   struct GNUNET_MessageHeader header;
82
83   /**
84    * Always zero.
85    */
86   uint32_t reserved GNUNET_PACKED;
87
88   /**
89    * Absolute time at the sender.  Only the most recent connect
90    * message implies which session is preferred by the sender.
91    */
92   struct GNUNET_TIME_AbsoluteNBO timestamp;
93
94 };
95
96
97 struct SessionDisconnectMessage
98 {
99   /**
100    * Header of type 'GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT'
101    */
102   struct GNUNET_MessageHeader header;
103
104   /**
105    * Always zero.
106    */
107   uint32_t reserved GNUNET_PACKED;
108
109   /**
110    * Purpose of the signature.  Extends over the timestamp.
111    * Purpose should be GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DISCONNECT.
112    */
113   struct GNUNET_CRYPTO_RsaSignaturePurpose purpose;
114
115   /**
116    * Absolute time at the sender.  Only the most recent connect
117    * message implies which session is preferred by the sender.
118    */
119   struct GNUNET_TIME_AbsoluteNBO timestamp;
120
121   /**
122    * Public key of the sender.
123    */
124   struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded public_key;
125   
126   /**
127    * Signature of the peer that sends us the disconnect.  Only
128    * valid if the timestamp is AFTER the timestamp from the
129    * corresponding 'CONNECT' message.
130    */
131   struct GNUNET_CRYPTO_RsaSignature signature;
132
133 };
134
135
136 /**
137  * For each neighbour we keep a list of messages
138  * that we still want to transmit to the neighbour.
139  */
140 struct MessageQueue
141 {
142
143   /**
144    * This is a doubly linked list.
145    */
146   struct MessageQueue *next;
147
148   /**
149    * This is a doubly linked list.
150    */
151   struct MessageQueue *prev;
152
153   /**
154    * Once this message is actively being transmitted, which
155    * neighbour is it associated with?
156    */
157   struct NeighbourMapEntry *n;
158
159   /**
160    * Function to call once we're done.
161    */
162   GST_NeighbourSendContinuation cont;
163
164   /**
165    * Closure for 'cont'
166    */
167   void *cont_cls;
168
169   /**
170    * The message(s) we want to transmit, GNUNET_MessageHeader(s)
171    * stuck together in memory.  Allocated at the end of this struct.
172    */
173   const char *message_buf;
174
175   /**
176    * Size of the message buf
177    */
178   size_t message_buf_size;
179
180   /**
181    * At what time should we fail?
182    */
183   struct GNUNET_TIME_Absolute timeout;
184
185 };
186
187 enum State
188 {
189     /* fresh peer or completely disconnected */
190     S_NOT_CONNECTED = 0,
191     /* sent CONNECT message to other peer, waiting for CONNECT_ACK */
192     S_CONNECT_SENT = 1,
193     /* received CONNECT message to other peer, sending CONNECT_ACK */
194     S_CONNECT_RECV = 4,
195     /* sent CONNECT_ACK message to other peer, wait for ACK or payload */
196     S_CONNECT_RECV_ACK_SENT = 8,
197     /* received ACK or payload */
198     S_CONNECTED = 16,
199     /* Disconnect in progress */
200     S_DISCONNECT = 32
201 };
202
203 /**
204  * Entry in neighbours.
205  */
206 struct NeighbourMapEntry
207 {
208
209   /**
210    * Head of list of messages we would like to send to this peer;
211    * must contain at most one message per client.
212    */
213   struct MessageQueue *messages_head;
214
215   /**
216    * Tail of list of messages we would like to send to this peer; must
217    * contain at most one message per client.
218    */
219   struct MessageQueue *messages_tail;
220
221   /**
222    * Performance data for the peer.
223    */
224   //struct GNUNET_ATS_Information *ats;
225
226   /**
227    * Are we currently trying to send a message? If so, which one?
228    */
229   struct MessageQueue *is_active;
230
231   /**
232    * Active session for communicating with the peer.
233    */
234   struct Session *session;
235
236   /**
237    * Name of the plugin we currently use.
238    */
239   char *plugin_name;
240
241   /**
242    * Address used for communicating with the peer, NULL for inbound connections.
243    */
244   void *addr;
245
246   /**
247    * Number of bytes in 'addr'.
248    */
249   size_t addrlen;
250
251   /**
252    * Identity of this neighbour.
253    */
254   struct GNUNET_PeerIdentity id;
255
256   /**
257    * ID of task scheduled to run when this peer is about to
258    * time out (will free resources associated with the peer).
259    */
260   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
261
262   /**
263    * ID of task scheduled to send keepalives.
264    */
265   GNUNET_SCHEDULER_TaskIdentifier keepalive_task;
266
267   /**
268    * ID of task scheduled to run when we should try transmitting
269    * the head of the message queue.
270    */
271   GNUNET_SCHEDULER_TaskIdentifier transmission_task;
272
273   /**
274    * Tracker for inbound bandwidth.
275    */
276   struct GNUNET_BANDWIDTH_Tracker in_tracker;
277
278   /**
279    * Inbound bandwidth from ATS, activated when connection is up
280    */
281   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in;
282
283   /**
284    * Inbound bandwidth from ATS, activated when connection is up
285    */
286   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out;
287
288   /**
289    * Timestamp of the 'SESSION_CONNECT' message we got from the other peer
290    */
291   struct GNUNET_TIME_Absolute connect_ts;
292
293   /**
294    * Timeout for ATS
295    * We asked ATS for a new address for this peer
296    */
297   GNUNET_SCHEDULER_TaskIdentifier ats_suggest;
298
299   /**
300    * Task the resets the peer state after due to an pending
301    * unsuccessful connection setup
302    */
303   GNUNET_SCHEDULER_TaskIdentifier state_reset;
304
305   /**
306    * How often has the other peer (recently) violated the inbound
307    * traffic limit?  Incremented by 10 per violation, decremented by 1
308    * per non-violation (for each time interval).
309    */
310   unsigned int quota_violation_count;
311
312
313   /**
314    * The current state of the peer
315    * Element of enum State
316    */
317   int state;
318
319 };
320
321
322 /**
323  * All known neighbours and their HELLOs.
324  */
325 static struct GNUNET_CONTAINER_MultiHashMap *neighbours;
326
327 /**
328  * Closure for connect_notify_cb and disconnect_notify_cb
329  */
330 static void *callback_cls;
331
332 /**
333  * Function to call when we connected to a neighbour.
334  */
335 static GNUNET_TRANSPORT_NotifyConnect connect_notify_cb;
336
337 /**
338  * Function to call when we disconnected from a neighbour.
339  */
340 static GNUNET_TRANSPORT_NotifyDisconnect disconnect_notify_cb;
341
342 /**
343  * counter for connected neighbours
344  */
345 static int neighbours_connected;
346
347 /**
348  * Lookup a neighbour entry in the neighbours hash map.
349  *
350  * @param pid identity of the peer to look up
351  * @return the entry, NULL if there is no existing record
352  */
353 static struct NeighbourMapEntry *
354 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
355 {
356   return GNUNET_CONTAINER_multihashmap_get (neighbours, &pid->hashPubKey);
357 }
358
359 #define change_state(n, state, ...) change (n, state, __LINE__)
360
361 static int
362 is_connecting (struct NeighbourMapEntry * n)
363 {
364   if ((n->state > S_NOT_CONNECTED) && (n->state < S_CONNECTED))
365     return GNUNET_YES;
366   return GNUNET_NO;
367 }
368
369 static int
370 is_connected (struct NeighbourMapEntry * n)
371 {
372   if (n->state == S_CONNECTED)
373     return GNUNET_YES;
374   return GNUNET_NO;
375 }
376
377 static int
378 is_disconnecting (struct NeighbourMapEntry * n)
379 {
380   if (n->state == S_DISCONNECT)
381     return GNUNET_YES;
382   return GNUNET_NO;
383 }
384
385 static const char *
386 print_state (int state)
387 {
388   switch (state) {
389     case S_CONNECTED:
390         return "S_CONNECTED";
391       break;
392     case S_CONNECT_RECV:
393       return "S_CONNECT_RECV";
394       break;
395     case S_CONNECT_RECV_ACK_SENT:
396       return"S_CONNECT_RECV_ACK_SENT";
397       break;
398     case S_CONNECT_SENT:
399       return "S_CONNECT_SENT";
400       break;
401     case S_DISCONNECT:
402       return "S_DISCONNECT";
403       break;
404     case S_NOT_CONNECTED:
405       return "S_NOT_CONNECTED";
406       break;
407     default:
408       GNUNET_break (0);
409       break;
410   }
411   return NULL;
412 }
413
414 static int
415 change (struct NeighbourMapEntry * n, int state, int line);
416
417 static void
418 ats_suggest_cancel (void *cls,
419     const struct GNUNET_SCHEDULER_TaskContext *tc);
420
421 static void
422 reset_task (void *cls,
423             const struct GNUNET_SCHEDULER_TaskContext *tc)
424 {
425   struct NeighbourMapEntry * n = cls;
426
427   n->state_reset = GNUNET_SCHEDULER_NO_TASK;
428
429 #if DEBUG_TRANSPORT
430   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
431       "Connection to peer `%s' %s failed in state `%s', resetting connection attempt \n",
432       GNUNET_i2s (&n->id), GST_plugins_a2s(n->plugin_name, n->addr, n->addrlen), print_state(n->state));
433 #endif
434   GNUNET_STATISTICS_update (GST_stats,
435                             gettext_noop ("# failed connection attempts due to timeout"),
436                             1,
437                             GNUNET_NO);
438
439   /* resetting state */
440   n->state = S_NOT_CONNECTED;
441
442   /* destroying address */
443   GNUNET_ATS_address_destroyed (GST_ats,
444                                 &n->id,
445                                 n->plugin_name,
446                                 n->addr,
447                                 n->addrlen,
448                                 NULL);
449
450   /* request new address */
451   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
452     GNUNET_SCHEDULER_cancel(n->ats_suggest);
453   n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
454   GNUNET_ATS_suggest_address(GST_ats, &n->id);
455 }
456
457 static int
458 change (struct NeighbourMapEntry * n, int state, int line)
459 {
460   char * old = strdup(print_state(n->state));
461   char * new = strdup(print_state(state));
462
463   /* allowed transitions */
464   int allowed = GNUNET_NO;
465   switch (n->state) {
466   case S_NOT_CONNECTED:
467     if ((state == S_CONNECT_RECV) || (state == S_CONNECT_SENT) ||
468         (state == S_DISCONNECT))
469     {
470       allowed = GNUNET_YES;
471
472       /* Schedule reset task */
473       if ((state == S_CONNECT_RECV) || (state == S_CONNECT_SENT) )
474       {
475         GNUNET_assert (n->state_reset == GNUNET_SCHEDULER_NO_TASK);
476         n->state_reset = GNUNET_SCHEDULER_add_delayed (SETUP_CONNECTION_TIMEOUT, &reset_task, n);
477       }
478
479       break;
480     }
481     break;
482   case S_CONNECT_RECV:
483     if ((state == S_NOT_CONNECTED) || (state == S_DISCONNECT) ||
484         (state == S_CONNECTED) || /* FIXME SENT -> RECV ISSUE!*/ (state == S_CONNECT_SENT))
485     {
486       if ((state == S_CONNECTED) || (state == S_DISCONNECT) || (state == S_NOT_CONNECTED))
487       {
488 #if DEBUG_TRANSPORT
489         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
490             "Removed reset task for peer `%s' %s failed in state transition `%s' -> `%s' \n",
491             GNUNET_i2s (&n->id), GST_plugins_a2s(n->plugin_name, n->addr, n->addrlen), print_state(n->state), print_state(state));
492 #endif
493         GNUNET_assert (n->state_reset != GNUNET_SCHEDULER_NO_TASK);
494         GNUNET_SCHEDULER_cancel (n->state_reset);
495         n->state_reset = GNUNET_SCHEDULER_NO_TASK;
496       }
497
498       allowed = GNUNET_YES;
499       break;
500     }
501     break;
502   case S_CONNECT_SENT:
503     if ((state == S_NOT_CONNECTED) || (state == S_CONNECTED) ||
504         (state == S_DISCONNECT) || /* FIXME SENT -> RECV ISSUE!*/ (state == S_CONNECT_RECV))
505     {
506       if ((state == S_CONNECTED) || (state == S_DISCONNECT) || (state == S_NOT_CONNECTED))
507       {
508 #if DEBUG_TRANSPORT
509         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
510             "Removed reset task for peer `%s' %s failed in state transition `%s' -> `%s' \n",
511             GNUNET_i2s (&n->id), GST_plugins_a2s(n->plugin_name, n->addr, n->addrlen), print_state(n->state), print_state(state));
512 #endif
513         GNUNET_assert (n->state_reset != GNUNET_SCHEDULER_NO_TASK);
514         GNUNET_SCHEDULER_cancel (n->state_reset);
515         n->state_reset = GNUNET_SCHEDULER_NO_TASK;
516       }
517
518       allowed = GNUNET_YES;
519       break;
520     }
521     break;
522   case S_CONNECTED:
523     if (state == S_DISCONNECT)
524     {
525       allowed = GNUNET_YES;
526       break;
527     }
528     break;
529   case S_DISCONNECT:
530     /*
531     if (state == S_NOT_CONNECTED)
532     {
533       allowed = GNUNET_YES;
534       break;
535     }*/
536     break;
537   default:
538     GNUNET_break (0);
539     break;
540
541   }
542
543   if (allowed == GNUNET_NO)
544   {
545     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
546         "Illegal state transition from `%s' to `%s' in line %u \n",
547         old, new, line);
548     GNUNET_break (0);
549     GNUNET_free (old);
550     GNUNET_free (new);
551     return GNUNET_SYSERR;
552   }
553
554   n->state = state;
555 #if DEBUG_TRANSPORT
556   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "State for neighbour `%s' %X changed from `%s' to `%s' in line %u\n",
557       GNUNET_i2s (&n->id), n, old, new, line);
558 #endif
559   GNUNET_free (old);
560   GNUNET_free (new);
561   return GNUNET_OK;
562 }
563
564 static ssize_t
565 send_with_plugin ( const struct GNUNET_PeerIdentity * target,
566     const char *msgbuf,
567     size_t msgbuf_size,
568     uint32_t priority,
569     struct GNUNET_TIME_Relative timeout,
570     struct Session * session,
571     const char * plugin_name,
572     const void *addr,
573     size_t addrlen,
574     int force_address,
575     GNUNET_TRANSPORT_TransmitContinuation cont,
576     void *cont_cls)
577
578 {
579   struct GNUNET_TRANSPORT_PluginFunctions *papi;
580   size_t ret = GNUNET_SYSERR;
581
582   papi = GST_plugins_find (plugin_name);
583   if (papi == NULL)
584   {
585     if (cont != NULL)
586       cont (cont_cls, target, GNUNET_SYSERR);
587     return GNUNET_SYSERR;
588   }
589
590   ret = papi->send (papi->cls,
591       target,
592       msgbuf, msgbuf_size,
593       0,
594       timeout,
595       session,
596       addr, addrlen,
597       GNUNET_YES,
598       cont, cont_cls);
599
600   if (ret == -1)
601   {
602     if (cont != NULL)
603       cont (cont_cls, target, GNUNET_SYSERR);
604   }
605   return ret;
606 }
607
608 /**
609  * Task invoked to start a transmission to another peer.
610  *
611  * @param cls the 'struct NeighbourMapEntry'
612  * @param tc scheduler context
613  */
614 static void
615 transmission_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
616
617
618 /**
619  * We're done with our transmission attempt, continue processing.
620  *
621  * @param cls the 'struct MessageQueue' of the message
622  * @param receiver intended receiver
623  * @param success whether it worked or not
624  */
625 static void
626 transmit_send_continuation (void *cls,
627                             const struct GNUNET_PeerIdentity *receiver,
628                             int success)
629 {
630   struct MessageQueue *mq;
631   struct NeighbourMapEntry *n;
632
633   mq = cls;
634   n = mq->n;
635   if (NULL != n)
636   {
637     GNUNET_assert (n->is_active == mq);
638     n->is_active = NULL;
639     if (success == GNUNET_YES)
640     {
641       GNUNET_assert (n->transmission_task == GNUNET_SCHEDULER_NO_TASK);
642       n->transmission_task = GNUNET_SCHEDULER_add_now (&transmission_task, n);
643     }
644   }
645 #if DEBUG_TRANSPORT
646   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending message of type %u was %s\n",
647               ntohs (((struct GNUNET_MessageHeader *) mq->message_buf)->type),
648               (success == GNUNET_OK) ? "successful" : "FAILED");
649 #endif
650   if (NULL != mq->cont)
651     mq->cont (mq->cont_cls, success);
652   GNUNET_free (mq);
653 }
654
655
656 /**
657  * Check the ready list for the given neighbour and if a plugin is
658  * ready for transmission (and if we have a message), do so!
659  *
660  * @param n target peer for which to transmit
661  */
662 static void
663 try_transmission_to_peer (struct NeighbourMapEntry *n)
664 {
665   struct MessageQueue *mq;
666   struct GNUNET_TIME_Relative timeout;
667   ssize_t ret;
668
669   if (n->is_active != NULL)
670   {
671     GNUNET_break (0);
672     return;                     /* transmission already pending */
673   }
674   if (n->transmission_task != GNUNET_SCHEDULER_NO_TASK)
675   {
676     GNUNET_break (0);
677     return;                     /* currently waiting for bandwidth */
678   }
679   while (NULL != (mq = n->messages_head))
680   {
681     timeout = GNUNET_TIME_absolute_get_remaining (mq->timeout);
682     if (timeout.rel_value > 0)
683       break;
684     GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
685     n->is_active = mq;
686     mq->n = n;
687     transmit_send_continuation (mq, &n->id, GNUNET_SYSERR);     /* timeout */
688   }
689   if (NULL == mq)
690     return;                     /* no more messages */
691
692   if (GST_plugins_find (n->plugin_name) == NULL)
693   {
694     GNUNET_break (0);
695     return;
696   }
697   GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
698   n->is_active = mq;
699   mq->n = n;
700
701   if  ((n->session == NULL) && (n->addr == NULL) && (n->addrlen == 0))
702   {
703     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No address for peer `%s'\n",
704                 GNUNET_i2s (&n->id));
705     transmit_send_continuation (mq, &n->id, GNUNET_SYSERR);
706     GNUNET_assert (n->transmission_task == GNUNET_SCHEDULER_NO_TASK);
707     n->transmission_task = GNUNET_SCHEDULER_add_now (&transmission_task, n);
708     return;
709   }
710
711   ret = send_with_plugin (&n->id,
712                           mq->message_buf, mq->message_buf_size, 0,
713                           timeout,
714                           n->session, n->plugin_name, n->addr, n->addrlen,
715                           GNUNET_YES,
716                           &transmit_send_continuation, mq);
717   if (ret == -1)
718   {
719     /* failure, but 'send' would not call continuation in this case,
720      * so we need to do it here! */
721     transmit_send_continuation (mq, &n->id, GNUNET_SYSERR);
722   }
723
724 }
725
726
727 /**
728  * Task invoked to start a transmission to another peer.
729  *
730  * @param cls the 'struct NeighbourMapEntry'
731  * @param tc scheduler context
732  */
733 static void
734 transmission_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
735 {
736   struct NeighbourMapEntry *n = cls;
737   GNUNET_assert (NULL != lookup_neighbour(&n->id));
738   n->transmission_task = GNUNET_SCHEDULER_NO_TASK;
739   try_transmission_to_peer (n);
740 }
741
742
743 /**
744  * Initialize the neighbours subsystem.
745  *
746  * @param cls closure for callbacks
747  * @param connect_cb function to call if we connect to a peer
748  * @param disconnect_cb function to call if we disconnect from a peer
749  */
750 void
751 GST_neighbours_start (void *cls, GNUNET_TRANSPORT_NotifyConnect connect_cb,
752                       GNUNET_TRANSPORT_NotifyDisconnect disconnect_cb)
753 {
754   callback_cls = cls;
755   connect_notify_cb = connect_cb;
756   disconnect_notify_cb = disconnect_cb;
757   neighbours = GNUNET_CONTAINER_multihashmap_create (NEIGHBOUR_TABLE_SIZE);
758 }
759
760
761 static void
762 send_disconnect_cont (void *cls,
763     const struct GNUNET_PeerIdentity * target,
764     int result)
765 {
766 #if DEBUG_TRANSPORT
767   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending DISCONNECT message to peer `%4s': %i\n",
768               GNUNET_i2s (target), result);
769 #endif
770 }
771
772
773 static int
774 send_disconnect (const struct GNUNET_PeerIdentity * target,
775                  const char *plugin_name,
776                  const char *sender_address, uint16_t sender_address_len,
777                  struct Session *session)
778 {
779   size_t ret;
780   struct SessionDisconnectMessage disconnect_msg;
781
782 #if DEBUG_TRANSPORT
783   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending DISCONNECT message to peer `%4s'\n",
784               GNUNET_i2s (target));
785 #endif
786
787   disconnect_msg.header.size = htons (sizeof (struct SessionDisconnectMessage));
788   disconnect_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
789   disconnect_msg.reserved = htonl (0);
790   disconnect_msg.purpose.size = htonl (sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) +
791                                        sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) +
792                                        sizeof (struct GNUNET_TIME_AbsoluteNBO) );
793   disconnect_msg.purpose.purpose = htonl (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT);
794   disconnect_msg.timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
795   disconnect_msg.public_key = GST_my_public_key;
796   GNUNET_assert (GNUNET_OK ==
797                  GNUNET_CRYPTO_rsa_sign (GST_my_private_key,
798                                          &disconnect_msg.purpose,
799                                          &disconnect_msg.signature));
800
801   ret = send_with_plugin(target,
802                          (const char *) &disconnect_msg, sizeof (disconnect_msg),
803                          UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
804                          session, plugin_name, sender_address,  sender_address_len,
805                          GNUNET_YES, &send_disconnect_cont, NULL);
806
807   if (ret == GNUNET_SYSERR)
808     return GNUNET_SYSERR;
809
810   GNUNET_STATISTICS_update (GST_stats,
811                             gettext_noop ("# peers disconnected due to external request"), 1,
812                             GNUNET_NO);
813   return GNUNET_OK;
814 }
815
816 /**
817  * Disconnect from the given neighbour, clean up the record.
818  *
819  * @param n neighbour to disconnect from
820  */
821 static void
822 disconnect_neighbour (struct NeighbourMapEntry *n)
823 {
824   struct MessageQueue *mq;
825   int was_connected = is_connected(n);
826
827   /* send DISCONNECT MESSAGE */
828   if (is_connected(n) || is_connecting(n))
829   {
830     if (GNUNET_OK == send_disconnect(&n->id, n->plugin_name, n->addr, n->addrlen, n->session))
831       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sent DISCONNECT_MSG to `%s'\n",
832                   GNUNET_i2s (&n->id));
833     else
834       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Could not send DISCONNECT_MSG to `%s'\n",
835                   GNUNET_i2s (&n->id));
836   }
837
838
839   if (is_disconnecting(n))
840     return;
841   change_state (n, S_DISCONNECT);
842
843   while (NULL != (mq = n->messages_head))
844   {
845     GNUNET_CONTAINER_DLL_remove (n->messages_head, n->messages_tail, mq);
846     if (NULL != mq->cont)
847       mq->cont (mq->cont_cls, GNUNET_SYSERR);
848     GNUNET_free (mq);
849   }
850   if (NULL != n->is_active)
851   {
852     n->is_active->n = NULL;
853     n->is_active = NULL;
854   }
855   if (was_connected)
856   {
857     GNUNET_assert (GNUNET_SCHEDULER_NO_TASK != n->keepalive_task);
858     GNUNET_SCHEDULER_cancel (n->keepalive_task);
859     n->keepalive_task = GNUNET_SCHEDULER_NO_TASK;  
860     GNUNET_assert (neighbours_connected > 0);
861     neighbours_connected--;
862     GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# peers connected"), -1,
863                               GNUNET_NO);
864     disconnect_notify_cb (callback_cls, &n->id);
865   }
866   GNUNET_assert (GNUNET_YES ==
867                  GNUNET_CONTAINER_multihashmap_remove (neighbours,
868                                                        &n->id.hashPubKey, n));
869   if (GNUNET_SCHEDULER_NO_TASK != n->ats_suggest)
870   {
871     GNUNET_SCHEDULER_cancel (n->ats_suggest);
872     n->ats_suggest = GNUNET_SCHEDULER_NO_TASK;
873   }
874   if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
875   {
876     GNUNET_SCHEDULER_cancel (n->timeout_task);
877     n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
878   }
879   if (GNUNET_SCHEDULER_NO_TASK != n->transmission_task)
880   {
881     GNUNET_SCHEDULER_cancel (n->transmission_task);
882     n->transmission_task = GNUNET_SCHEDULER_NO_TASK;
883   }
884   if (NULL != n->plugin_name)
885   {
886     GNUNET_free (n->plugin_name);
887     n->plugin_name = NULL;
888   }
889   if (NULL != n->addr)
890   {
891     GNUNET_free (n->addr);
892     n->addr = NULL;
893     n->addrlen = 0;
894   }
895   n->session = NULL;
896   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Deleting peer `%4s', %X\n",
897               GNUNET_i2s (&n->id), n);
898   GNUNET_free (n);
899 }
900
901
902 /**
903  * Peer has been idle for too long. Disconnect.
904  *
905  * @param cls the 'struct NeighbourMapEntry' of the neighbour that went idle
906  * @param tc scheduler context
907  */
908 static void
909 neighbour_timeout_task (void *cls,
910                         const struct GNUNET_SCHEDULER_TaskContext *tc)
911 {
912   struct NeighbourMapEntry *n = cls;
913
914   n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
915
916   GNUNET_STATISTICS_update (GST_stats,
917                             gettext_noop ("# peers disconnected due to timeout"), 1,
918                             GNUNET_NO);
919   disconnect_neighbour (n);
920 }
921
922
923 /**
924  * Send another keepalive message.
925  *
926  * @param cls the 'struct NeighbourMapEntry' of the neighbour that went idle
927  * @param tc scheduler context
928  */
929 static void
930 neighbour_keepalive_task (void *cls,
931                           const struct GNUNET_SCHEDULER_TaskContext *tc)
932 {
933   struct NeighbourMapEntry *n = cls;
934   struct GNUNET_MessageHeader m;
935
936   n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
937                                                     &neighbour_keepalive_task,
938                                                     n);
939   GNUNET_assert (is_connected(n));
940   GNUNET_STATISTICS_update (GST_stats,
941                             gettext_noop ("# keepalives sent"), 1,
942                             GNUNET_NO);
943   m.size = htons (sizeof (struct GNUNET_MessageHeader));
944   m.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_KEEPALIVE);
945
946   send_with_plugin(&n->id, (const void *) &m,
947                    sizeof (m),
948                    UINT32_MAX /* priority */ ,
949                    GNUNET_TIME_UNIT_FOREVER_REL,
950                    n->session, n->plugin_name, n->addr, n->addrlen,
951                    GNUNET_YES, NULL, NULL);
952 }
953
954
955 /**
956  * Disconnect from the given neighbour.
957  *
958  * @param cls unused
959  * @param key hash of neighbour's public key (not used)
960  * @param value the 'struct NeighbourMapEntry' of the neighbour
961  */
962 static int
963 disconnect_all_neighbours (void *cls, const GNUNET_HashCode * key, void *value)
964 {
965   struct NeighbourMapEntry *n = value;
966
967 #if DEBUG_TRANSPORT
968   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting peer `%4s', %s\n",
969               GNUNET_i2s (&n->id), "SHUTDOWN_TASK");
970 #endif
971   if (is_connected(n))
972     GNUNET_STATISTICS_update (GST_stats,
973                               gettext_noop ("# peers disconnected due to global disconnect"), 1,
974                               GNUNET_NO);
975   disconnect_neighbour (n);
976   return GNUNET_OK;
977 }
978
979
980 static void
981 ats_suggest_cancel (void *cls,
982     const struct GNUNET_SCHEDULER_TaskContext *tc)
983 {
984   struct NeighbourMapEntry *n = cls;
985
986   n->ats_suggest = GNUNET_SCHEDULER_NO_TASK;
987
988   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
989               " ATS did not suggested address to connect to peer `%s'\n",
990               GNUNET_i2s (&n->id));
991
992   disconnect_neighbour(n);
993 }
994
995
996 /**
997  * Cleanup the neighbours subsystem.
998  */
999 void
1000 GST_neighbours_stop ()
1001 {
1002   GNUNET_assert (neighbours != NULL);
1003
1004   GNUNET_CONTAINER_multihashmap_iterate (neighbours, &disconnect_all_neighbours,
1005                                          NULL);
1006   GNUNET_CONTAINER_multihashmap_destroy (neighbours);
1007   GNUNET_assert (neighbours_connected == 0);
1008   neighbours = NULL;
1009   callback_cls = NULL;
1010   connect_notify_cb = NULL;
1011   disconnect_notify_cb = NULL;
1012 }
1013
1014
1015 /**
1016  * We tried to send a SESSION_CONNECT message to another peer.  If this
1017  * succeeded, we change the state.  If it failed, we should tell
1018  * ATS to not use this address anymore (until it is re-validated).
1019  *
1020  * @param cls the 'struct NeighbourMapEntry'
1021  * @param success GNUNET_OK on success
1022  */
1023 static void
1024 send_connect_continuation (void *cls,
1025       const struct GNUNET_PeerIdentity * target,
1026       int success)
1027
1028 {
1029   struct NeighbourMapEntry *n = cls;
1030
1031   GNUNET_assert (n != NULL);
1032   GNUNET_assert (!is_connected(n));
1033
1034   if (is_disconnecting(n))
1035     return; /* neighbour is going away */
1036   if (GNUNET_YES != success)
1037   {
1038 #if DEBUG_TRANSPORT
1039     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1040               "Failed to send CONNECT_MSG to peer `%4s' with plugin `%s' address '%s' session %X, asking ATS for new address \n",
1041               GNUNET_i2s (&n->id), n->plugin_name,
1042               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1043                                                                   n->addr,
1044                                                                   n->addrlen),
1045               n->session);
1046 #endif
1047
1048     GNUNET_ATS_address_destroyed (GST_ats,
1049                                   &n->id,
1050                                   n->plugin_name, 
1051                                   n->addr,
1052                                   n->addrlen,
1053                                   NULL);
1054
1055     if (n->ats_suggest!= GNUNET_SCHEDULER_NO_TASK)
1056       GNUNET_SCHEDULER_cancel(n->ats_suggest);
1057     n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
1058     GNUNET_ATS_suggest_address(GST_ats, &n->id);
1059     return;
1060   }
1061   change_state(n, S_CONNECT_SENT);
1062 }
1063
1064
1065 /**
1066  * We tried to switch addresses with an peer already connected. If it failed,
1067  * we should tell ATS to not use this address anymore (until it is re-validated).
1068  *
1069  * @param cls the 'struct NeighbourMapEntry'
1070  * @param success GNUNET_OK on success
1071  */
1072 static void
1073 send_switch_address_continuation (void *cls,
1074       const struct GNUNET_PeerIdentity * target,
1075       int success)
1076
1077 {
1078   struct NeighbourMapEntry *n = cls;
1079
1080   GNUNET_assert (n != NULL);
1081   if (is_disconnecting(n))
1082     return; /* neighbour is going away */
1083
1084   GNUNET_assert (n->state == S_CONNECTED);
1085   if (GNUNET_YES != success)
1086   {
1087 #if DEBUG_TRANSPORT
1088     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1089               "Failed to switch connected peer `%s' to plugin `%s' address '%s' session %X, asking ATS for new address \n",
1090               GNUNET_i2s (&n->id), n->plugin_name,
1091               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1092                                                                   n->addr,
1093                                                                   n->addrlen),
1094               n->session);
1095 #endif
1096
1097     GNUNET_ATS_address_destroyed (GST_ats,
1098                                   &n->id,
1099                                   n->plugin_name,
1100                                   n->addr,
1101                                   n->addrlen,
1102                                   NULL);
1103
1104     if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1105       GNUNET_SCHEDULER_cancel(n->ats_suggest);
1106     n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
1107     GNUNET_ATS_suggest_address(GST_ats, &n->id);
1108     return;
1109   }
1110 }
1111
1112 /**
1113  * We tried to send a SESSION_CONNECT message to another peer.  If this
1114  * succeeded, we change the state.  If it failed, we should tell
1115  * ATS to not use this address anymore (until it is re-validated).
1116  *
1117  * @param cls the 'struct NeighbourMapEntry'
1118  * @param success GNUNET_OK on success
1119  */
1120 static void
1121 send_connect_ack_continuation (void *cls,
1122       const struct GNUNET_PeerIdentity * target,
1123       int success)
1124
1125 {
1126   struct NeighbourMapEntry *n = cls;
1127
1128   GNUNET_assert (n != NULL);
1129
1130   if (GNUNET_YES == success)
1131     return; /* sending successful */
1132
1133   /* sending failed, ask for next address  */
1134 #if DEBUG_TRANSPORT
1135     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1136               "Failed to send CONNECT_MSG to peer `%4s' with plugin `%s' address '%s' session %X, asking ATS for new address \n",
1137               GNUNET_i2s (&n->id), n->plugin_name,
1138               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1139                                                                   n->addr,
1140                                                                   n->addrlen),
1141               n->session);
1142 #endif
1143     change_state(n, S_NOT_CONNECTED);
1144
1145     GNUNET_ATS_address_destroyed (GST_ats,
1146                                   &n->id,
1147                                   n->plugin_name,
1148                                   n->addr,
1149                                   n->addrlen,
1150                                   NULL);
1151
1152     if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1153       GNUNET_SCHEDULER_cancel(n->ats_suggest);
1154     n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
1155     GNUNET_ATS_suggest_address(GST_ats, &n->id);
1156 }
1157
1158 /**
1159  * For an existing neighbour record, set the active connection to
1160  * the given address.
1161  *
1162  * @param peer identity of the peer to switch the address for
1163  * @param plugin_name name of transport that delivered the PONG
1164  * @param address address of the other peer, NULL if other peer
1165  *                       connected to us
1166  * @param address_len number of bytes in address
1167  * @param session session to use (or NULL)
1168  * @param ats performance data
1169  * @param ats_count number of entries in ats
1170  * @return GNUNET_YES if we are currently connected, GNUNET_NO if the
1171  *         connection is not up (yet)
1172  */
1173 int
1174 GST_neighbours_switch_to_address_3way (const struct GNUNET_PeerIdentity *peer,
1175                                   const char *plugin_name, const void *address,
1176                                   size_t address_len, struct Session *session,
1177                                   const struct GNUNET_ATS_Information
1178                                   *ats, uint32_t ats_count,
1179                                   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
1180                                   struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out)
1181 {
1182   struct NeighbourMapEntry *n;
1183   struct SessionConnectMessage connect_msg;
1184   size_t msg_len;
1185   size_t ret;
1186
1187   GNUNET_assert (neighbours != NULL);
1188   n = lookup_neighbour (peer);
1189   if (NULL == n)
1190   {
1191     if (NULL == session)
1192       GNUNET_ATS_address_destroyed (GST_ats,
1193                                     peer,
1194                                     plugin_name, address,
1195                                     address_len, NULL);    
1196     return GNUNET_NO;
1197   }
1198
1199   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1200   {
1201     GNUNET_SCHEDULER_cancel(n->ats_suggest);
1202     n->ats_suggest = GNUNET_SCHEDULER_NO_TASK;
1203   }
1204
1205 #if DEBUG_TRANSPORT
1206   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1207               "ATS tells us to switch to plugin `%s' address '%s' session %X for %s peer `%s'\n",
1208               plugin_name,
1209               (address_len == 0) ? "<inbound>" : GST_plugins_a2s (plugin_name,
1210                                                                   address,
1211                                                                   address_len),
1212               session, (is_connected(n) ? "CONNECTED" : "NOT CONNECTED"),
1213               GNUNET_i2s (peer));
1214 #endif
1215
1216   // do not switch addresses just update quotas
1217   if (n != NULL)
1218   {
1219     if ((is_connected(n)) && (address_len == n->addrlen))
1220     {
1221       if ((0 == memcmp (address, n->addr, address_len)) &&
1222           (n->session == session))
1223       {
1224         struct QuotaSetMessage q_msg;
1225
1226 #if DEBUG_TRANSPORT
1227   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1228               "Sending outbound quota of %u Bps and inbound quota of % Bps for peer `%s' to all clients\n",
1229               ntohl (n->bandwidth_out.value__),
1230               ntohl (n->bandwidth_in.value__),
1231               GNUNET_i2s (peer));
1232 #endif
1233
1234         n->bandwidth_in = bandwidth_in;
1235         n->bandwidth_out = bandwidth_out;
1236         GST_neighbours_set_incoming_quota(&n->id, n->bandwidth_in);
1237
1238         q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
1239         q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
1240         q_msg.quota = n->bandwidth_out;
1241         q_msg.peer = (*peer);
1242         GST_clients_broadcast (&q_msg.header, GNUNET_NO);
1243         return GNUNET_NO;
1244       }
1245     }
1246   }
1247
1248   GNUNET_free_non_null (n->addr);
1249   n->addr = GNUNET_malloc (address_len);
1250   memcpy (n->addr, address, address_len);
1251   n->bandwidth_in = bandwidth_in;
1252   n->bandwidth_out = bandwidth_out;
1253   n->addrlen = address_len;
1254   n->session = session;
1255   GNUNET_free_non_null (n->plugin_name);
1256   n->plugin_name = GNUNET_strdup (plugin_name);
1257   GNUNET_SCHEDULER_cancel (n->timeout_task);
1258   n->timeout_task =
1259       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1260                                     &neighbour_timeout_task, n);
1261
1262   if (n->state == S_DISCONNECT)
1263   {
1264     /* We are disconnecting, nothing to do here */
1265     return GNUNET_NO;
1266   }
1267   /* We are not connected/connecting and initiate a fresh connect */
1268   if (n->state == S_NOT_CONNECTED)
1269   {
1270     msg_len = sizeof (struct SessionConnectMessage);
1271     connect_msg.header.size = htons (msg_len);
1272     connect_msg.header.type =
1273         htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT);
1274     connect_msg.reserved = htonl (0);
1275     connect_msg.timestamp =
1276         GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1277
1278     ret = send_with_plugin (peer, (const char *) &connect_msg, msg_len, UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
1279                             session, plugin_name, address, address_len,
1280                             GNUNET_YES, &send_connect_continuation, n);
1281
1282     return GNUNET_NO;
1283   }
1284   /* We received a CONNECT message and asked ATS for an address */
1285   else if (n->state == S_CONNECT_RECV)
1286   {
1287     msg_len = sizeof (struct SessionConnectMessage);
1288     connect_msg.header.size = htons (msg_len);
1289     connect_msg.header.type =
1290       htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT_ACK);
1291     connect_msg.reserved = htonl (0);
1292     connect_msg.timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1293
1294     ret = send_with_plugin(&n->id, (const void *) &connect_msg, msg_len, UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
1295                            session, plugin_name, address, address_len,
1296                            GNUNET_YES, &send_connect_ack_continuation, n);
1297     return GNUNET_NO;
1298   }
1299   /* connected peer is switching addresses */
1300   else if (n->state == S_CONNECTED)
1301   {
1302     msg_len = sizeof (struct SessionConnectMessage);
1303     connect_msg.header.size = htons (msg_len);
1304     connect_msg.header.type =
1305         htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT);
1306     connect_msg.reserved = htonl (0);
1307     connect_msg.timestamp =
1308         GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1309
1310     ret = send_with_plugin (peer, (const char *) &connect_msg, msg_len, UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_REL,
1311                             session, plugin_name, address, address_len,
1312                             GNUNET_YES, &send_switch_address_continuation, n);
1313     if (ret == GNUNET_SYSERR)
1314     {
1315       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1316                 "Failed to send CONNECT_MESSAGE to `%4s' using plugin `%s' address '%s' session %X\n",
1317                 GNUNET_i2s (peer), plugin_name,
1318                 (address_len == 0) ? "<inbound>" : GST_plugins_a2s (plugin_name,
1319                                                                     address,
1320                                                                     address_len),
1321                 session);
1322     }
1323     return GNUNET_NO;
1324   }
1325   else if (n->state == S_CONNECT_SENT)
1326   {
1327       //FIXME
1328      return GNUNET_NO;
1329   }
1330   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Invalid connection state to switch addresses %u \n", n->state);
1331   GNUNET_break_op (0);
1332   return GNUNET_NO;
1333 }
1334
1335
1336 /**
1337  * Create an entry in the neighbour map for the given peer
1338  * 
1339  * @param peer peer to create an entry for
1340  * @return new neighbour map entry
1341  */
1342 static struct NeighbourMapEntry *
1343 setup_neighbour (const struct GNUNET_PeerIdentity *peer)
1344 {
1345   struct NeighbourMapEntry *n;
1346
1347 #if DEBUG_TRANSPORT
1348   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1349               "Unknown peer `%s', creating new neighbour\n",
1350               GNUNET_i2s (peer));
1351 #endif
1352   n = GNUNET_malloc (sizeof (struct NeighbourMapEntry));
1353   n->id = *peer;
1354   n->state = S_NOT_CONNECTED;
1355   GNUNET_BANDWIDTH_tracker_init (&n->in_tracker,
1356                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
1357                                  MAX_BANDWIDTH_CARRY_S);
1358   n->timeout_task =
1359     GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1360                                   &neighbour_timeout_task, n);
1361   GNUNET_assert (GNUNET_OK ==
1362                  GNUNET_CONTAINER_multihashmap_put (neighbours,
1363                                                     &n->id.hashPubKey, n,
1364                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1365   return n;
1366 }
1367
1368
1369 /**
1370  * Try to create a connection to the given target (eventually).
1371  *
1372  * @param target peer to try to connect to
1373  */
1374 void
1375 GST_neighbours_try_connect (const struct GNUNET_PeerIdentity *target)
1376 {
1377   struct NeighbourMapEntry *n;
1378
1379   GNUNET_assert (neighbours != NULL);
1380 #if DEBUG_TRANSPORT
1381   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Trying to connect to peer `%s'\n",
1382               GNUNET_i2s (target));
1383 #endif
1384   GNUNET_assert (0 !=
1385                  memcmp (target, &GST_my_identity,
1386                          sizeof (struct GNUNET_PeerIdentity)));
1387   n = lookup_neighbour (target);
1388
1389   if (NULL != n)
1390   {
1391     if ((is_connected(n)) || (is_connecting(n)))
1392       return;                     /* already connecting or connected */
1393     if (is_disconnecting(n))
1394       change_state (n, S_NOT_CONNECTED);
1395   }
1396
1397
1398   if (n == NULL)
1399     n = setup_neighbour (target);
1400 #if DEBUG_TRANSPORT
1401   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1402               "Asking ATS for suggested address to connect to peer `%s'\n",
1403               GNUNET_i2s (&n->id));
1404 #endif
1405
1406    GNUNET_ATS_suggest_address (GST_ats, &n->id);
1407 }
1408
1409 /**
1410  * Test if we're connected to the given peer.
1411  *
1412  * @param target peer to test
1413  * @return GNUNET_YES if we are connected, GNUNET_NO if not
1414  */
1415 int
1416 GST_neighbours_test_connected (const struct GNUNET_PeerIdentity *target)
1417 {
1418   struct NeighbourMapEntry *n;
1419
1420   GNUNET_assert (neighbours != NULL);
1421
1422   n = lookup_neighbour (target);
1423
1424   if ((NULL == n) || (!is_connected(n)))
1425     return GNUNET_NO;           /* not connected */
1426   return GNUNET_YES;
1427 }
1428
1429
1430 /**
1431  * A session was terminated. Take note.
1432  *
1433  * @param peer identity of the peer where the session died
1434  * @param session session that is gone
1435  */
1436 void
1437 GST_neighbours_session_terminated (const struct GNUNET_PeerIdentity *peer,
1438                                    struct Session *session)
1439 {
1440   struct NeighbourMapEntry *n;
1441
1442   GNUNET_assert (neighbours != NULL);
1443
1444 #if DEBUG_TRANSPORT
1445   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1446               "Session %X to peer `%s' ended \n",
1447               session, GNUNET_i2s (peer));
1448 #endif
1449
1450   n = lookup_neighbour (peer);
1451   if (NULL == n)
1452     return;
1453   if (session != n->session)
1454     return;                     /* doesn't affect us */
1455
1456   n->session = NULL;
1457   GNUNET_free (n->addr);
1458   n->addr = NULL;
1459   n->addrlen = 0;
1460
1461   /* not connected anymore anyway, shouldn't matter */
1462   if ((!is_connected(n)) && (!is_connecting(n)))
1463     return;
1464
1465   /* We are connected, so ask ATS to switch addresses */
1466   GNUNET_SCHEDULER_cancel (n->timeout_task);
1467   n->timeout_task =
1468       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_DISCONNECT_SESSION_TIMEOUT,
1469                                     &neighbour_timeout_task, n);
1470   /* try QUICKLY to re-establish a connection, reduce timeout! */
1471   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1472     GNUNET_SCHEDULER_cancel(n->ats_suggest);
1473   n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
1474   GNUNET_ATS_suggest_address (GST_ats, peer);
1475 }
1476
1477
1478 /**
1479  * Transmit a message to the given target using the active connection.
1480  *
1481  * @param target destination
1482  * @param msg message to send
1483  * @param msg_size number of bytes in msg
1484  * @param timeout when to fail with timeout
1485  * @param cont function to call when done
1486  * @param cont_cls closure for 'cont'
1487  */
1488 void
1489 GST_neighbours_send (const struct GNUNET_PeerIdentity *target, const void *msg,
1490                      size_t msg_size, struct GNUNET_TIME_Relative timeout,
1491                      GST_NeighbourSendContinuation cont, void *cont_cls)
1492 {
1493   struct NeighbourMapEntry *n;
1494   struct MessageQueue *mq;
1495
1496   GNUNET_assert (neighbours != NULL);
1497
1498   n = lookup_neighbour (target);
1499   if ((n == NULL) || (!is_connected(n)))
1500   {
1501     GNUNET_STATISTICS_update (GST_stats,
1502                               gettext_noop
1503                               ("# messages not sent (no such peer or not connected)"),
1504                               1, GNUNET_NO);
1505 #if DEBUG_TRANSPORT
1506     if (n == NULL)
1507       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1508                   "Could not send message to peer `%s': unknown neighbour",
1509                   GNUNET_i2s (target));
1510     else if (!is_connected(n))
1511       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1512                   "Could not send message to peer `%s': not connected\n",
1513                   GNUNET_i2s (target));
1514 #endif
1515     if (NULL != cont)
1516       cont (cont_cls, GNUNET_SYSERR);
1517     return;
1518   }
1519
1520   if ((n->session == NULL) && (n->addr == NULL) && (n->addrlen ==0))
1521   {
1522     GNUNET_STATISTICS_update (GST_stats,
1523                               gettext_noop
1524                               ("# messages not sent (no such peer or not connected)"),
1525                               1, GNUNET_NO);
1526 #if DEBUG_TRANSPORT
1527       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1528                   "Could not send message to peer `%s': no address available\n",
1529                   GNUNET_i2s (target));
1530 #endif
1531
1532     if (NULL != cont)
1533       cont (cont_cls, GNUNET_SYSERR);
1534     return;
1535   }
1536
1537   GNUNET_assert (msg_size >= sizeof (struct GNUNET_MessageHeader));
1538   GNUNET_STATISTICS_update (GST_stats,
1539                             gettext_noop
1540                             ("# bytes in message queue for other peers"),
1541                             msg_size, GNUNET_NO);
1542   mq = GNUNET_malloc (sizeof (struct MessageQueue) + msg_size);
1543   mq->cont = cont;
1544   mq->cont_cls = cont_cls;
1545   /* FIXME: this memcpy can be up to 7% of our total runtime! */
1546   memcpy (&mq[1], msg, msg_size);
1547   mq->message_buf = (const char *) &mq[1];
1548   mq->message_buf_size = msg_size;
1549   mq->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1550   GNUNET_CONTAINER_DLL_insert_tail (n->messages_head, n->messages_tail, mq);
1551
1552   if ((GNUNET_SCHEDULER_NO_TASK == n->transmission_task) &&
1553       (NULL == n->is_active))
1554     n->transmission_task = GNUNET_SCHEDULER_add_now (&transmission_task, n);
1555 }
1556
1557
1558 /**
1559  * We have received a message from the given sender.  How long should
1560  * we delay before receiving more?  (Also used to keep the peer marked
1561  * as live).
1562  *
1563  * @param sender sender of the message
1564  * @param size size of the message
1565  * @param do_forward set to GNUNET_YES if the message should be forwarded to clients
1566  *                   GNUNET_NO if the neighbour is not connected or violates the quota,
1567  *                   GNUNET_SYSERR if the connection is not fully up yet
1568  * @return how long to wait before reading more from this sender
1569  */
1570 struct GNUNET_TIME_Relative
1571 GST_neighbours_calculate_receive_delay (const struct GNUNET_PeerIdentity
1572                                         *sender, ssize_t size, int *do_forward)
1573 {
1574   struct NeighbourMapEntry *n;
1575   struct GNUNET_TIME_Relative ret;
1576
1577   GNUNET_assert (neighbours != NULL);
1578
1579   n = lookup_neighbour (sender);
1580   if (n == NULL)
1581   {
1582     GST_neighbours_try_connect (sender);
1583     n = lookup_neighbour (sender);
1584     if (NULL == n)
1585     {
1586       GNUNET_STATISTICS_update (GST_stats,
1587                                 gettext_noop
1588                                 ("# messages discarded due to lack of neighbour record"),
1589                                 1, GNUNET_NO);
1590       *do_forward = GNUNET_NO;
1591       return GNUNET_TIME_UNIT_ZERO;
1592     }
1593   }
1594   if (!is_connected(n))
1595   {
1596     *do_forward = GNUNET_SYSERR;
1597     return GNUNET_TIME_UNIT_ZERO;
1598   }
1599   if (GNUNET_YES == GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, size))
1600   {
1601     n->quota_violation_count++;
1602 #if DEBUG_TRANSPORT
1603     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1604                 "Bandwidth quota (%u b/s) violation detected (total of %u).\n",
1605                 n->in_tracker.available_bytes_per_s__,
1606                 n->quota_violation_count);
1607 #endif
1608     /* Discount 32k per violation */
1609     GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, -32 * 1024);
1610   }
1611   else
1612   {
1613     if (n->quota_violation_count > 0)
1614     {
1615       /* try to add 32k back */
1616       GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, 32 * 1024);
1617       n->quota_violation_count--;
1618     }
1619   }
1620   if (n->quota_violation_count > QUOTA_VIOLATION_DROP_THRESHOLD)
1621   {
1622     GNUNET_STATISTICS_update (GST_stats,
1623                               gettext_noop
1624                               ("# bandwidth quota violations by other peers"),
1625                               1, GNUNET_NO);
1626     *do_forward = GNUNET_NO;
1627     return GNUNET_CONSTANTS_QUOTA_VIOLATION_TIMEOUT;
1628   }
1629   *do_forward = GNUNET_YES;
1630   ret = GNUNET_BANDWIDTH_tracker_get_delay (&n->in_tracker, 32 * 1024);
1631   if (ret.rel_value > 0)
1632   {
1633 #if DEBUG_TRANSPORT
1634     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1635                 "Throttling read (%llu bytes excess at %u b/s), waiting %llu ms before reading more.\n",
1636                 (unsigned long long) n->in_tracker.
1637                 consumption_since_last_update__,
1638                 (unsigned int) n->in_tracker.available_bytes_per_s__,
1639                 (unsigned long long) ret.rel_value);
1640 #endif
1641     GNUNET_STATISTICS_update (GST_stats,
1642                               gettext_noop ("# ms throttling suggested"),
1643                               (int64_t) ret.rel_value, GNUNET_NO);
1644   }
1645   return ret;
1646 }
1647
1648
1649 /**
1650  * Keep the connection to the given neighbour alive longer,
1651  * we received a KEEPALIVE (or equivalent).
1652  *
1653  * @param neighbour neighbour to keep alive
1654  */
1655 void
1656 GST_neighbours_keepalive (const struct GNUNET_PeerIdentity *neighbour)
1657 {
1658   struct NeighbourMapEntry *n;
1659
1660   GNUNET_assert (neighbours != NULL);
1661
1662   n = lookup_neighbour (neighbour);
1663   if (NULL == n)
1664   {
1665     GNUNET_STATISTICS_update (GST_stats,
1666                               gettext_noop
1667                               ("# KEEPALIVE messages discarded (not connected)"),
1668                               1, GNUNET_NO);
1669     return;
1670   }
1671   GNUNET_SCHEDULER_cancel (n->timeout_task);
1672   n->timeout_task =
1673       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1674                                     &neighbour_timeout_task, n);
1675 }
1676
1677
1678 /**
1679  * Change the incoming quota for the given peer.
1680  *
1681  * @param neighbour identity of peer to change qutoa for
1682  * @param quota new quota
1683  */
1684 void
1685 GST_neighbours_set_incoming_quota (const struct GNUNET_PeerIdentity *neighbour,
1686                                    struct GNUNET_BANDWIDTH_Value32NBO quota)
1687 {
1688   struct NeighbourMapEntry *n;
1689
1690   GNUNET_assert (neighbours != NULL);
1691
1692   n = lookup_neighbour (neighbour);
1693   if (n == NULL)
1694   {
1695     GNUNET_STATISTICS_update (GST_stats,
1696                               gettext_noop
1697                               ("# SET QUOTA messages ignored (no such peer)"),
1698                               1, GNUNET_NO);
1699     return;
1700   }
1701   GNUNET_BANDWIDTH_tracker_update_quota (&n->in_tracker, quota);
1702   if (0 != ntohl (quota.value__))
1703     return;
1704 #if DEBUG_TRANSPORT
1705   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting peer `%4s' due to `%s'\n",
1706               GNUNET_i2s (&n->id), "SET_QUOTA");
1707 #endif
1708   if (is_connected(n))
1709     GNUNET_STATISTICS_update (GST_stats,
1710                               gettext_noop ("# disconnects due to quota of 0"), 1,
1711                               GNUNET_NO);
1712   disconnect_neighbour (n);
1713 }
1714
1715
1716 /**
1717  * Closure for the neighbours_iterate function.
1718  */
1719 struct IteratorContext
1720 {
1721   /**
1722    * Function to call on each connected neighbour.
1723    */
1724   GST_NeighbourIterator cb;
1725
1726   /**
1727    * Closure for 'cb'.
1728    */
1729   void *cb_cls;
1730 };
1731
1732
1733 /**
1734  * Call the callback from the closure for each connected neighbour.
1735  *
1736  * @param cls the 'struct IteratorContext'
1737  * @param key the hash of the public key of the neighbour
1738  * @param value the 'struct NeighbourMapEntry'
1739  * @return GNUNET_OK (continue to iterate)
1740  */
1741 static int
1742 neighbours_iterate (void *cls, const GNUNET_HashCode * key, void *value)
1743 {
1744   struct IteratorContext *ic = cls;
1745   struct NeighbourMapEntry *n = value;
1746
1747   if (!is_connected(n))
1748     return GNUNET_OK;
1749
1750   ic->cb (ic->cb_cls, &n->id, NULL, 0, n->plugin_name, n->addr, n->addrlen);
1751   return GNUNET_OK;
1752 }
1753
1754
1755 /**
1756  * Iterate over all connected neighbours.
1757  *
1758  * @param cb function to call
1759  * @param cb_cls closure for cb
1760  */
1761 void
1762 GST_neighbours_iterate (GST_NeighbourIterator cb, void *cb_cls)
1763 {
1764   struct IteratorContext ic;
1765
1766   GNUNET_assert (neighbours != NULL);
1767
1768   ic.cb = cb;
1769   ic.cb_cls = cb_cls;
1770   GNUNET_CONTAINER_multihashmap_iterate (neighbours, &neighbours_iterate, &ic);
1771 }
1772
1773 /**
1774  * If we have an active connection to the given target, it must be shutdown.
1775  *
1776  * @param target peer to disconnect from
1777  */
1778 void
1779 GST_neighbours_force_disconnect (const struct GNUNET_PeerIdentity *target)
1780 {
1781   struct NeighbourMapEntry *n;
1782
1783   GNUNET_assert (neighbours != NULL);
1784
1785   n = lookup_neighbour (target);
1786   if (NULL == n)
1787     return;                     /* not active */
1788   if (is_connected(n))
1789   {
1790     send_disconnect(&n->id, n->plugin_name, n->addr, n->addrlen, n->session);
1791
1792     n = lookup_neighbour (target);
1793     if (NULL == n)
1794       return;                     /* gone already */
1795   }
1796   disconnect_neighbour (n);
1797 }
1798
1799
1800 /**
1801  * We received a disconnect message from the given peer,
1802  * validate and process.
1803  * 
1804  * @param peer sender of the message
1805  * @param msg the disconnect message
1806  */
1807 void
1808 GST_neighbours_handle_disconnect_message (const struct GNUNET_PeerIdentity *peer,
1809                                           const struct GNUNET_MessageHeader *msg)
1810 {
1811   struct NeighbourMapEntry *n;
1812   const struct SessionDisconnectMessage *sdm;
1813   GNUNET_HashCode hc;
1814
1815 #if DEBUG_TRANSPORT
1816   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1817       "Received DISCONNECT message from peer `%s'\n", GNUNET_i2s (peer));
1818 #endif
1819
1820   if (ntohs (msg->size) != sizeof (struct SessionDisconnectMessage))
1821   {
1822     // GNUNET_break_op (0);
1823     GNUNET_STATISTICS_update (GST_stats,
1824                               gettext_noop ("# disconnect messages ignored (old format)"), 1,
1825                               GNUNET_NO);
1826     return;
1827   }
1828   sdm = (const struct SessionDisconnectMessage* ) msg;
1829   n = lookup_neighbour (peer);
1830   if (NULL == n)
1831     return;                     /* gone already */
1832   if (GNUNET_TIME_absolute_ntoh (sdm->timestamp).abs_value <=
1833       n->connect_ts.abs_value)
1834   {
1835     GNUNET_STATISTICS_update (GST_stats,
1836                               gettext_noop ("# disconnect messages ignored (timestamp)"), 1,
1837                               GNUNET_NO);
1838     return;
1839   }
1840   GNUNET_CRYPTO_hash (&sdm->public_key,
1841                       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1842                       &hc);
1843   if (0 != memcmp (peer,
1844                    &hc,
1845                    sizeof (struct GNUNET_PeerIdentity)))
1846   {
1847     GNUNET_break_op (0);
1848     return;
1849   }
1850   if (ntohl (sdm->purpose.size) != 
1851       sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) +
1852       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) +
1853       sizeof (struct GNUNET_TIME_AbsoluteNBO))
1854   {
1855     GNUNET_break_op (0);
1856     return;
1857   }
1858   if (GNUNET_OK !=
1859       GNUNET_CRYPTO_rsa_verify (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT,
1860                                 &sdm->purpose,
1861                                 &sdm->signature,
1862                                 &sdm->public_key))
1863   {
1864     GNUNET_break_op (0);
1865     return;
1866   }
1867   GST_neighbours_force_disconnect (peer);
1868 }
1869
1870 /**
1871  * We received a 'SESSION_CONNECT_ACK' message from the other peer.
1872  * Consider switching to it.
1873  *
1874  * @param message possibly a 'struct SessionConnectMessage' (check format)
1875  * @param peer identity of the peer to switch the address for
1876  * @param plugin_name name of transport that delivered the PONG
1877  * @param address address of the other peer, NULL if other peer
1878  *                       connected to us
1879  * @param address_len number of bytes in address
1880  * @param session session to use (or NULL)
1881  * @param ats performance data
1882  * @param ats_count number of entries in ats
1883   */
1884 void
1885 GST_neighbours_handle_connect_ack (const struct GNUNET_MessageHeader *message,
1886                                const struct GNUNET_PeerIdentity *peer,
1887                                const char *plugin_name,
1888                                const char *sender_address, uint16_t sender_address_len,
1889                                struct Session *session,
1890                                const struct GNUNET_ATS_Information *ats,
1891                                uint32_t ats_count)
1892 {
1893   const struct SessionConnectMessage *scm;
1894   struct QuotaSetMessage q_msg;
1895   struct GNUNET_MessageHeader msg;
1896   struct NeighbourMapEntry *n;
1897   size_t msg_len;
1898   size_t ret;
1899   int was_connected;
1900
1901 #if DEBUG_TRANSPORT
1902   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1903       "Received CONNECT_ACK message from peer `%s'\n", GNUNET_i2s (peer));
1904 #endif
1905
1906   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
1907   {
1908     GNUNET_break_op (0);
1909     return;
1910   }
1911
1912   scm = (const struct SessionConnectMessage *) message;
1913   GNUNET_break_op (ntohl (scm->reserved) == 0);
1914   n = lookup_neighbour (peer);
1915   if (NULL == n)
1916     n = setup_neighbour (peer);
1917 /*
1918   if (n->state != S_CONNECT_SENT)
1919   {
1920     GNUNET_break (0);
1921     send_disconnect(&n->id, n->plugin_name, n->addr, n->addrlen, n->session);
1922     return;
1923   }
1924 */
1925   if (NULL != session)
1926     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1927                      "transport-ats",
1928                      "Giving ATS session %p of plugin %s for peer %s\n",
1929                      session,
1930                      plugin_name,
1931                      GNUNET_i2s (peer));
1932   GNUNET_ATS_address_update (GST_ats,
1933                              peer,
1934                              plugin_name, sender_address, sender_address_len,
1935                              session, ats, ats_count);
1936
1937   was_connected = is_connected(n);
1938   if (!is_connected(n))
1939     change_state (n, S_CONNECTED);
1940
1941 #if DEBUG_TRANSPORT
1942   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1943               "Setting inbound quota of %u for peer `%s' to \n",
1944               ntohl (n->bandwidth_in.value__), GNUNET_i2s (&n->id));
1945 #endif
1946   GST_neighbours_set_incoming_quota(&n->id, n->bandwidth_in);
1947
1948   /* send ACK (ACK)*/
1949   msg_len =  sizeof (msg);
1950   msg.size = htons (msg_len);
1951   msg.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_ACK);
1952
1953   ret = send_with_plugin (&n->id, (const char *) &msg, msg_len, UINT32_MAX,
1954                           GNUNET_TIME_UNIT_FOREVER_REL,
1955                           n->session, n->plugin_name, n->addr, n->addrlen,
1956                           GNUNET_YES, NULL, NULL);
1957
1958   if (ret == GNUNET_SYSERR)
1959     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1960               "Failed to send SESSION_ACK to `%4s' using plugin `%s' address '%s' session %X\n",
1961               GNUNET_i2s (&n->id), n->plugin_name,
1962               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1963                                                                  n->addr,
1964                                                                  n->addrlen),
1965               n->session);
1966
1967
1968   if (!was_connected)
1969   {
1970     if (n->keepalive_task == GNUNET_SCHEDULER_NO_TASK)
1971       n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
1972                                                         &neighbour_keepalive_task,
1973                                                         n);
1974
1975     neighbours_connected++;
1976     GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# peers connected"), 1,
1977                               GNUNET_NO);
1978 #if DEBUG_TRANSPORT
1979     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1980               "Notify about connect of `%4s' using plugin `%s' address '%s' session %X LINE %u\n",
1981               GNUNET_i2s (&n->id), n->plugin_name,
1982               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1983                                                                  n->addr,
1984                                                                  n->addrlen),
1985               n->session, __LINE__);
1986 #endif
1987     connect_notify_cb (callback_cls, &n->id, ats, ats_count);
1988   }
1989
1990 #if DEBUG_TRANSPORT
1991     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1992                 "Sending outbound quota of %u Bps for peer `%s' to all clients\n",
1993                 ntohl (n->bandwidth_out.value__), GNUNET_i2s (peer));
1994 #endif
1995     q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
1996     q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
1997     q_msg.quota = n->bandwidth_out;
1998     q_msg.peer = (*peer);
1999     GST_clients_broadcast (&q_msg.header, GNUNET_NO);
2000
2001 }
2002
2003 void
2004 GST_neighbours_handle_ack (const struct GNUNET_MessageHeader *message,
2005     const struct GNUNET_PeerIdentity *peer,
2006     const char *plugin_name,
2007     const char *sender_address, uint16_t sender_address_len,
2008     struct Session *session,
2009     const struct GNUNET_ATS_Information *ats,
2010     uint32_t ats_count)
2011 {
2012   struct NeighbourMapEntry *n;
2013   struct QuotaSetMessage q_msg;
2014   int was_connected;
2015
2016 #if DEBUG_TRANSPORT
2017   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2018       "Received ACK message from peer `%s'\n", GNUNET_i2s (peer));
2019 #endif
2020
2021   if (ntohs (message->size) != sizeof (struct GNUNET_MessageHeader))
2022   {
2023     GNUNET_break_op (0);
2024     return;
2025   }
2026
2027   n = lookup_neighbour (peer);
2028   if (NULL == n)
2029   {
2030     send_disconnect(peer, plugin_name, sender_address, sender_address_len, session);
2031     GNUNET_break (0);
2032   }
2033
2034   if (is_connected(n))
2035     return;
2036
2037   if (NULL != session)
2038     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2039                      "transport-ats",
2040                      "Giving ATS session %p of plugin %s for peer %s\n",
2041                      session,
2042                      plugin_name,
2043                      GNUNET_i2s (peer));
2044   GNUNET_ATS_address_update (GST_ats,
2045                              peer,
2046                              plugin_name, sender_address, sender_address_len,
2047                              session, ats, ats_count);
2048
2049   was_connected = is_connected(n);
2050   change_state (n, S_CONNECTED);
2051
2052   GST_neighbours_set_incoming_quota(&n->id, n->bandwidth_in);
2053
2054   if (n->keepalive_task == GNUNET_SCHEDULER_NO_TASK)
2055     n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
2056                                                       &neighbour_keepalive_task,
2057                                                       n);
2058
2059   if (!was_connected)
2060   {
2061   neighbours_connected++;
2062   GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# peers connected"), 1,
2063                             GNUNET_NO);
2064
2065 #if DEBUG_TRANSPORT
2066     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2067               "Notify about connect of `%4s' using plugin `%s' address '%s' session %X LINE %u\n",
2068               GNUNET_i2s (&n->id), n->plugin_name,
2069               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
2070                                                                  n->addr,
2071                                                                  n->addrlen),
2072               n->session, __LINE__);
2073 #endif
2074   connect_notify_cb (callback_cls, &n->id, ats, ats_count);
2075   }
2076 #if DEBUG_TRANSPORT
2077   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2078               "Sending outbound quota of %u Bps for peer `%s' to all clients\n",
2079               ntohl (n->bandwidth_out.value__), GNUNET_i2s (peer));
2080 #endif
2081
2082   q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
2083   q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
2084   q_msg.quota = n->bandwidth_out;
2085   q_msg.peer = (*peer);
2086   GST_clients_broadcast (&q_msg.header, GNUNET_NO);
2087
2088 }
2089
2090 struct BlackListCheckContext
2091 {
2092   struct GNUNET_ATS_Information *ats;
2093
2094   uint32_t ats_count;
2095
2096   struct Session *session;
2097
2098   char *sender_address;
2099
2100   uint16_t sender_address_len;
2101
2102   char *plugin_name;
2103
2104   struct GNUNET_TIME_Absolute ts;
2105 };
2106
2107
2108 static void
2109 handle_connect_blacklist_cont (void *cls,
2110                                const struct GNUNET_PeerIdentity
2111                                * peer, int result)
2112 {
2113   struct NeighbourMapEntry *n;
2114   struct BlackListCheckContext * bcc = cls;
2115
2116 #if DEBUG_TRANSPORT
2117   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2118       "Blacklist check due to CONNECT message: `%s'\n", GNUNET_i2s (peer), (result == GNUNET_OK) ? "ALLOWED" : "FORBIDDEN");
2119 #endif
2120
2121   /* not allowed */
2122   if (GNUNET_OK != result)
2123   {
2124     GNUNET_free (bcc);
2125     return;
2126   }
2127
2128   n = lookup_neighbour (peer);
2129   if (NULL == n)
2130     n = setup_neighbour (peer);
2131
2132   if (bcc->ts.abs_value > n->connect_ts.abs_value)
2133   {
2134     if (NULL != bcc->session)
2135       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2136                        "transport-ats",
2137                        "Giving ATS session %p of plugin %s address `%s' for peer %s\n",
2138                        bcc->session,
2139                        bcc->plugin_name,
2140                        GST_plugins_a2s (bcc->plugin_name, bcc->sender_address, bcc->sender_address_len),
2141                        GNUNET_i2s (peer));
2142     GNUNET_ATS_address_update (GST_ats,
2143                                peer,
2144                                bcc->plugin_name, bcc->sender_address, bcc->sender_address_len,
2145                                bcc->session, bcc->ats, bcc->ats_count);
2146     n->connect_ts = bcc->ts;
2147   }
2148
2149   GNUNET_free (bcc);
2150 /*
2151   if (n->state != S_NOT_CONNECTED)
2152     return;*/
2153   change_state (n, S_CONNECT_RECV);
2154
2155   /* Ask ATS for an address to connect via that address */
2156   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
2157     GNUNET_SCHEDULER_cancel(n->ats_suggest);
2158   n->ats_suggest = GNUNET_SCHEDULER_add_delayed (ATS_RESPONSE_TIMEOUT, ats_suggest_cancel, n);
2159   GNUNET_ATS_suggest_address(GST_ats, peer);
2160 }
2161
2162 /**
2163  * We received a 'SESSION_CONNECT' message from the other peer.
2164  * Consider switching to it.
2165  *
2166  * @param message possibly a 'struct SessionConnectMessage' (check format)
2167  * @param peer identity of the peer to switch the address for
2168  * @param plugin_name name of transport that delivered the PONG
2169  * @param address address of the other peer, NULL if other peer
2170  *                       connected to us
2171  * @param address_len number of bytes in address
2172  * @param session session to use (or NULL)
2173  * @param ats performance data
2174  * @param ats_count number of entries in ats (excluding 0-termination)
2175   */
2176 void
2177 GST_neighbours_handle_connect (const struct GNUNET_MessageHeader *message,
2178                                const struct GNUNET_PeerIdentity *peer,
2179                                const char *plugin_name,
2180                                const char *sender_address, uint16_t sender_address_len,
2181                                struct Session *session,
2182                                const struct GNUNET_ATS_Information *ats,
2183                                uint32_t ats_count)
2184 {
2185   const struct SessionConnectMessage *scm;
2186   struct NeighbourMapEntry * n;
2187   struct BlackListCheckContext * bcc = NULL;
2188
2189 #if DEBUG_TRANSPORT
2190   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2191       "Received CONNECT message from peer `%s'\n", GNUNET_i2s (peer));
2192 #endif
2193
2194   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
2195   {
2196     GNUNET_break_op (0);
2197     return;
2198   }
2199
2200   scm = (const struct SessionConnectMessage *) message;
2201   GNUNET_break_op (ntohl (scm->reserved) == 0);
2202
2203   n = lookup_neighbour(peer);
2204   if (n != NULL)
2205   {
2206     /* connected peer switches addresses */
2207     if (is_connected(n))
2208     {
2209       GNUNET_ATS_address_update(GST_ats, peer, plugin_name, sender_address, sender_address_len, session, ats, ats_count);
2210       return;
2211     }
2212   }
2213
2214   /* we are not connected to this peer */
2215   /* do blacklist check*/
2216   bcc = GNUNET_malloc (sizeof (struct BlackListCheckContext) +
2217             sizeof (struct GNUNET_ATS_Information) * ats_count +
2218             sender_address_len +
2219             strlen (plugin_name)+1);
2220
2221   bcc->ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
2222
2223   bcc->ats_count = ats_count;
2224   bcc->sender_address_len = sender_address_len;
2225   bcc->session = session;
2226
2227   bcc->ats = (struct GNUNET_ATS_Information *) &bcc[1];
2228   memcpy (bcc->ats, ats,sizeof (struct GNUNET_ATS_Information) * ats_count );
2229
2230   bcc->sender_address = (char *) &bcc->ats[ats_count];
2231   memcpy (bcc->sender_address, sender_address , sender_address_len);
2232
2233   bcc->plugin_name = &bcc->sender_address[sender_address_len];
2234   strcpy (bcc->plugin_name, plugin_name);
2235
2236   GST_blacklist_test_allowed (peer, plugin_name, handle_connect_blacklist_cont, bcc);
2237 }
2238
2239
2240 /* end of file gnunet-service-transport_neighbours.c */