src/transport/gnunet-service-transport_neighbours.c

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