src/transport/gnunet-service-transport_neighbours_3way.c

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