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 DEBUG_TRANSPORT
1110   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1111               "ATS tells us to switch to plugin `%s' address '%s' session %X for %s peer `%s'\n",
1112               plugin_name,
1113               (address_len == 0) ? "<inbound>" : GST_plugins_a2s (plugin_name,
1114                                                                   address,
1115                                                                   address_len),
1116               session, (is_connected(n) ? "CONNECTED" : "NOT CONNECTED"),
1117               GNUNET_i2s (peer));
1118 #endif
1119
1120   GNUNET_free_non_null (n->addr);
1121   n->addr = GNUNET_malloc (address_len);
1122   memcpy (n->addr, address, address_len);
1123   n->bandwidth_in = bandwidth_in;
1124   n->bandwidth_out = bandwidth_out;
1125   n->addrlen = address_len;
1126   n->session = session;
1127   GNUNET_free_non_null (n->plugin_name);
1128   n->plugin_name = GNUNET_strdup (plugin_name);
1129   GNUNET_SCHEDULER_cancel (n->timeout_task);
1130   n->timeout_task =
1131       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1132                                     &neighbour_timeout_task, n);
1133
1134   if (n->state == S_DISCONNECT)
1135   {
1136     /* We are disconnecting, nothing to do here */
1137     return GNUNET_NO;
1138   }
1139   /* We are not connected/connecting and initiate a fresh connect */
1140   if (n->state == S_NOT_CONNECTED)
1141   {
1142     msg_len = sizeof (struct SessionConnectMessage);
1143     connect_msg.header.size = htons (msg_len);
1144     connect_msg.header.type =
1145         htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT);
1146     connect_msg.reserved = htonl (0);
1147     connect_msg.timestamp =
1148         GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1149
1150     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);
1151     if (ret == GNUNET_SYSERR)
1152     {
1153       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1154                 "Failed to send CONNECT_MESSAGE to `%4s' using plugin `%s' address '%s' session %X\n",
1155                 GNUNET_i2s (peer), plugin_name,
1156                 (address_len == 0) ? "<inbound>" : GST_plugins_a2s (plugin_name,
1157                                                                     address,
1158                                                                     address_len),
1159                 session);
1160     }
1161     return GNUNET_NO;
1162   }
1163   /* We received a CONNECT message and asked ATS for an address */
1164   else if (n->state == S_CONNECT_RECV)
1165   {
1166     msg_len = sizeof (struct SessionConnectMessage);
1167     connect_msg.header.size = htons (msg_len);
1168     connect_msg.header.type =
1169       htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT_ACK);
1170     connect_msg.reserved = htonl (0);
1171     connect_msg.timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1172
1173     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);
1174     if (ret == GNUNET_SYSERR)
1175     {
1176       change_state (n, S_NOT_CONNECTED);
1177       GNUNET_break (0);
1178     }
1179     return GNUNET_NO;
1180   }
1181   /* connected peer is switching addresses */
1182   else if (n->state == S_CONNECTED)
1183   {
1184     msg_len = sizeof (struct SessionConnectMessage);
1185     connect_msg.header.size = htons (msg_len);
1186     connect_msg.header.type =
1187         htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_CONNECT);
1188     connect_msg.reserved = htonl (0);
1189     connect_msg.timestamp =
1190         GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
1191
1192     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);
1193     if (ret == GNUNET_SYSERR)
1194     {
1195       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1196                 "Failed to send CONNECT_MESSAGE to `%4s' using plugin `%s' address '%s' session %X\n",
1197                 GNUNET_i2s (peer), plugin_name,
1198                 (address_len == 0) ? "<inbound>" : GST_plugins_a2s (plugin_name,
1199                                                                     address,
1200                                                                     address_len),
1201                 session);
1202     }
1203     return GNUNET_NO;
1204   }
1205
1206   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Invalid connection state to switch addresses %u ", n->state);
1207   GNUNET_break_op (0);
1208   return GNUNET_NO;
1209 }
1210
1211
1212 /**
1213  * Create an entry in the neighbour map for the given peer
1214  *
1215  * @param peer peer to create an entry for
1216  * @return new neighbour map entry
1217  */
1218 static struct NeighbourMapEntry *
1219 setup_neighbour (const struct GNUNET_PeerIdentity *peer)
1220 {
1221   struct NeighbourMapEntry *n;
1222
1223 #if DEBUG_TRANSPORT
1224   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1225               "Unknown peer `%s', creating new neighbour\n",
1226               GNUNET_i2s (peer));
1227 #endif
1228   n = GNUNET_malloc (sizeof (struct NeighbourMapEntry));
1229   n->id = *peer;
1230   n->state = S_NOT_CONNECTED;
1231   GNUNET_BANDWIDTH_tracker_init (&n->in_tracker,
1232                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
1233                                  MAX_BANDWIDTH_CARRY_S);
1234   n->timeout_task =
1235     GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1236                                   &neighbour_timeout_task, n);
1237   GNUNET_assert (GNUNET_OK ==
1238                  GNUNET_CONTAINER_multihashmap_put (neighbours,
1239                                                     &n->id.hashPubKey, n,
1240                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1241   return n;
1242 }
1243
1244
1245 /**
1246  * Try to create a connection to the given target (eventually).
1247  *
1248  * @param target peer to try to connect to
1249  */
1250 void
1251 GST_neighbours_try_connect (const struct GNUNET_PeerIdentity *target)
1252 {
1253   struct NeighbourMapEntry *n;
1254
1255   GNUNET_assert (neighbours != NULL);
1256 #if DEBUG_TRANSPORT
1257   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Trying to connect to peer `%s'\n",
1258               GNUNET_i2s (target));
1259 #endif
1260   GNUNET_assert (0 !=
1261                  memcmp (target, &GST_my_identity,
1262                          sizeof (struct GNUNET_PeerIdentity)));
1263   n = lookup_neighbour (target);
1264
1265   if (NULL != n)
1266   {
1267     if ((is_connected(n)) || (is_connecting(n)))
1268       return;                     /* already connecting or connected */
1269     if (is_disconnecting(n))
1270       change_state (n, S_NOT_CONNECTED);
1271   }
1272
1273
1274   if (n == NULL)
1275     n = setup_neighbour (target);
1276 #if DEBUG_TRANSPORT
1277   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1278               "Asking ATS for suggested address to connect to peer `%s'\n",
1279               GNUNET_i2s (&n->id));
1280 #endif
1281
1282   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1283     GNUNET_SCHEDULER_cancel(n->ats_suggest);
1284    n->ats_suggest = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS, ats_suggest_cancel, n);
1285    GNUNET_ATS_suggest_address (GST_ats, &n->id);
1286 }
1287
1288 /**
1289  * Test if we're connected to the given peer.
1290  *
1291  * @param target peer to test
1292  * @return GNUNET_YES if we are connected, GNUNET_NO if not
1293  */
1294 int
1295 GST_neighbours_test_connected (const struct GNUNET_PeerIdentity *target)
1296 {
1297   struct NeighbourMapEntry *n;
1298
1299   GNUNET_assert (neighbours != NULL);
1300
1301   n = lookup_neighbour (target);
1302
1303   if ((NULL == n) || (!is_connected(n)))
1304     return GNUNET_NO;           /* not connected */
1305   return GNUNET_YES;
1306 }
1307
1308
1309 /**
1310  * A session was terminated. Take note.
1311  *
1312  * @param peer identity of the peer where the session died
1313  * @param session session that is gone
1314  */
1315 void
1316 GST_neighbours_session_terminated (const struct GNUNET_PeerIdentity *peer,
1317                                    struct Session *session)
1318 {
1319   struct NeighbourMapEntry *n;
1320
1321   GNUNET_assert (neighbours != NULL);
1322
1323 #if DEBUG_TRANSPORT
1324   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1325               "Session %X to peer `%s' ended \n",
1326               session, GNUNET_i2s (peer));
1327 #endif
1328
1329   n = lookup_neighbour (peer);
1330   if (NULL == n)
1331     return;
1332   if (session != n->session)
1333     return;                     /* doesn't affect us */
1334
1335   n->session = NULL;
1336   GNUNET_free (n->addr);
1337   n->addr = NULL;
1338   n->addrlen = 0;
1339
1340   /* not connected anymore anyway, shouldn't matter */
1341   if ((!is_connected(n)) && (!is_connecting(n)))
1342     return;
1343
1344   /* We are connected, so ask ATS to switch addresses */
1345   GNUNET_SCHEDULER_cancel (n->timeout_task);
1346   n->timeout_task =
1347       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_DISCONNECT_SESSION_TIMEOUT,
1348                                     &neighbour_timeout_task, n);
1349   /* try QUICKLY to re-establish a connection, reduce timeout! */
1350   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
1351     GNUNET_SCHEDULER_cancel(n->ats_suggest);
1352   n->ats_suggest = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS, ats_suggest_cancel, n);
1353   GNUNET_ATS_suggest_address (GST_ats, peer);
1354 }
1355
1356
1357 /**
1358  * Transmit a message to the given target using the active connection.
1359  *
1360  * @param target destination
1361  * @param msg message to send
1362  * @param msg_size number of bytes in msg
1363  * @param timeout when to fail with timeout
1364  * @param cont function to call when done
1365  * @param cont_cls closure for 'cont'
1366  */
1367 void
1368 GST_neighbours_send (const struct GNUNET_PeerIdentity *target, const void *msg,
1369                      size_t msg_size, struct GNUNET_TIME_Relative timeout,
1370                      GST_NeighbourSendContinuation cont, void *cont_cls)
1371 {
1372   struct NeighbourMapEntry *n;
1373   struct MessageQueue *mq;
1374
1375   GNUNET_assert (neighbours != NULL);
1376
1377   n = lookup_neighbour (target);
1378   if ((n == NULL) || (!is_connected(n)))
1379   {
1380     GNUNET_STATISTICS_update (GST_stats,
1381                               gettext_noop
1382                               ("# messages not sent (no such peer or not connected)"),
1383                               1, GNUNET_NO);
1384 #if DEBUG_TRANSPORT
1385     if (n == NULL)
1386       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1387                   "Could not send message to peer `%s': unknown neighbour",
1388                   GNUNET_i2s (target));
1389     else if (!is_connected(n))
1390       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1391                   "Could not send message to peer `%s': not connected\n",
1392                   GNUNET_i2s (target));
1393 #endif
1394     if (NULL != cont)
1395       cont (cont_cls, GNUNET_SYSERR);
1396     return;
1397   }
1398
1399   if ((n->session == NULL) && (n->addr == NULL) && (n->addrlen ==0))
1400   {
1401     GNUNET_STATISTICS_update (GST_stats,
1402                               gettext_noop
1403                               ("# messages not sent (no such peer or not connected)"),
1404                               1, GNUNET_NO);
1405 #if DEBUG_TRANSPORT
1406       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1407                   "Could not send message to peer `%s': no address available\n",
1408                   GNUNET_i2s (target));
1409 #endif
1410
1411     if (NULL != cont)
1412       cont (cont_cls, GNUNET_SYSERR);
1413     return;
1414   }
1415
1416   GNUNET_assert (msg_size >= sizeof (struct GNUNET_MessageHeader));
1417   GNUNET_STATISTICS_update (GST_stats,
1418                             gettext_noop
1419                             ("# bytes in message queue for other peers"),
1420                             msg_size, GNUNET_NO);
1421   mq = GNUNET_malloc (sizeof (struct MessageQueue) + msg_size);
1422   mq->cont = cont;
1423   mq->cont_cls = cont_cls;
1424   /* FIXME: this memcpy can be up to 7% of our total runtime! */
1425   memcpy (&mq[1], msg, msg_size);
1426   mq->message_buf = (const char *) &mq[1];
1427   mq->message_buf_size = msg_size;
1428   mq->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1429   GNUNET_CONTAINER_DLL_insert_tail (n->messages_head, n->messages_tail, mq);
1430
1431   if ((GNUNET_SCHEDULER_NO_TASK == n->transmission_task) &&
1432       (NULL == n->is_active))
1433     n->transmission_task = GNUNET_SCHEDULER_add_now (&transmission_task, n);
1434 }
1435
1436
1437 /**
1438  * We have received a message from the given sender.  How long should
1439  * we delay before receiving more?  (Also used to keep the peer marked
1440  * as live).
1441  *
1442  * @param sender sender of the message
1443  * @param size size of the message
1444  * @param do_forward set to GNUNET_YES if the message should be forwarded to clients
1445  *                   GNUNET_NO if the neighbour is not connected or violates the quota,
1446  *                   GNUNET_SYSERR if the connection is not fully up yet
1447  * @return how long to wait before reading more from this sender
1448  */
1449 struct GNUNET_TIME_Relative
1450 GST_neighbours_calculate_receive_delay (const struct GNUNET_PeerIdentity
1451                                         *sender, ssize_t size, int *do_forward)
1452 {
1453   struct NeighbourMapEntry *n;
1454   struct GNUNET_TIME_Relative ret;
1455
1456   GNUNET_assert (neighbours != NULL);
1457
1458   n = lookup_neighbour (sender);
1459   if (n == NULL)
1460   {
1461     GST_neighbours_try_connect (sender);
1462     n = lookup_neighbour (sender);
1463     if (NULL == n)
1464     {
1465       GNUNET_STATISTICS_update (GST_stats,
1466                                 gettext_noop
1467                                 ("# messages discarded due to lack of neighbour record"),
1468                                 1, GNUNET_NO);
1469       *do_forward = GNUNET_NO;
1470       return GNUNET_TIME_UNIT_ZERO;
1471     }
1472   }
1473   if (!is_connected(n))
1474   {
1475     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1476                 _("Plugin gave us %d bytes of data but somehow the session is not marked as UP yet!\n"),
1477                 (int) size);
1478     *do_forward = GNUNET_SYSERR;
1479     return GNUNET_TIME_UNIT_ZERO;
1480   }
1481   if (GNUNET_YES == GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, size))
1482   {
1483     n->quota_violation_count++;
1484 #if DEBUG_TRANSPORT
1485     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1486                 "Bandwidth quota (%u b/s) violation detected (total of %u).\n",
1487                 n->in_tracker.available_bytes_per_s__,
1488                 n->quota_violation_count);
1489 #endif
1490     /* Discount 32k per violation */
1491     GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, -32 * 1024);
1492   }
1493   else
1494   {
1495     if (n->quota_violation_count > 0)
1496     {
1497       /* try to add 32k back */
1498       GNUNET_BANDWIDTH_tracker_consume (&n->in_tracker, 32 * 1024);
1499       n->quota_violation_count--;
1500     }
1501   }
1502   if (n->quota_violation_count > QUOTA_VIOLATION_DROP_THRESHOLD)
1503   {
1504     GNUNET_STATISTICS_update (GST_stats,
1505                               gettext_noop
1506                               ("# bandwidth quota violations by other peers"),
1507                               1, GNUNET_NO);
1508     *do_forward = GNUNET_NO;
1509     return GNUNET_CONSTANTS_QUOTA_VIOLATION_TIMEOUT;
1510   }
1511   *do_forward = GNUNET_YES;
1512   ret = GNUNET_BANDWIDTH_tracker_get_delay (&n->in_tracker, 32 * 1024);
1513   if (ret.rel_value > 0)
1514   {
1515 #if DEBUG_TRANSPORT
1516     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1517                 "Throttling read (%llu bytes excess at %u b/s), waiting %llu ms before reading more.\n",
1518                 (unsigned long long) n->in_tracker.
1519                 consumption_since_last_update__,
1520                 (unsigned int) n->in_tracker.available_bytes_per_s__,
1521                 (unsigned long long) ret.rel_value);
1522 #endif
1523     GNUNET_STATISTICS_update (GST_stats,
1524                               gettext_noop ("# ms throttling suggested"),
1525                               (int64_t) ret.rel_value, GNUNET_NO);
1526   }
1527   return ret;
1528 }
1529
1530
1531 /**
1532  * Keep the connection to the given neighbour alive longer,
1533  * we received a KEEPALIVE (or equivalent).
1534  *
1535  * @param neighbour neighbour to keep alive
1536  */
1537 void
1538 GST_neighbours_keepalive (const struct GNUNET_PeerIdentity *neighbour)
1539 {
1540   struct NeighbourMapEntry *n;
1541
1542   GNUNET_assert (neighbours != NULL);
1543
1544   n = lookup_neighbour (neighbour);
1545   if (NULL == n)
1546   {
1547     GNUNET_STATISTICS_update (GST_stats,
1548                               gettext_noop
1549                               ("# KEEPALIVE messages discarded (not connected)"),
1550                               1, GNUNET_NO);
1551     return;
1552   }
1553   GNUNET_SCHEDULER_cancel (n->timeout_task);
1554   n->timeout_task =
1555       GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1556                                     &neighbour_timeout_task, n);
1557 }
1558
1559
1560 /**
1561  * Change the incoming quota for the given peer.
1562  *
1563  * @param neighbour identity of peer to change qutoa for
1564  * @param quota new quota
1565  */
1566 void
1567 GST_neighbours_set_incoming_quota (const struct GNUNET_PeerIdentity *neighbour,
1568                                    struct GNUNET_BANDWIDTH_Value32NBO quota)
1569 {
1570   struct NeighbourMapEntry *n;
1571
1572   GNUNET_assert (neighbours != NULL);
1573
1574   n = lookup_neighbour (neighbour);
1575   if (n == NULL)
1576   {
1577     GNUNET_STATISTICS_update (GST_stats,
1578                               gettext_noop
1579                               ("# SET QUOTA messages ignored (no such peer)"),
1580                               1, GNUNET_NO);
1581     return;
1582   }
1583   GNUNET_BANDWIDTH_tracker_update_quota (&n->in_tracker, quota);
1584   if (0 != ntohl (quota.value__))
1585     return;
1586 #if DEBUG_TRANSPORT
1587   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting peer `%4s' due to `%s'\n",
1588               GNUNET_i2s (&n->id), "SET_QUOTA");
1589 #endif
1590   if (is_connected(n))
1591     GNUNET_STATISTICS_update (GST_stats,
1592                               gettext_noop ("# disconnects due to quota of 0"), 1,
1593                               GNUNET_NO);
1594   disconnect_neighbour (n);
1595 }
1596
1597
1598 /**
1599  * Closure for the neighbours_iterate function.
1600  */
1601 struct IteratorContext
1602 {
1603   /**
1604    * Function to call on each connected neighbour.
1605    */
1606   GST_NeighbourIterator cb;
1607
1608   /**
1609    * Closure for 'cb'.
1610    */
1611   void *cb_cls;
1612 };
1613
1614
1615 /**
1616  * Call the callback from the closure for each connected neighbour.
1617  *
1618  * @param cls the 'struct IteratorContext'
1619  * @param key the hash of the public key of the neighbour
1620  * @param value the 'struct NeighbourMapEntry'
1621  * @return GNUNET_OK (continue to iterate)
1622  */
1623 static int
1624 neighbours_iterate (void *cls, const GNUNET_HashCode * key, void *value)
1625 {
1626   struct IteratorContext *ic = cls;
1627   struct NeighbourMapEntry *n = value;
1628
1629   if (is_connected(n))
1630     return GNUNET_OK;
1631
1632   ic->cb (ic->cb_cls, &n->id, NULL, 0, n->plugin_name, n->addr, n->addrlen);
1633   return GNUNET_OK;
1634 }
1635
1636
1637 /**
1638  * Iterate over all connected neighbours.
1639  *
1640  * @param cb function to call
1641  * @param cb_cls closure for cb
1642  */
1643 void
1644 GST_neighbours_iterate (GST_NeighbourIterator cb, void *cb_cls)
1645 {
1646   struct IteratorContext ic;
1647
1648   GNUNET_assert (neighbours != NULL);
1649
1650   ic.cb = cb;
1651   ic.cb_cls = cb_cls;
1652   GNUNET_CONTAINER_multihashmap_iterate (neighbours, &neighbours_iterate, &ic);
1653 }
1654
1655 /**
1656  * If we have an active connection to the given target, it must be shutdown.
1657  *
1658  * @param target peer to disconnect from
1659  */
1660 void
1661 GST_neighbours_force_disconnect (const struct GNUNET_PeerIdentity *target)
1662 {
1663   struct NeighbourMapEntry *n;
1664
1665   GNUNET_assert (neighbours != NULL);
1666
1667   n = lookup_neighbour (target);
1668   if (NULL == n)
1669     return;                     /* not active */
1670   if (is_connected(n))
1671   {
1672     send_disconnect(n);
1673
1674     n = lookup_neighbour (target);
1675     if (NULL == n)
1676       return;                     /* gone already */
1677   }
1678   disconnect_neighbour (n);
1679 }
1680
1681
1682 /**
1683  * We received a disconnect message from the given peer,
1684  * validate and process.
1685  *
1686  * @param peer sender of the message
1687  * @param msg the disconnect message
1688  */
1689 void
1690 GST_neighbours_handle_disconnect_message (const struct GNUNET_PeerIdentity *peer,
1691                                           const struct GNUNET_MessageHeader *msg)
1692 {
1693   struct NeighbourMapEntry *n;
1694   const struct SessionDisconnectMessage *sdm;
1695   GNUNET_HashCode hc;
1696
1697 #if DEBUG_TRANSPORT
1698   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1699       "Received DISCONNECT message from peer `%s'\n", GNUNET_i2s (peer));
1700 #endif
1701
1702   if (ntohs (msg->size) != sizeof (struct SessionDisconnectMessage))
1703   {
1704     // GNUNET_break_op (0);
1705     GNUNET_STATISTICS_update (GST_stats,
1706                               gettext_noop ("# disconnect messages ignored (old format)"), 1,
1707                               GNUNET_NO);
1708     return;
1709   }
1710   sdm = (const struct SessionDisconnectMessage* ) msg;
1711   n = lookup_neighbour (peer);
1712   if (NULL == n)
1713     return;                     /* gone already */
1714   if (GNUNET_TIME_absolute_ntoh (sdm->timestamp).abs_value <=
1715       n->connect_ts.abs_value)
1716   {
1717     GNUNET_STATISTICS_update (GST_stats,
1718                               gettext_noop ("# disconnect messages ignored (timestamp)"), 1,
1719                               GNUNET_NO);
1720     return;
1721   }
1722   GNUNET_CRYPTO_hash (&sdm->public_key,
1723                       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1724                       &hc);
1725   if (0 != memcmp (peer,
1726                    &hc,
1727                    sizeof (struct GNUNET_PeerIdentity)))
1728   {
1729     GNUNET_break_op (0);
1730     return;
1731   }
1732   if (ntohl (sdm->purpose.size) !=
1733       sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) +
1734       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) +
1735       sizeof (struct GNUNET_TIME_AbsoluteNBO))
1736   {
1737     GNUNET_break_op (0);
1738     return;
1739   }
1740   if (GNUNET_OK !=
1741       GNUNET_CRYPTO_rsa_verify (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_DISCONNECT,
1742                                 &sdm->purpose,
1743                                 &sdm->signature,
1744                                 &sdm->public_key))
1745   {
1746     GNUNET_break_op (0);
1747     return;
1748   }
1749   GST_neighbours_force_disconnect (peer);
1750 }
1751
1752 /**
1753  * We received a 'SESSION_CONNECT_ACK' message from the other peer.
1754  * Consider switching to it.
1755  *
1756  * @param message possibly a 'struct SessionConnectMessage' (check format)
1757  * @param peer identity of the peer to switch the address for
1758  * @param plugin_name name of transport that delivered the PONG
1759  * @param address address of the other peer, NULL if other peer
1760  *                       connected to us
1761  * @param address_len number of bytes in address
1762  * @param session session to use (or NULL)
1763  * @param ats performance data
1764  * @param ats_count number of entries in ats
1765   */
1766 void
1767 GST_neighbours_handle_connect_ack (const struct GNUNET_MessageHeader *message,
1768                                const struct GNUNET_PeerIdentity *peer,
1769                                const char *plugin_name,
1770                                const char *sender_address, uint16_t sender_address_len,
1771                                struct Session *session,
1772                                const struct GNUNET_ATS_Information *ats,
1773                                uint32_t ats_count)
1774 {
1775   const struct SessionConnectMessage *scm;
1776   struct QuotaSetMessage q_msg;
1777   struct GNUNET_MessageHeader msg;
1778   struct GNUNET_TIME_Absolute ts;
1779   struct NeighbourMapEntry *n;
1780   size_t msg_len;
1781   size_t ret;
1782
1783 #if DEBUG_TRANSPORT
1784 #endif
1785   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1786       "Received CONNECT_ACK message from peer `%s'\n", GNUNET_i2s (peer));
1787
1788
1789   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
1790   {
1791     GNUNET_break_op (0);
1792     return;
1793   }
1794
1795   scm = (const struct SessionConnectMessage *) message;
1796   GNUNET_break_op (ntohl (scm->reserved) == 0);
1797   ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
1798   n = lookup_neighbour (peer);
1799   if (NULL == n)
1800     n = setup_neighbour (peer);
1801 /*
1802   if (n->state != S_CONNECT_SENT)
1803   {
1804     GNUNET_break (0);
1805     send_disconnect(n);
1806     return;
1807   }
1808 */
1809   if (NULL != session)
1810     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1811                      "transport-ats",
1812                      "Giving ATS session %p of plugin %s for peer %s\n",
1813                      session,
1814                      plugin_name,
1815                      GNUNET_i2s (peer));
1816   GNUNET_ATS_address_update (GST_ats,
1817                              peer,
1818                              plugin_name, sender_address, sender_address_len,
1819                              session, ats, ats_count);
1820
1821   change_state (n, S_CONNECTED);
1822
1823 #if DEBUG_TRANSPORT
1824   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1825               "Setting inbound quota of %u for peer `%s' to \n",
1826               ntohl (n->bandwidth_in.value__), GNUNET_i2s (&n->id));
1827 #endif
1828   GST_neighbours_set_incoming_quota(&n->id, n->bandwidth_in);
1829
1830   n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
1831                                                       &neighbour_keepalive_task,
1832                                                       n);
1833   /* send ACK (ACK)*/
1834   msg_len =  sizeof (msg);
1835   msg.size = htons (msg_len);
1836   msg.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SESSION_ACK);
1837
1838   ret = send_with_plugin (&n->id, (const char *) &msg, msg_len, 0,
1839       GNUNET_TIME_UNIT_FOREVER_REL,
1840       n->session, n->plugin_name, n->addr, n->addrlen,
1841       GNUNET_YES, NULL, NULL);
1842
1843   if (ret == GNUNET_SYSERR)
1844     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1845               "Failed to send SESSION_ACK to `%4s' using plugin `%s' address '%s' session %X\n",
1846               GNUNET_i2s (&n->id), n->plugin_name,
1847               (n->addrlen == 0) ? "<inbound>" : GST_plugins_a2s (n->plugin_name,
1848                                                                  n->addr,
1849                                                                  n->addrlen),
1850               n->session);
1851
1852   neighbours_connected++;
1853   GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# peers connected"), 1,
1854                             GNUNET_NO);
1855   connect_notify_cb (callback_cls, &n->id, ats, ats_count);
1856
1857 #if DEBUG_TRANSPORT
1858   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1859               "Sending outbound quota of %u Bps for peer `%s' to all clients\n",
1860               ntohl (n->bandwidth_out.value__), GNUNET_i2s (peer));
1861 #endif
1862   q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
1863   q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
1864   q_msg.quota = n->bandwidth_out;
1865   q_msg.peer = (*peer);
1866   GST_clients_broadcast (&q_msg.header, GNUNET_NO);
1867 }
1868
1869 void
1870 GST_neighbours_handle_ack (const struct GNUNET_MessageHeader *message,
1871     const struct GNUNET_PeerIdentity *peer,
1872     const char *plugin_name,
1873     const char *sender_address, uint16_t sender_address_len,
1874     struct Session *session,
1875     const struct GNUNET_ATS_Information *ats,
1876     uint32_t ats_count)
1877 {
1878   struct NeighbourMapEntry *n;
1879   struct QuotaSetMessage q_msg;
1880
1881 #if DEBUG_TRANSPORT
1882   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1883       "Received ACK message from peer `%s'\n", GNUNET_i2s (peer));
1884 #endif
1885
1886   if (ntohs (message->size) != sizeof (struct GNUNET_MessageHeader))
1887   {
1888     GNUNET_break_op (0);
1889     return;
1890   }
1891
1892   n = lookup_neighbour (peer);
1893   if (NULL == n)
1894   {
1895     send_disconnect(n);
1896     GNUNET_break (0);
1897   }
1898 // FIXME check this
1899 //  if (n->state != S_CONNECT_RECV)
1900 /*  if (is_connecting(n))
1901   {
1902     send_disconnect (n);
1903     change_state (n, S_DISCONNECT);
1904     GNUNET_break (0);
1905     return;
1906   }
1907 */
1908   if (is_connected(n))
1909     return;
1910
1911   if (NULL != session)
1912     GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1913                      "transport-ats",
1914                      "Giving ATS session %p of plugin %s for peer %s\n",
1915                      session,
1916                      plugin_name,
1917                      GNUNET_i2s (peer));
1918   GNUNET_ATS_address_update (GST_ats,
1919                              peer,
1920                              plugin_name, sender_address, sender_address_len,
1921                              session, ats, ats_count);
1922
1923   change_state (n, S_CONNECTED);
1924
1925   GST_neighbours_set_incoming_quota(&n->id, n->bandwidth_in);
1926
1927   n->keepalive_task = GNUNET_SCHEDULER_add_delayed (KEEPALIVE_FREQUENCY,
1928                                                       &neighbour_keepalive_task,
1929                                                       n);
1930
1931   neighbours_connected++;
1932   GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# peers connected"), 1,
1933                             GNUNET_NO);
1934   connect_notify_cb (callback_cls, &n->id, ats, ats_count);
1935
1936 #if DEBUG_TRANSPORT
1937   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1938               "Sending outbound quota of %u Bps for peer `%s' to all clients\n",
1939               ntohl (n->bandwidth_out.value__), GNUNET_i2s (peer));
1940 #endif
1941
1942   q_msg.header.size = htons (sizeof (struct QuotaSetMessage));
1943   q_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
1944   q_msg.quota = n->bandwidth_out;
1945   q_msg.peer = (*peer);
1946   GST_clients_broadcast (&q_msg.header, GNUNET_NO);
1947 }
1948
1949 struct BlackListCheckContext
1950 {
1951   struct GNUNET_ATS_Information *ats;
1952
1953   uint32_t ats_count;
1954
1955   struct Session *session;
1956
1957   char *sender_address;
1958
1959   uint16_t sender_address_len;
1960
1961   char *plugin_name;
1962
1963   struct GNUNET_TIME_Absolute ts;
1964 };
1965
1966
1967 static void
1968 handle_connect_blacklist_cont (void *cls,
1969                                const struct GNUNET_PeerIdentity
1970                                * peer, int result)
1971 {
1972   struct NeighbourMapEntry *n;
1973   struct BlackListCheckContext * bcc = cls;
1974
1975 #if DEBUG_TRANSPORT
1976   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1977       "Blacklist check due to CONNECT message: `%s'\n", GNUNET_i2s (peer), (result == GNUNET_OK) ? "ALLOWED" : "FORBIDDEN");
1978 #endif
1979
1980   /* not allowed */
1981   if (GNUNET_OK != result)
1982   {
1983     GNUNET_free (bcc);
1984     return;
1985   }
1986
1987   n = lookup_neighbour (peer);
1988   if (NULL == n)
1989     n = setup_neighbour (peer);
1990
1991   if (bcc->ts.abs_value > n->connect_ts.abs_value)
1992   {
1993     if (NULL != bcc->session)
1994       GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1995                        "transport-ats",
1996                        "Giving ATS session %p of plugin %s address `%s' for peer %s\n",
1997                        bcc->session,
1998                        bcc->plugin_name,
1999                        GST_plugins_a2s (bcc->plugin_name, bcc->sender_address, bcc->sender_address_len),
2000                        GNUNET_i2s (peer));
2001     GNUNET_ATS_address_update (GST_ats,
2002                                peer,
2003                                bcc->plugin_name, bcc->sender_address, bcc->sender_address_len,
2004                                bcc->session, bcc->ats, bcc->ats_count);
2005     n->connect_ts = bcc->ts;
2006   }
2007
2008   GNUNET_free (bcc);
2009 /*
2010   if (n->state != S_NOT_CONNECTED)
2011     return;*/
2012   change_state (n, S_CONNECT_RECV);
2013
2014   /* Ask ATS for an address to connect via that address */
2015   if (n->ats_suggest != GNUNET_SCHEDULER_NO_TASK)
2016     GNUNET_SCHEDULER_cancel(n->ats_suggest);
2017   n->ats_suggest = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS, ats_suggest_cancel, n);
2018   GNUNET_ATS_suggest_address(GST_ats, peer);
2019 }
2020
2021 /**
2022  * We received a 'SESSION_CONNECT' message from the other peer.
2023  * Consider switching to it.
2024  *
2025  * @param message possibly a 'struct SessionConnectMessage' (check format)
2026  * @param peer identity of the peer to switch the address for
2027  * @param plugin_name name of transport that delivered the PONG
2028  * @param address address of the other peer, NULL if other peer
2029  *                       connected to us
2030  * @param address_len number of bytes in address
2031  * @param session session to use (or NULL)
2032  * @param ats performance data
2033  * @param ats_count number of entries in ats (excluding 0-termination)
2034   */
2035 void
2036 GST_neighbours_handle_connect (const struct GNUNET_MessageHeader *message,
2037                                const struct GNUNET_PeerIdentity *peer,
2038                                const char *plugin_name,
2039                                const char *sender_address, uint16_t sender_address_len,
2040                                struct Session *session,
2041                                const struct GNUNET_ATS_Information *ats,
2042                                uint32_t ats_count)
2043 {
2044   const struct SessionConnectMessage *scm;
2045   struct NeighbourMapEntry * n;
2046   struct BlackListCheckContext * bcc = NULL;
2047
2048 #if DEBUG_TRANSPORT
2049 #endif
2050   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2051       "Received CONNECT message from peer `%s'\n", GNUNET_i2s (peer));
2052
2053
2054   if (ntohs (message->size) != sizeof (struct SessionConnectMessage))
2055   {
2056     GNUNET_break_op (0);
2057     return;
2058   }
2059
2060   scm = (const struct SessionConnectMessage *) message;
2061   GNUNET_break_op (ntohl (scm->reserved) == 0);
2062
2063   n = lookup_neighbour(peer);
2064   if (n != NULL)
2065   {
2066     /* connected peer switches addresses */
2067     if (is_connected(n))
2068     {
2069       GNUNET_ATS_address_update(GST_ats, peer, plugin_name, sender_address, sender_address_len, session, ats, ats_count);
2070       return;
2071     }
2072   }
2073
2074   /* we are not connected to this peer */
2075   /* do blacklist check*/
2076   bcc = GNUNET_malloc (sizeof (struct BlackListCheckContext) +
2077             sizeof (struct GNUNET_ATS_Information) * ats_count +
2078             sender_address_len +
2079             strlen (plugin_name)+1);
2080
2081   bcc->ts = GNUNET_TIME_absolute_ntoh (scm->timestamp);
2082
2083   bcc->ats_count = ats_count;
2084   bcc->sender_address_len = sender_address_len;
2085   bcc->session = session;
2086
2087   bcc->ats = (struct GNUNET_ATS_Information *) &bcc[1];
2088   memcpy (bcc->ats, ats,sizeof (struct GNUNET_ATS_Information) * ats_count );
2089
2090   bcc->sender_address = (char *) &bcc->ats[ats_count];
2091   memcpy (bcc->sender_address, sender_address , sender_address_len);
2092
2093   bcc->plugin_name = &bcc->sender_address[sender_address_len];
2094   strcpy (bcc->plugin_name, plugin_name);
2095
2096   GST_blacklist_test_allowed (peer, plugin_name, handle_connect_blacklist_cont, bcc);
2097 }
2098
2099
2100 /* end of file gnunet-service-transport_neighbours.c */