src/transport/gnunet-service-transport_neighbours.c

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