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