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