src/transport/gnunet-service-transport.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009 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 2, 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.c
  23  * @brief low-level P2P messaging
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - if we do not receive an ACK in response to our
  28  *   HELLO, retransmit HELLO!
  29  */
  30 #include "platform.h"
  31 #include "gnunet_client_lib.h"
  32 #include "gnunet_constants.h"
  33 #include "gnunet_getopt_lib.h"
  34 #include "gnunet_hello_lib.h"
  35 #include "gnunet_os_lib.h"
  36 #include "gnunet_peerinfo_service.h"
  37 #include "gnunet_plugin_lib.h"
  38 #include "gnunet_protocols.h"
  39 #include "gnunet_service_lib.h"
  40 #include "gnunet_signatures.h"
  41 #include "plugin_transport.h"
  42 #include "transport.h"
  43
  44 /**
  45  * How many messages can we have pending for a given client process
  46  * before we start to drop incoming messages?  We typically should
  47  * have only one client and so this would be the primary buffer for
  48  * messages, so the number should be chosen rather generously.
  49  *
  50  * The expectation here is that most of the time the queue is large
  51  * enough so that a drop is virtually never required.
  52  */
  53 #define MAX_PENDING 128
  54
  55 /**
  56  * How often should we try to reconnect to a peer using a particular
  57  * transport plugin before giving up?  Note that the plugin may be
  58  * added back to the list after PLUGIN_RETRY_FREQUENCY expires.
  59  */
  60 #define MAX_CONNECT_RETRY 3
  61
  62 /**
  63  * How often must a peer violate bandwidth quotas before we start
  64  * to simply drop its messages?
  65  */
  66 #define QUOTA_VIOLATION_DROP_THRESHOLD 100
  67
  68 /**
  69  * How long until a HELLO verification attempt should time out?
  70  * Must be rather small, otherwise a partially successful HELLO
  71  * validation (some addresses working) might not be available
  72  * before a client's request for a connection fails for good.
  73  * Besides, if a single request to an address takes a long time,
  74  * then the peer is unlikely worthwhile anyway.
  75  */
  76 #define HELLO_VERIFICATION_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 3)
  77
  78 /**
  79  * How often do we re-add (cheaper) plugins to our list of plugins
  80  * to try for a given connected peer?
  81  */
  82 #define PLUGIN_RETRY_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 15)
  83
  84 /**
  85  * After how long do we expire an address in a HELLO
  86  * that we just validated?  This value is also used
  87  * for our own addresses when we create a HELLO.
  88  */
  89 #define HELLO_ADDRESS_EXPIRATION GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 12)
  90
  91
  92 /**
  93  * Entry in linked list of network addresses.
  94  */
  95 struct AddressList
  96 {
  97   /**
  98    * This is a linked list.
  99    */
 100   struct AddressList *next;
 101
 102   /**
 103    * The address, actually a pointer to the end
 104    * of this struct.  Do not free!
 105    */
 106   void *addr;
 107
 108   /**
 109    * How long until we auto-expire this address (unless it is
 110    * re-confirmed by the transport)?
 111    */
 112   struct GNUNET_TIME_Absolute expires;
 113
 114   /**
 115    * Length of addr.
 116    */
 117   size_t addrlen;
 118
 119 };
 120
 121
 122 /**
 123  * Entry in linked list of all of our plugins.
 124  */
 125 struct TransportPlugin
 126 {
 127
 128   /**
 129    * This is a linked list.
 130    */
 131   struct TransportPlugin *next;
 132
 133   /**
 134    * API of the transport as returned by the plugin's
 135    * initialization function.
 136    */
 137   struct GNUNET_TRANSPORT_PluginFunctions *api;
 138
 139   /**
 140    * Short name for the plugin (i.e. "tcp").
 141    */
 142   char *short_name;
 143
 144   /**
 145    * Name of the library (i.e. "gnunet_plugin_transport_tcp").
 146    */
 147   char *lib_name;
 148
 149   /**
 150    * List of our known addresses for this transport.
 151    */
 152   struct AddressList *addresses;
 153
 154   /**
 155    * Environment this transport service is using
 156    * for this plugin.
 157    */
 158   struct GNUNET_TRANSPORT_PluginEnvironment env;
 159
 160   /**
 161    * ID of task that is used to clean up expired addresses.
 162    */
 163   GNUNET_SCHEDULER_TaskIdentifier address_update_task;
 164
 165
 166   /**
 167    * Set to GNUNET_YES if we need to scrap the existing
 168    * list of "addresses" and start fresh when we receive
 169    * the next address update from a transport.  Set to
 170    * GNUNET_NO if we should just add the new address
 171    * to the list and wait for the commit call.
 172    */
 173   int rebuild;
 174 };
 175
 176 struct NeighbourList;
 177
 178 /**
 179  * For each neighbour we keep a list of messages
 180  * that we still want to transmit to the neighbour.
 181  */
 182 struct MessageQueue
 183 {
 184
 185   /**
 186    * This is a linked list.
 187    */
 188   struct MessageQueue *next;
 189
 190   /**
 191    * The message we want to transmit.
 192    */
 193   struct GNUNET_MessageHeader *message;
 194
 195   /**
 196    * Client responsible for queueing the message;
 197    * used to check that a client has not two messages
 198    * pending for the same target.  Can be NULL.
 199    */
 200   struct TransportClient *client;
 201
 202   /**
 203    * Neighbour this entry belongs to.
 204    */
 205   struct NeighbourList *neighbour;
 206
 207   /**
 208    * Plugin that we used for the transmission.
 209    * NULL until we scheduled a transmission.
 210    */
 211   struct TransportPlugin *plugin;
 212
 213   /**
 214    * Internal message of the transport system that should not be
 215    * included in the usual SEND-SEND_OK transmission confirmation
 216    * traffic management scheme.  Typically, "internal_msg" will
 217    * be set whenever "client" is NULL (but it is not strictly
 218    * required).
 219    */
 220   int internal_msg;
 221
 222   /**
 223    * How important is the message?
 224    */
 225   unsigned int priority;
 226
 227 };
 228
 229
 230 /**
 231  * For a given Neighbour, which plugins are available
 232  * to talk to this peer and what are their costs?
 233  */
 234 struct ReadyList
 235 {
 236
 237   /**
 238    * This is a linked list.
 239    */
 240   struct ReadyList *next;
 241
 242   /**
 243    * Which of our transport plugins does this entry
 244    * represent?
 245    */
 246   struct TransportPlugin *plugin;
 247
 248   /**
 249    * Neighbour this entry belongs to.
 250    */
 251   struct NeighbourList *neighbour;
 252
 253   /**
 254    * Opaque handle (specific to the plugin) for the
 255    * connection to our target; can be NULL.
 256    */
 257   void *plugin_handle;
 258
 259   /**
 260    * What was the last latency observed for this plugin
 261    * and peer?  Invalid if connected is GNUNET_NO.
 262    */
 263   struct GNUNET_TIME_Relative latency;
 264
 265   /**
 266    * If we did not successfully transmit a message to the
 267    * given peer via this connection during the specified
 268    * time, we should consider the connection to be dead.
 269    * This is used in the case that a TCP transport simply
 270    * stalls writing to the stream but does not formerly
 271    * get a signal that the other peer died.
 272    */
 273   struct GNUNET_TIME_Absolute timeout;
 274
 275   /**
 276    * Is this plugin currently connected?  The first time
 277    * we transmit or send data to a peer via a particular
 278    * plugin, we set this to GNUNET_YES.  If we later get
 279    * an error (disconnect notification or transmission
 280    * failure), we set it back to GNUNET_NO.  Each time the
 281    * value is set to GNUNET_YES, we increment the
 282    * "connect_attempts" counter.  If that one reaches a
 283    * particular threshold, we consider the plugin to not
 284    * be working properly at this time for the given peer
 285    * and remove it from the eligible list.
 286    */
 287   int connected;
 288
 289   /**
 290    * How often have we tried to connect using this plugin?
 291    */
 292   unsigned int connect_attempts;
 293
 294   /**
 295    * Is this plugin ready to transmit to the specific
 296    * target?  GNUNET_NO if not.  Initially, all plugins
 297    * are marked ready.  If a transmission is in progress,
 298    * "transmit_ready" is set to GNUNET_NO.
 299    */
 300   int transmit_ready;
 301
 302 };
 303
 304
 305 /**
 306  * Entry in linked list of all of our current neighbours.
 307  */
 308 struct NeighbourList
 309 {
 310
 311   /**
 312    * This is a linked list.
 313    */
 314   struct NeighbourList *next;
 315
 316   /**
 317    * Which of our transports is connected to this peer
 318    * and what is their status?
 319    */
 320   struct ReadyList *plugins;
 321
 322   /**
 323    * List of messages we would like to send to this peer;
 324    * must contain at most one message per client.
 325    */
 326   struct MessageQueue *messages;
 327
 328   /**
 329    * Identity of this neighbour.
 330    */
 331   struct GNUNET_PeerIdentity id;
 332
 333   /**
 334    * ID of task scheduled to run when this peer is about to
 335    * time out (will free resources associated with the peer).
 336    */
 337   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 338
 339   /**
 340    * How long until we should consider this peer dead
 341    * (if we don't receive another message in the
 342    * meantime)?
 343    */
 344   struct GNUNET_TIME_Absolute peer_timeout;
 345
 346   /**
 347    * At what time did we reset last_received last?
 348    */
 349   struct GNUNET_TIME_Absolute last_quota_update;
 350
 351   /**
 352    * At what time should we try to again add plugins to
 353    * our ready list?
 354    */
 355   struct GNUNET_TIME_Absolute retry_plugins_time;
 356
 357   /**
 358    * How many bytes have we received since the "last_quota_update"
 359    * timestamp?
 360    */
 361   uint64_t last_received;
 362
 363   /**
 364    * Global quota for inbound traffic for the neighbour in bytes/ms.
 365    */
 366   uint32_t quota_in;
 367
 368   /**
 369    * How often has the other peer (recently) violated the
 370    * inbound traffic limit?  Incremented by 10 per violation,
 371    * decremented by 1 per non-violation (for each
 372    * time interval).
 373    */
 374   unsigned int quota_violation_count;
 375
 376   /**
 377    * Have we seen an ACK from this neighbour in the past?
 378    * (used to make up a fake ACK for clients connecting after
 379    * the neighbour connected to us).
 380    */
 381   int saw_ack;
 382
 383 };
 384
 385
 386 /**
 387  * Linked list of messages to be transmitted to
 388  * the client.  Each entry is followed by the
 389  * actual message.
 390  */
 391 struct ClientMessageQueueEntry
 392 {
 393   /**
 394    * This is a linked list.
 395    */
 396   struct ClientMessageQueueEntry *next;
 397 };
 398
 399
 400 /**
 401  * Client connected to the transport service.
 402  */
 403 struct TransportClient
 404 {
 405
 406   /**
 407    * This is a linked list.
 408    */
 409   struct TransportClient *next;
 410
 411   /**
 412    * Handle to the client.
 413    */
 414   struct GNUNET_SERVER_Client *client;
 415
 416   /**
 417    * Linked list of messages yet to be transmitted to
 418    * the client.
 419    */
 420   struct ClientMessageQueueEntry *message_queue_head;
 421
 422   /**
 423    * Tail of linked list of messages yet to be transmitted to the
 424    * client.
 425    */
 426   struct ClientMessageQueueEntry *message_queue_tail;
 427
 428   /**
 429    * Is a call to "transmit_send_continuation" pending?  If so, we
 430    * must not free this struct (even if the corresponding client
 431    * disconnects) and instead only remove it from the linked list and
 432    * set the "client" field to NULL.
 433    */
 434   int tcs_pending;
 435
 436   /**
 437    * Length of the list of messages pending for this client.
 438    */
 439   unsigned int message_count;
 440
 441 };
 442
 443
 444 /**
 445  * For each HELLO, we may have to validate multiple addresses;
 446  * each address gets its own request entry.
 447  */
 448 struct ValidationAddress
 449 {
 450   /**
 451    * This is a linked list.
 452    */
 453   struct ValidationAddress *next;
 454
 455   /**
 456    * Name of the transport.
 457    */
 458   char *transport_name;
 459
 460   /**
 461    * When should this validated address expire?
 462    */
 463   struct GNUNET_TIME_Absolute expiration;
 464
 465   /**
 466    * Length of the address we are validating.
 467    */
 468   size_t addr_len;
 469
 470   /**
 471    * Challenge number we used.
 472    */
 473   uint32_t challenge;
 474
 475   /**
 476    * Set to GNUNET_YES if the challenge was met,
 477    * GNUNET_SYSERR if we know it failed, GNUNET_NO
 478    * if we are waiting on a response.
 479    */
 480   int ok;
 481 };
 482
 483
 484 /**
 485  * Entry in linked list of all HELLOs awaiting validation.
 486  */
 487 struct ValidationList
 488 {
 489
 490   /**
 491    * This is a linked list.
 492    */
 493   struct ValidationList *next;
 494
 495   /**
 496    * Linked list with one entry per address from the HELLO
 497    * that needs to be validated.
 498    */
 499   struct ValidationAddress *addresses;
 500
 501   /**
 502    * The public key of the peer.
 503    */
 504   struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded publicKey;
 505
 506   /**
 507    * When does this record time-out? (assuming the
 508    * challenge goes unanswered)
 509    */
 510   struct GNUNET_TIME_Absolute timeout;
 511
 512 };
 513
 514
 515 /**
 516  * HELLOs awaiting validation.
 517  */
 518 static struct ValidationList *pending_validations;
 519
 520 /**
 521  * Our HELLO message.
 522  */
 523 static struct GNUNET_HELLO_Message *our_hello;
 524
 525 /**
 526  * "version" of "our_hello".  Used to see if a given
 527  * neighbour has already been sent the latest version
 528  * of our HELLO message.
 529  */
 530 static unsigned int our_hello_version;
 531
 532 /**
 533  * Our public key.
 534  */
 535 static struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded my_public_key;
 536
 537 /**
 538  * Our identity.
 539  */
 540 static struct GNUNET_PeerIdentity my_identity;
 541
 542 /**
 543  * Our private key.
 544  */
 545 static struct GNUNET_CRYPTO_RsaPrivateKey *my_private_key;
 546
 547 /**
 548  * Our scheduler.
 549  */
 550 struct GNUNET_SCHEDULER_Handle *sched;
 551
 552 /**
 553  * Our configuration.
 554  */
 555 const struct GNUNET_CONFIGURATION_Handle *cfg;
 556
 557 /**
 558  * Linked list of all clients to this service.
 559  */
 560 static struct TransportClient *clients;
 561
 562 /**
 563  * All loaded plugins.
 564  */
 565 static struct TransportPlugin *plugins;
 566
 567 /**
 568  * Our server.
 569  */
 570 static struct GNUNET_SERVER_Handle *server;
 571
 572 /**
 573  * All known neighbours and their HELLOs.
 574  */
 575 static struct NeighbourList *neighbours;
 576
 577 /**
 578  * Number of neighbours we'd like to have.
 579  */
 580 static uint32_t max_connect_per_transport;
 581
 582
 583 /**
 584  * Find an entry in the neighbour list for a particular peer.
 585  *
 586  * @return NULL if not found.
 587  */
 588 static struct NeighbourList *
 589 find_neighbour (const struct GNUNET_PeerIdentity *key)
 590 {
 591   struct NeighbourList *head = neighbours;
 592   while ((head != NULL) &&
 593          (0 != memcmp (key, &head->id, sizeof (struct GNUNET_PeerIdentity))))
 594     head = head->next;
 595   return head;
 596 }
 597
 598
 599 /**
 600  * Find an entry in the transport list for a particular transport.
 601  *
 602  * @return NULL if not found.
 603  */
 604 static struct TransportPlugin *
 605 find_transport (const char *short_name)
 606 {
 607   struct TransportPlugin *head = plugins;
 608   while ((head != NULL) && (0 != strcmp (short_name, head->short_name)))
 609     head = head->next;
 610   return head;
 611 }
 612
 613
 614 /**
 615  * Update the quota values for the given neighbour now.
 616  */
 617 static void
 618 update_quota (struct NeighbourList *n)
 619 {
 620   struct GNUNET_TIME_Relative delta;
 621   uint64_t allowed;
 622   uint64_t remaining;
 623
 624   delta = GNUNET_TIME_absolute_get_duration (n->last_quota_update);
 625   if (delta.value < MIN_QUOTA_REFRESH_TIME)
 626     return;                     /* not enough time passed for doing quota update */
 627   allowed = delta.value * n->quota_in;
 628   if (n->last_received < allowed)
 629     {
 630       remaining = allowed - n->last_received;
 631       if (n->quota_in > 0)
 632         remaining /= n->quota_in;
 633       else
 634         remaining = 0;
 635       if (remaining > MAX_BANDWIDTH_CARRY)
 636         remaining = MAX_BANDWIDTH_CARRY;
 637       n->last_received = 0;
 638       n->last_quota_update = GNUNET_TIME_absolute_get ();
 639       n->last_quota_update.value -= remaining;
 640       if (n->quota_violation_count > 0)
 641         n->quota_violation_count--;
 642     }
 643   else
 644     {
 645       n->last_received -= allowed;
 646       n->last_quota_update = GNUNET_TIME_absolute_get ();
 647       if (n->last_received > allowed)
 648         {
 649           /* more than twice the allowed rate! */
 650           n->quota_violation_count += 10;
 651         }
 652     }
 653 }
 654
 655
 656 /**
 657  * Function called to notify a client about the socket
 658  * being ready to queue more data.  "buf" will be
 659  * NULL and "size" zero if the socket was closed for
 660  * writing in the meantime.
 661  *
 662  * @param cls closure
 663  * @param size number of bytes available in buf
 664  * @param buf where the callee should write the message
 665  * @return number of bytes written to buf
 666  */
 667 static size_t
 668 transmit_to_client_callback (void *cls, size_t size, void *buf)
 669 {
 670   struct TransportClient *client = cls;
 671   struct ClientMessageQueueEntry *q;
 672   uint16_t msize;
 673   size_t tsize;
 674   const struct GNUNET_MessageHeader *msg;
 675   struct GNUNET_CONNECTION_TransmitHandle *th;
 676   char *cbuf;
 677
 678   if (buf == NULL)
 679     {
 680       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 681                   "Transmission to client failed, closing connection.\n");
 682       /* fatal error with client, free message queue! */
 683       while (NULL != (q = client->message_queue_head))
 684         {
 685           client->message_queue_head = q->next;
 686           GNUNET_free (q);
 687         }
 688       client->message_queue_tail = NULL;
 689       client->message_count = 0;
 690       return 0;
 691     }
 692   cbuf = buf;
 693   tsize = 0;
 694   while (NULL != (q = client->message_queue_head))
 695     {
 696       msg = (const struct GNUNET_MessageHeader *) &q[1];
 697       msize = ntohs (msg->size);
 698       if (msize + tsize > size)
 699         break;
 700 #if DEBUG_TRANSPORT
 701       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 702                   "Transmitting message of type %u to client.\n",
 703                   ntohs (msg->type));
 704 #endif
 705       client->message_queue_head = q->next;
 706       if (q->next == NULL)
 707         client->message_queue_tail = NULL;
 708       memcpy (&cbuf[tsize], msg, msize);
 709       tsize += msize;
 710       GNUNET_free (q);
 711       client->message_count--;
 712     }
 713   GNUNET_assert (tsize > 0);
 714   if (NULL != q)
 715     {
 716       th = GNUNET_SERVER_notify_transmit_ready (client->client,
 717                                                 msize,
 718                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 719                                                 &transmit_to_client_callback,
 720                                                 client);
 721       GNUNET_assert (th != NULL);
 722     }
 723   return tsize;
 724 }
 725
 726
 727 /**
 728  * Send the specified message to the specified client.  Since multiple
 729  * messages may be pending for the same client at a time, this code
 730  * makes sure that no message is lost.
 731  *
 732  * @param client client to transmit the message to
 733  * @param msg the message to send
 734  * @param may_drop can this message be dropped if the
 735  *        message queue for this client is getting far too large?
 736  */
 737 static void
 738 transmit_to_client (struct TransportClient *client,
 739                     const struct GNUNET_MessageHeader *msg, int may_drop)
 740 {
 741   struct ClientMessageQueueEntry *q;
 742   uint16_t msize;
 743   struct GNUNET_CONNECTION_TransmitHandle *th;
 744
 745   if ((client->message_count >= MAX_PENDING) && (GNUNET_YES == may_drop))
 746     {
 747       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 748                   _
 749                   ("Dropping message, have %u messages pending (%u is the soft limit)\n"),
 750                   client->message_count, MAX_PENDING);
 751       /* TODO: call to statistics... */
 752       return;
 753     }
 754   client->message_count++;
 755   msize = ntohs (msg->size);
 756   q = GNUNET_malloc (sizeof (struct ClientMessageQueueEntry) + msize);
 757   memcpy (&q[1], msg, msize);
 758   /* append to message queue */
 759   if (client->message_queue_tail == NULL)
 760     {
 761       client->message_queue_tail = q;
 762     }
 763   else
 764     {
 765       client->message_queue_tail->next = q;
 766       client->message_queue_tail = q;
 767     }
 768   if (client->message_queue_head == NULL)
 769     {
 770       client->message_queue_head = q;
 771       th = GNUNET_SERVER_notify_transmit_ready (client->client,
 772                                                 msize,
 773                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 774                                                 &transmit_to_client_callback,
 775                                                 client);
 776       GNUNET_assert (th != NULL);
 777     }
 778 }
 779
 780
 781 /**
 782  * Find alternative plugins for communication.
 783  *
 784  * @param neighbour for which neighbour should we try to find
 785  *        more plugins?
 786  */
 787 static void
 788 try_alternative_plugins (struct NeighbourList *neighbour)
 789 {
 790   struct ReadyList *rl;
 791
 792   if ((neighbour->plugins != NULL) &&
 793       (neighbour->retry_plugins_time.value >
 794        GNUNET_TIME_absolute_get ().value))
 795     return;                     /* don't try right now */
 796   neighbour->retry_plugins_time
 797     = GNUNET_TIME_relative_to_absolute (PLUGIN_RETRY_FREQUENCY);
 798
 799   rl = neighbour->plugins;
 800   while (rl != NULL)
 801     {
 802       if (rl->connect_attempts > 0)
 803         rl->connect_attempts--; /* amnesty */
 804       rl = rl->next;
 805     }
 806
 807 }
 808
 809
 810 /**
 811  * The peer specified by the given neighbour has timed-out or a plugin
 812  * has disconnected.  We may either need to do nothing (other plugins
 813  * still up), or trigger a full disconnect and clean up.  This
 814  * function updates our state and do the necessary notifications.
 815  * Also notifies our clients that the neighbour is now officially
 816  * gone.
 817  *
 818  * @param n the neighbour list entry for the peer
 819  * @param check should we just check if all plugins
 820  *        disconnected or must we ask all plugins to
 821  *        disconnect?
 822  */
 823 static void
 824 disconnect_neighbour (struct NeighbourList *n,
 825                       int check);
 826
 827
 828 /**
 829  * Check the ready list for the given neighbour and
 830  * if a plugin is ready for transmission (and if we
 831  * have a message), do so!
 832  *
 833  * @param neighbour target peer for which to check the plugins
 834  */
 835 static void
 836 try_transmission_to_peer (struct NeighbourList *neighbour);
 837
 838
 839 /**
 840  * Function called by the GNUNET_TRANSPORT_TransmitFunction
 841  * upon "completion" of a send request.  This tells the API
 842  * that it is now legal to send another message to the given
 843  * peer.
 844  *
 845  * @param cls closure, identifies the entry on the
 846  *            message queue that was transmitted and the
 847  *            client responsible for queueing the message
 848  * @param rl identifies plugin used for the transmission for
 849  *           this neighbour; needs to be re-enabled for
 850  *           future transmissions
 851  * @param target the peer receiving the message
 852  * @param result GNUNET_OK on success, if the transmission
 853  *           failed, we should not tell the client to transmit
 854  *           more messages
 855  */
 856 static void
 857 transmit_send_continuation (void *cls,
 858                             struct ReadyList *rl,
 859                             const struct GNUNET_PeerIdentity *target,
 860                             int result)
 861 {
 862   struct MessageQueue *mq = cls;
 863   struct SendOkMessage send_ok_msg;
 864   struct NeighbourList *n;
 865
 866   GNUNET_assert (mq != NULL);
 867   n = mq->neighbour;
 868   GNUNET_assert (n != NULL);
 869   GNUNET_assert (0 ==
 870                  memcmp (&n->id, target,
 871                          sizeof (struct GNUNET_PeerIdentity)));
 872   if (rl == NULL)
 873     {
 874       rl = n->plugins;
 875       while ((rl != NULL) && (rl->plugin != mq->plugin))
 876         rl = rl->next;
 877       GNUNET_assert (rl != NULL);
 878     }
 879   if (result == GNUNET_OK)
 880     {
 881       rl->timeout = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
 882     }
 883   else
 884     {
 885       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 886                   "Transmission to peer `%s' failed, marking connection as down.\n",
 887                   GNUNET_i2s(target));
 888       rl->connected = GNUNET_NO;
 889       rl->plugin_handle = NULL;
 890     }
 891   if (!mq->internal_msg)
 892     rl->transmit_ready = GNUNET_YES;
 893   if (mq->client != NULL)
 894     {
 895       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 896                   "Notifying client %p about failed transission to peer `%4s'.\n",
 897                   mq->client,
 898                   GNUNET_i2s(target));
 899       send_ok_msg.header.size = htons (sizeof (send_ok_msg));
 900       send_ok_msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
 901       send_ok_msg.success = htonl (result);
 902       send_ok_msg.peer = n->id;
 903       transmit_to_client (mq->client, &send_ok_msg.header, GNUNET_NO);
 904     }
 905   GNUNET_free (mq->message);
 906   GNUNET_free (mq);
 907   /* one plugin just became ready again, try transmitting
 908      another message (if available) */
 909   if (result == GNUNET_OK)
 910     try_transmission_to_peer (n);
 911   else
 912     disconnect_neighbour (n, GNUNET_YES);
 913 }
 914
 915
 916 /**
 917  * Check the ready list for the given neighbour and
 918  * if a plugin is ready for transmission (and if we
 919  * have a message), do so!
 920  */
 921 static void
 922 try_transmission_to_peer (struct NeighbourList *neighbour)
 923 {
 924   struct ReadyList *pos;
 925   struct GNUNET_TIME_Relative min_latency;
 926   struct ReadyList *rl;
 927   struct MessageQueue *mq;
 928   struct GNUNET_TIME_Absolute now;
 929
 930   if (neighbour->messages == NULL)
 931     return;                     /* nothing to do */
 932   try_alternative_plugins (neighbour);
 933   min_latency = GNUNET_TIME_UNIT_FOREVER_REL;
 934   rl = NULL;
 935   mq = neighbour->messages;
 936   now = GNUNET_TIME_absolute_get ();
 937   pos = neighbour->plugins;
 938   while (pos != NULL)
 939     {
 940       /* set plugins that are inactive for a long time back to disconnected */
 941       if ((pos->timeout.value < now.value) && (pos->connected == GNUNET_YES))
 942         {
 943 #if DEBUG_TRANSPORT
 944           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 945                       "Marking long-time inactive connection to `%4s' as down.\n",
 946                       GNUNET_i2s (&neighbour->id));
 947 #endif
 948           pos->connected = GNUNET_NO;
 949         }
 950       if (((GNUNET_YES == pos->transmit_ready) ||
 951            (mq->internal_msg)) &&
 952           (pos->connect_attempts < MAX_CONNECT_RETRY) &&
 953           ((rl == NULL) || (min_latency.value > pos->latency.value)))
 954         {
 955           rl = pos;
 956           min_latency = pos->latency;
 957         }
 958       pos = pos->next;
 959     }
 960   if (rl == NULL)
 961     {
 962 #if DEBUG_TRANSPORT
 963       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 964                   "No plugin ready to transmit message\n");
 965 #endif
 966       return;                   /* nobody ready */
 967     }
 968   if (GNUNET_NO == rl->connected)
 969     {
 970       rl->connect_attempts++;
 971       rl->connected = GNUNET_YES;
 972 #if DEBUG_TRANSPORT
 973   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 974               "Establishing fresh connection with `%4s' via plugin `%s'\n",
 975               GNUNET_i2s (&neighbour->id), rl->plugin->short_name);
 976 #endif
 977     }
 978   neighbour->messages = mq->next;
 979   mq->plugin = rl->plugin;
 980   if (!mq->internal_msg)
 981     rl->transmit_ready = GNUNET_NO;
 982 #if DEBUG_TRANSPORT
 983   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 984               "Giving message of type `%u' for `%4s' to plugin `%s'\n",
 985               ntohs (mq->message->type),
 986               GNUNET_i2s (&neighbour->id), rl->plugin->short_name);
 987 #endif
 988   rl->plugin_handle
 989     = rl->plugin->api->send (rl->plugin->api->cls,
 990                              rl->plugin_handle,
 991                              rl,
 992                              &neighbour->id,
 993                              mq->priority,
 994                              mq->message,
 995                              GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
 996                              &transmit_send_continuation, mq);
 997 }
 998
 999
1000 /**
1001  * Send the specified message to the specified peer.
1002  *
1003  * @param client source of the transmission request (can be NULL)
1004  * @param priority how important is the message
1005  * @param msg message to send
1006  * @param is_internal is this an internal message
1007  * @param neighbour handle to the neighbour for transmission
1008  */
1009 static void
1010 transmit_to_peer (struct TransportClient *client,
1011                   unsigned int priority,
1012                   const struct GNUNET_MessageHeader *msg,
1013                   int is_internal, struct NeighbourList *neighbour)
1014 {
1015   struct MessageQueue *mq;
1016   struct MessageQueue *mqe;
1017   struct GNUNET_MessageHeader *m;
1018
1019 #if DEBUG_TRANSPORT
1020   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1021               _("Sending message of type %u to peer `%4s'\n"),
1022               ntohs (msg->type), GNUNET_i2s (&neighbour->id));
1023 #endif
1024   if (client != NULL)
1025     {
1026       /* check for duplicate submission */
1027       mq = neighbour->messages;
1028       while (NULL != mq)
1029         {
1030           if (mq->client == client)
1031             {
1032               /* client transmitted to same peer twice
1033                  before getting SendOk! */
1034               GNUNET_break (0);
1035               return;
1036             }
1037           mq = mq->next;
1038         }
1039     }
1040   mq = GNUNET_malloc (sizeof (struct MessageQueue));
1041   mq->client = client;
1042   m = GNUNET_malloc (ntohs (msg->size));
1043   memcpy (m, msg, ntohs (msg->size));
1044   mq->message = m;
1045   mq->neighbour = neighbour;
1046   mq->internal_msg = is_internal;
1047   mq->priority = priority;
1048
1049   /* find tail */
1050   mqe = neighbour->messages;
1051   if (mqe != NULL)
1052     while (mqe->next != NULL)
1053       mqe = mqe->next;
1054   if (mqe == NULL)
1055     {
1056       /* new head */
1057       neighbour->messages = mq;
1058       try_transmission_to_peer (neighbour);
1059     }
1060   else
1061     {
1062       /* append */
1063       mqe->next = mq;
1064     }
1065 }
1066
1067
1068 /**
1069  * FIXME: document.
1070  */
1071 struct GeneratorContext
1072 {
1073   struct TransportPlugin *plug_pos;
1074   struct AddressList *addr_pos;
1075   struct GNUNET_TIME_Absolute expiration;
1076 };
1077
1078
1079 /**
1080  * FIXME: document.
1081  */
1082 static size_t
1083 address_generator (void *cls, size_t max, void *buf)
1084 {
1085   struct GeneratorContext *gc = cls;
1086   size_t ret;
1087
1088   while ((gc->addr_pos == NULL) && (gc->plug_pos != NULL))
1089     {
1090       gc->plug_pos = gc->plug_pos->next;
1091       gc->addr_pos = (gc->plug_pos != NULL) ? gc->plug_pos->addresses : NULL;
1092     }
1093   if (NULL == gc->plug_pos)
1094     return 0;
1095   ret = GNUNET_HELLO_add_address (gc->plug_pos->short_name,
1096                                   gc->expiration,
1097                                   gc->addr_pos->addr,
1098                                   gc->addr_pos->addrlen, buf, max);
1099   gc->addr_pos = gc->addr_pos->next;
1100   return ret;
1101 }
1102
1103
1104 /**
1105  * Construct our HELLO message from all of the addresses of
1106  * all of the transports.
1107  */
1108 static void
1109 refresh_hello ()
1110 {
1111   struct GNUNET_HELLO_Message *hello;
1112   struct TransportClient *cpos;
1113   struct NeighbourList *npos;
1114   struct GeneratorContext gc;
1115
1116 #if DEBUG_TRANSPORT
1117   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1118               "Refreshing my `%s'\n",
1119               "HELLO");
1120 #endif
1121   gc.plug_pos = plugins;
1122   gc.addr_pos = plugins != NULL ? plugins->addresses : NULL;
1123   gc.expiration = GNUNET_TIME_relative_to_absolute (HELLO_ADDRESS_EXPIRATION);
1124   hello = GNUNET_HELLO_create (&my_public_key, &address_generator, &gc);
1125   cpos = clients;
1126   while (cpos != NULL)
1127     {
1128       transmit_to_client (cpos,
1129                           (const struct GNUNET_MessageHeader *) hello,
1130                           GNUNET_NO);
1131       cpos = cpos->next;
1132     }
1133
1134   GNUNET_free_non_null (our_hello);
1135   our_hello = hello;
1136   our_hello_version++;
1137   npos = neighbours;
1138   while (npos != NULL)
1139     {
1140 #if DEBUG_TRANSPORT
1141       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1142                   "Transmitting updated `%s' to neighbour `%4s'\n",
1143                   "HELLO",
1144                   GNUNET_i2s(&npos->id));
1145 #endif
1146       transmit_to_peer (NULL, 0,
1147                         (const struct GNUNET_MessageHeader *) our_hello,
1148                         GNUNET_YES, npos);
1149       npos = npos->next;
1150     }
1151 }
1152
1153
1154 /**
1155  * Task used to clean up expired addresses for a plugin.
1156  *
1157  * @param cls closure
1158  * @param tc context
1159  */
1160 static void
1161 expire_address_task (void *cls,
1162                      const struct GNUNET_SCHEDULER_TaskContext *tc);
1163
1164
1165 /**
1166  * Update the list of addresses for this plugin,
1167  * expiring those that are past their expiration date.
1168  *
1169  * @param plugin addresses of which plugin should be recomputed?
1170  * @param fresh set to GNUNET_YES if a new address was added
1171  *        and we need to regenerate the HELLO even if nobody
1172  *        expired
1173  */
1174 static void
1175 update_addresses (struct TransportPlugin *plugin, int fresh)
1176 {
1177   struct GNUNET_TIME_Relative min_remaining;
1178   struct GNUNET_TIME_Relative remaining;
1179   struct GNUNET_TIME_Absolute now;
1180   struct AddressList *pos;
1181   struct AddressList *prev;
1182   struct AddressList *next;
1183   int expired;
1184
1185   if (plugin->address_update_task != GNUNET_SCHEDULER_NO_TASK)
1186     GNUNET_SCHEDULER_cancel (plugin->env.sched, plugin->address_update_task);
1187   plugin->address_update_task = GNUNET_SCHEDULER_NO_TASK;
1188   now = GNUNET_TIME_absolute_get ();
1189   min_remaining = GNUNET_TIME_UNIT_FOREVER_REL;
1190   expired = GNUNET_NO;
1191   prev = NULL;
1192   pos = plugin->addresses;
1193   while (pos != NULL)
1194     {
1195       next = pos->next;
1196       if (pos->expires.value < now.value)
1197         {
1198           expired = GNUNET_YES;
1199           if (prev == NULL)
1200             plugin->addresses = pos->next;
1201           else
1202             prev->next = pos->next;
1203           GNUNET_free (pos);
1204         }
1205       else
1206         {
1207           remaining = GNUNET_TIME_absolute_get_remaining (pos->expires);
1208           if (remaining.value < min_remaining.value)
1209             min_remaining = remaining;
1210           prev = pos;
1211         }
1212       pos = next;
1213     }
1214
1215   if (expired || fresh)
1216     refresh_hello ();
1217   if (min_remaining.value < GNUNET_TIME_UNIT_FOREVER_REL.value)
1218     plugin->address_update_task
1219       = GNUNET_SCHEDULER_add_delayed (plugin->env.sched,
1220                                       GNUNET_NO,
1221                                       GNUNET_SCHEDULER_PRIORITY_IDLE,
1222                                       GNUNET_SCHEDULER_NO_TASK,
1223                                       min_remaining,
1224                                       &expire_address_task, plugin);
1225
1226 }
1227
1228
1229 /**
1230  * Task used to clean up expired addresses for a plugin.
1231  *
1232  * @param cls closure
1233  * @param tc context
1234  */
1235 static void
1236 expire_address_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1237 {
1238   struct TransportPlugin *plugin = cls;
1239   plugin->address_update_task = GNUNET_SCHEDULER_NO_TASK;
1240   update_addresses (plugin, GNUNET_NO);
1241 }
1242
1243
1244 /**
1245  * Function that must be called by each plugin to notify the
1246  * transport service about the addresses under which the transport
1247  * provided by the plugin can be reached.
1248  *
1249  * @param cls closure
1250  * @param name name of the transport that generated the address
1251  * @param addr one of the addresses of the host, NULL for the last address
1252  *        the specific address format depends on the transport
1253  * @param addrlen length of the address
1254  * @param expires when should this address automatically expire?
1255  */
1256 static void
1257 plugin_env_notify_address (void *cls,
1258                            const char *name,
1259                            const void *addr,
1260                            size_t addrlen,
1261                            struct GNUNET_TIME_Relative expires)
1262 {
1263   struct TransportPlugin *p = cls;
1264   struct AddressList *al;
1265   struct GNUNET_TIME_Absolute abex;
1266
1267   abex = GNUNET_TIME_relative_to_absolute (expires);
1268   GNUNET_assert (p == find_transport (name));
1269
1270   al = p->addresses;
1271   while (al != NULL)
1272     {
1273       if ((addrlen == al->addrlen) && (0 == memcmp (addr, &al[1], addrlen)))
1274         {
1275           if (al->expires.value < abex.value)
1276             al->expires = abex;
1277           return;
1278         }
1279       al = al->next;
1280     }
1281 #if DEBUG_TRANSPORT
1282   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1283               "Plugin `%s' informs us about a new address `%s'\n", name,
1284               GNUNET_a2s(addr, addrlen));
1285 #endif
1286   al = GNUNET_malloc (sizeof (struct AddressList) + addrlen);
1287   al->addr = &al[1];
1288   al->next = p->addresses;
1289   p->addresses = al;
1290   al->expires = abex;
1291   al->addrlen = addrlen;
1292   memcpy (&al[1], addr, addrlen);
1293   update_addresses (p, GNUNET_YES);
1294 }
1295
1296
1297 /**
1298  * FIXME: document.
1299  */
1300 struct LookupHelloContext
1301 {
1302   GNUNET_TRANSPORT_AddressCallback iterator;
1303
1304   void *iterator_cls;
1305 };
1306
1307
1308 /**
1309  * FIXME: document.
1310  */
1311 static int
1312 lookup_address_callback (void *cls,
1313                          const char *tname,
1314                          struct GNUNET_TIME_Absolute expiration,
1315                          const void *addr, size_t addrlen)
1316 {
1317   struct LookupHelloContext *lhc = cls;
1318   lhc->iterator (lhc->iterator_cls, tname, addr, addrlen);
1319   return GNUNET_OK;
1320 }
1321
1322
1323 /**
1324  * FIXME: document.
1325  */
1326 static void
1327 lookup_hello_callback (void *cls,
1328                        const struct GNUNET_PeerIdentity *peer,
1329                        const struct GNUNET_HELLO_Message *h, uint32_t trust)
1330 {
1331   struct LookupHelloContext *lhc = cls;
1332
1333   if (peer == NULL)
1334     {
1335       lhc->iterator (lhc->iterator_cls, NULL, NULL, 0);
1336       GNUNET_free (lhc);
1337       return;
1338     }
1339   if (h == NULL)
1340     return;
1341   GNUNET_HELLO_iterate_addresses (h,
1342                                   GNUNET_NO, &lookup_address_callback, lhc);
1343 }
1344
1345
1346 /**
1347  * Function that allows a transport to query the known
1348  * network addresses for a given peer.
1349  *
1350  * @param cls closure
1351  * @param timeout after how long should we time out?
1352  * @param target which peer are we looking for?
1353  * @param iter function to call for each known address
1354  * @param iter_cls closure for iter
1355  */
1356 static void
1357 plugin_env_lookup_address (void *cls,
1358                            struct GNUNET_TIME_Relative timeout,
1359                            const struct GNUNET_PeerIdentity *target,
1360                            GNUNET_TRANSPORT_AddressCallback iter,
1361                            void *iter_cls)
1362 {
1363   struct LookupHelloContext *lhc;
1364
1365   lhc = GNUNET_malloc (sizeof (struct LookupHelloContext));
1366   lhc->iterator = iter;
1367   lhc->iterator_cls = iter_cls;
1368   GNUNET_PEERINFO_for_all (cfg,
1369                            sched,
1370                            target, 0, timeout, &lookup_hello_callback, &lhc);
1371 }
1372
1373
1374 /**
1375  * Notify all of our clients about a peer connecting.
1376  */
1377 static void
1378 notify_clients_connect (const struct GNUNET_PeerIdentity *peer,
1379                         struct GNUNET_TIME_Relative latency)
1380 {
1381   struct ConnectInfoMessage cim;
1382   struct TransportClient *cpos;
1383
1384 #if DEBUG_TRANSPORT
1385   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1386               "Informing clients about peer `%4s' connecting to us\n",
1387               GNUNET_i2s (peer));
1388 #endif
1389   cim.header.size = htons (sizeof (struct ConnectInfoMessage));
1390   cim.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
1391   cim.quota_out = htonl (GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT / (60*1000));
1392   cim.latency = GNUNET_TIME_relative_hton (latency);
1393   memcpy (&cim.id, peer, sizeof (struct GNUNET_PeerIdentity));
1394   cpos = clients;
1395   while (cpos != NULL)
1396     {
1397       transmit_to_client (cpos, &cim.header, GNUNET_NO);
1398       cpos = cpos->next;
1399     }
1400 }
1401
1402
1403 /**
1404  * Notify all of our clients about a peer disconnecting.
1405  */
1406 static void
1407 notify_clients_disconnect (const struct GNUNET_PeerIdentity *peer)
1408 {
1409   struct DisconnectInfoMessage dim;
1410   struct TransportClient *cpos;
1411
1412 #if DEBUG_TRANSPORT
1413   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1414               "Informing clients about peer `%4s' disconnecting\n",
1415               GNUNET_i2s (peer));
1416 #endif
1417   dim.header.size = htons (sizeof (struct DisconnectInfoMessage));
1418   dim.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
1419   dim.reserved = htonl (0);
1420   memcpy (&dim.peer, peer, sizeof (struct GNUNET_PeerIdentity));
1421   cpos = clients;
1422   while (cpos != NULL)
1423     {
1424       transmit_to_client (cpos, &dim.header, GNUNET_NO);
1425       cpos = cpos->next;
1426     }
1427 }
1428
1429
1430 /**
1431  * Copy any validated addresses to buf.
1432  *
1433  * @return 0 once all addresses have been
1434  *         returned
1435  */
1436 static size_t
1437 list_validated_addresses (void *cls, size_t max, void *buf)
1438 {
1439   struct ValidationAddress **va = cls;
1440   size_t ret;
1441
1442   while ((NULL != *va) && ((*va)->ok != GNUNET_YES))
1443     *va = (*va)->next;
1444   if (NULL == *va)
1445     return 0;
1446   ret = GNUNET_HELLO_add_address ((*va)->transport_name,
1447                                   (*va)->expiration,
1448                                   &(*va)[1], (*va)->addr_len, buf, max);
1449   *va = (*va)->next;
1450   return ret;
1451 }
1452
1453
1454 /**
1455  * HELLO validation cleanup task.
1456  */
1457 static void
1458 cleanup_validation (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1459 {
1460   struct ValidationAddress *va;
1461   struct ValidationList *pos;
1462   struct ValidationList *prev;
1463   struct GNUNET_TIME_Absolute now;
1464   struct GNUNET_TIME_Absolute first;
1465   struct GNUNET_HELLO_Message *hello;
1466   struct GNUNET_PeerIdentity pid;
1467   struct NeighbourList *n;
1468
1469 #if DEBUG_TRANSPORT
1470   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1471               "HELLO validation cleanup background task running...\n");
1472 #endif
1473   now = GNUNET_TIME_absolute_get ();
1474   prev = NULL;
1475   pos = pending_validations;
1476   while (pos != NULL)
1477     {
1478       if (pos->timeout.value < now.value)
1479         {
1480           if (prev == NULL)
1481             pending_validations = pos->next;
1482           else
1483             prev->next = pos->next;
1484           va = pos->addresses;
1485           hello = GNUNET_HELLO_create (&pos->publicKey,
1486                                        &list_validated_addresses, &va);
1487           GNUNET_CRYPTO_hash (&pos->publicKey,
1488                               sizeof (struct
1489                                       GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1490                               &pid.hashPubKey);
1491 #if DEBUG_TRANSPORT
1492           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1493                       "Creating persistent `%s' message for peer `%4s' based on confirmed addresses.\n",
1494                       "HELLO", GNUNET_i2s (&pid));
1495 #endif
1496           GNUNET_PEERINFO_add_peer (cfg, sched, &pid, hello);
1497           n = find_neighbour (&pid);
1498           if (NULL != n)
1499             try_transmission_to_peer (n);
1500           GNUNET_free (hello);
1501           while (NULL != (va = pos->addresses))
1502             {
1503               pos->addresses = va->next;
1504               GNUNET_free (va->transport_name);
1505               GNUNET_free (va);
1506             }
1507           GNUNET_free (pos);
1508           if (prev == NULL)
1509             pos = pending_validations;
1510           else
1511             pos = prev->next;
1512           continue;
1513         }
1514       prev = pos;
1515       pos = pos->next;
1516     }
1517
1518   /* finally, reschedule cleanup if needed; list is
1519      ordered by timeout, so we need the last element... */
1520   if (NULL != pending_validations)
1521     {
1522       first = pos->timeout;
1523       pos = pending_validations;
1524       while (pos != NULL)
1525         {
1526           first = GNUNET_TIME_absolute_min (first, pos->timeout);
1527           pos = pos->next;
1528         }
1529       GNUNET_SCHEDULER_add_delayed (sched,
1530                                     GNUNET_NO,
1531                                     GNUNET_SCHEDULER_PRIORITY_IDLE,
1532                                     GNUNET_SCHEDULER_NO_TASK,
1533                                     GNUNET_TIME_absolute_get_remaining (first),
1534                                     &cleanup_validation, NULL);
1535     }
1536 }
1537
1538
1539
1540
1541 /**
1542  * Function that will be called if we receive a validation
1543  * of an address challenge that we transmitted to another
1544  * peer.  Note that the validation should only be considered
1545  * acceptable if the challenge matches AND if the sender
1546  * address is at least a plausible address for this peer
1547  * (otherwise we may be seeing a MiM attack).
1548  *
1549  * @param cls closure
1550  * @param name name of the transport that generated the address
1551  * @param peer who responded to our challenge
1552  * @param challenge the challenge number we presumably used
1553  * @param sender_addr string describing our sender address (as observed
1554  *         by the other peer in human-readable format)
1555  */
1556 static void
1557 plugin_env_notify_validation (void *cls,
1558                               const char *name,
1559                               const struct GNUNET_PeerIdentity *peer,
1560                               uint32_t challenge,
1561                               const char *sender_addr)
1562 {
1563   unsigned int not_done;
1564   int matched;
1565   struct ValidationList *pos;
1566   struct ValidationAddress *va;
1567   struct GNUNET_PeerIdentity id;
1568
1569   pos = pending_validations;
1570   while (pos != NULL)
1571     {
1572       GNUNET_CRYPTO_hash (&pos->publicKey,
1573                           sizeof (struct
1574                                   GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1575                           &id.hashPubKey);
1576       if (0 ==
1577           memcmp (peer, &id, sizeof (struct GNUNET_PeerIdentity)))
1578         break;
1579       pos = pos->next;
1580     }
1581   if (pos == NULL)
1582     {
1583       /* TODO: call statistics (unmatched PONG) */
1584       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1585                   _
1586                   ("Received validation response but have no record of any validation request for `%4s'. Ignoring.\n"),
1587                   GNUNET_i2s(peer));
1588       return;
1589     }
1590   not_done = 0;
1591   matched = GNUNET_NO;
1592   va = pos->addresses;
1593   while (va != NULL)
1594     {
1595       if (va->challenge == challenge)
1596         {
1597 #if DEBUG_TRANSPORT
1598           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1599                       "Confirmed validity of address, peer `%4s' has address `%s'.\n",
1600                       GNUNET_i2s (peer),
1601                       GNUNET_a2s ((const struct sockaddr*) &va[1],
1602                                   va->addr_len));
1603 #endif
1604           GNUNET_log (GNUNET_ERROR_TYPE_INFO | GNUNET_ERROR_TYPE_BULK,
1605                       _("Another peer saw us using the address `%s' via `%s'. If this is not plausible, this address should be listed in the configuration as implausible to avoid MiM attacks.\n"),
1606                       sender_addr,
1607                       name);
1608           va->ok = GNUNET_YES;
1609           va->expiration =
1610             GNUNET_TIME_relative_to_absolute (HELLO_ADDRESS_EXPIRATION);
1611           matched = GNUNET_YES;
1612         }
1613       if (va->ok != GNUNET_YES)
1614         not_done++;
1615       va = va->next;
1616     }
1617   if (GNUNET_NO == matched)
1618     {
1619       /* TODO: call statistics (unmatched PONG) */
1620       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1621                   _
1622                   ("Received `%s' message but have no record of a matching `%s' message. Ignoring.\n"),
1623                   "PONG", "PING");
1624     }
1625   if (0 == not_done)
1626     {
1627 #if DEBUG_TRANSPORT
1628       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1629                   "All addresses validated, will now construct `%s' for `%4s'.\n",
1630                   "HELLO",
1631                   GNUNET_i2s (peer));
1632 #endif
1633       pos->timeout.value = 0;
1634       GNUNET_SCHEDULER_add_delayed (sched,
1635                                     GNUNET_NO,
1636                                     GNUNET_SCHEDULER_PRIORITY_IDLE,
1637                                     GNUNET_SCHEDULER_NO_TASK,
1638                                     GNUNET_TIME_UNIT_ZERO,
1639                                     &cleanup_validation, NULL);
1640     }
1641   else
1642     {
1643 #if DEBUG_TRANSPORT
1644       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1645                   "Still waiting for %u additional `%s' messages before constructing `%s' for `%4s'.\n",
1646                   not_done,
1647                   "PONG",
1648                   "HELLO",
1649                   GNUNET_i2s (peer));
1650 #endif
1651     }
1652 }
1653
1654
1655 struct CheckHelloValidatedContext
1656 {
1657   /**
1658    * Plugin for which we are validating.
1659    */
1660   struct TransportPlugin *plugin;
1661
1662   /**
1663    * Hello that we are validating.
1664    */
1665   struct GNUNET_HELLO_Message *hello;
1666
1667   /**
1668    * Validation list being build.
1669    */
1670   struct ValidationList *e;
1671
1672 };
1673
1674
1675 /**
1676  * Append the given address to the list of entries
1677  * that need to be validated.
1678  */
1679 static int
1680 run_validation (void *cls,
1681                 const char *tname,
1682                 struct GNUNET_TIME_Absolute expiration,
1683                 const void *addr, size_t addrlen)
1684 {
1685   struct ValidationList *e = cls;
1686   struct TransportPlugin *tp;
1687   struct ValidationAddress *va;
1688   struct GNUNET_PeerIdentity id;
1689
1690   tp = find_transport (tname);
1691   if (tp == NULL)
1692     {
1693       GNUNET_log (GNUNET_ERROR_TYPE_INFO |
1694                   GNUNET_ERROR_TYPE_BULK,
1695                   _
1696                   ("Transport `%s' not loaded, will not try to validate peer address using this transport.\n"),
1697                   tname);
1698       return GNUNET_OK;
1699     }
1700   GNUNET_CRYPTO_hash (&e->publicKey,
1701                       sizeof (struct
1702                               GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1703                       &id.hashPubKey);
1704   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1705               "Scheduling validation of address `%s' via `%s' for `%4s'\n",
1706               GNUNET_a2s(addr, addrlen),
1707               tname,
1708               GNUNET_i2s(&id));
1709
1710   va = GNUNET_malloc (sizeof (struct ValidationAddress) + addrlen);
1711   va->next = e->addresses;
1712   e->addresses = va;
1713   va->transport_name = GNUNET_strdup (tname);
1714   va->addr_len = addrlen;
1715   va->challenge = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1716                                             (unsigned int) -1);
1717   memcpy (&va[1], addr, addrlen);
1718   return GNUNET_OK;
1719 }
1720
1721
1722 /**
1723  * Check if addresses in validated hello "h" overlap with
1724  * those in "chvc->hello" and update "chvc->hello" accordingly,
1725  * removing those addresses that have already been validated.
1726  */
1727 static void
1728 check_hello_validated (void *cls,
1729                        const struct GNUNET_PeerIdentity *peer,
1730                        const struct GNUNET_HELLO_Message *h,
1731                        uint32_t trust)
1732 {
1733   struct CheckHelloValidatedContext *chvc = cls;
1734   struct ValidationAddress *va;
1735   struct TransportPlugin *tp;
1736   int first_call;
1737   struct GNUNET_PeerIdentity apeer;
1738
1739   first_call = GNUNET_NO;
1740   if (chvc->e == NULL)
1741     {
1742       first_call = GNUNET_YES;
1743       chvc->e = GNUNET_malloc (sizeof (struct ValidationList));
1744       GNUNET_assert (GNUNET_OK ==
1745                      GNUNET_HELLO_get_key (h != NULL ? h : chvc->hello,
1746                                            &chvc->e->publicKey));
1747       chvc->e->timeout =
1748         GNUNET_TIME_relative_to_absolute (HELLO_VERIFICATION_TIMEOUT);
1749       chvc->e->next = pending_validations;
1750       pending_validations = chvc->e;
1751     }
1752   if (h != NULL)
1753     {
1754       GNUNET_HELLO_iterate_new_addresses (chvc->hello,
1755                                           h,
1756                                           GNUNET_TIME_absolute_get (),
1757                                           &run_validation, chvc->e);
1758     }
1759   else if (GNUNET_YES == first_call)
1760     {
1761       /* no existing HELLO, all addresses are new */
1762       GNUNET_HELLO_iterate_addresses (chvc->hello,
1763                                       GNUNET_NO, &run_validation, chvc->e);
1764     }
1765   if (h != NULL)
1766     return;                     /* wait for next call */
1767   /* finally, transmit validation attempts */
1768   GNUNET_assert (GNUNET_OK ==
1769                  GNUNET_HELLO_get_id (chvc->hello,
1770                                       &apeer));
1771 #if DEBUG_TRANSPORT
1772   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1773               "Ready to validate addresses from `%s' message for peer `%4s'\n",
1774               "HELLO", GNUNET_i2s (&apeer));
1775 #endif
1776   va = chvc->e->addresses;
1777   while (va != NULL)
1778     {
1779 #if DEBUG_TRANSPORT
1780       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1781                   "Establishing `%s' connection to validate `%s' address `%s' of `%4s'\n",
1782                   va->transport_name,
1783                   "HELLO",
1784                   GNUNET_a2s ((const struct sockaddr*) &va[1],
1785                               va->addr_len),
1786                   GNUNET_i2s (&apeer));
1787 #endif
1788       tp = find_transport (va->transport_name);
1789       GNUNET_assert (tp != NULL);
1790       if (GNUNET_OK !=
1791           tp->api->validate (tp->api->cls,
1792                              &apeer,
1793                              va->challenge,
1794                              HELLO_VERIFICATION_TIMEOUT,
1795                              &va[1],
1796                              va->addr_len))
1797         va->ok = GNUNET_SYSERR;
1798       va = va->next;
1799     }
1800   GNUNET_SCHEDULER_add_delayed (sched,
1801                                 GNUNET_NO,
1802                                 GNUNET_SCHEDULER_PRIORITY_IDLE,
1803                                 GNUNET_SCHEDULER_NO_TASK,
1804                                 GNUNET_TIME_absolute_get_remaining (chvc->e->timeout),
1805                                 &cleanup_validation, NULL);
1806   GNUNET_free (chvc);
1807 }
1808
1809
1810 /**
1811  * Process HELLO-message.
1812  *
1813  * @param plugin transport involved, may be NULL
1814  * @param message the actual message
1815  * @return GNUNET_OK if the HELLO was well-formed, GNUNET_SYSERR otherwise
1816  */
1817 static int
1818 process_hello (struct TransportPlugin *plugin,
1819                const struct GNUNET_MessageHeader *message)
1820 {
1821   struct ValidationList *e;
1822   uint16_t hsize;
1823   struct GNUNET_PeerIdentity target;
1824   const struct GNUNET_HELLO_Message *hello;
1825   struct CheckHelloValidatedContext *chvc;
1826   struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded publicKey;
1827
1828   hsize = ntohs (message->size);
1829   if ((ntohs (message->type) != GNUNET_MESSAGE_TYPE_HELLO) ||
1830       (hsize < sizeof (struct GNUNET_MessageHeader)))
1831     {
1832       GNUNET_break (0);
1833       return GNUNET_SYSERR;
1834     }
1835   /* first, check if load is too high */
1836   if (GNUNET_OS_load_cpu_get (cfg) > 100)
1837     {
1838       /* TODO: call to stats? */
1839       return GNUNET_OK;
1840     }
1841   hello = (const struct GNUNET_HELLO_Message *) message;
1842   if (GNUNET_OK != GNUNET_HELLO_get_key (hello, &publicKey))
1843     {
1844       GNUNET_break_op (0);
1845       return GNUNET_SYSERR;
1846     }
1847   GNUNET_CRYPTO_hash (&publicKey,
1848                       sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1849                       &target.hashPubKey);
1850 #if DEBUG_TRANSPORT
1851   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1852               "Processing `%s' message for `%4s'\n",
1853               "HELLO", GNUNET_i2s (&target));
1854 #endif
1855   /* check if a HELLO for this peer is already on the validation list */
1856   e = pending_validations;
1857   while (e != NULL)
1858     {
1859       if (0 == memcmp (&e->publicKey,
1860                        &publicKey,
1861                        sizeof (struct
1862                                GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded)))
1863         {
1864           /* TODO: call to stats? */
1865 #if DEBUG_TRANSPORT
1866           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1867                       "`%s' message for peer `%4s' is already pending; ignoring new message\n",
1868                       "HELLO", GNUNET_i2s (&target));
1869 #endif
1870           return GNUNET_OK;
1871         }
1872       e = e->next;
1873     }
1874   chvc = GNUNET_malloc (sizeof (struct CheckHelloValidatedContext) + hsize);
1875   chvc->plugin = plugin;
1876   chvc->hello = (struct GNUNET_HELLO_Message *) &chvc[1];
1877   memcpy (chvc->hello, hello, hsize);
1878   /* finally, check if HELLO was previously validated
1879      (continuation will then schedule actual validation) */
1880   GNUNET_PEERINFO_for_all (cfg,
1881                            sched,
1882                            &target,
1883                            0,
1884                            HELLO_VERIFICATION_TIMEOUT,
1885                            &check_hello_validated, chvc);
1886   return GNUNET_OK;
1887 }
1888
1889
1890 /**
1891  * The peer specified by the given neighbour has timed-out or a plugin
1892  * has disconnected.  We may either need to do nothing (other plugins
1893  * still up), or trigger a full disconnect and clean up.  This
1894  * function updates our state and do the necessary notifications.
1895  * Also notifies our clients that the neighbour is now officially
1896  * gone.
1897  *
1898  * @param n the neighbour list entry for the peer
1899  * @param check should we just check if all plugins
1900  *        disconnected or must we ask all plugins to
1901  *        disconnect?
1902  */
1903 static void
1904 disconnect_neighbour (struct NeighbourList *n,
1905                       int check)
1906 {
1907   struct ReadyList *rpos;
1908   struct NeighbourList *npos;
1909   struct NeighbourList *nprev;
1910   struct MessageQueue *mq;
1911
1912   if (GNUNET_YES == check)
1913     {
1914       rpos = n->plugins;
1915       while (NULL != rpos)
1916         {
1917           if (GNUNET_YES == rpos->connected)
1918             return; /* still connected */
1919           rpos = rpos->next;
1920         }
1921     }
1922
1923 #if DEBUG_TRANSPORT
1924   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
1925               "Disconnecting from `%4s'\n",
1926               GNUNET_i2s(&n->id));
1927 #endif
1928   /* remove n from neighbours list */
1929   nprev = NULL;
1930   npos = neighbours;
1931   while ((npos != NULL) && (npos != n))
1932     {
1933       nprev = npos;
1934       npos = npos->next;
1935     }
1936   GNUNET_assert (npos != NULL);
1937   if (nprev == NULL)
1938     neighbours = n->next;
1939   else
1940     nprev->next = n->next;
1941
1942   /* notify all clients about disconnect */
1943   notify_clients_disconnect (&n->id);
1944
1945   /* clean up all plugins, cancel connections and pending transmissions */
1946   while (NULL != (rpos = n->plugins))
1947     {
1948       n->plugins = rpos->next;
1949       GNUNET_assert (rpos->neighbour == n);
1950       if (GNUNET_YES == rpos->connected)
1951         rpos->plugin->api->cancel (rpos->plugin->api->cls,
1952                                    rpos->plugin_handle,
1953                                    rpos,
1954                                    &n->id);
1955       GNUNET_free (rpos);
1956     }
1957
1958   /* free all messages on the queue */
1959   while (NULL != (mq = n->messages))
1960     {
1961       n->messages = mq->next;
1962       GNUNET_assert (mq->neighbour == n);
1963       GNUNET_free (mq);
1964     }
1965   if (n->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1966     GNUNET_SCHEDULER_cancel (sched,
1967                              n->timeout_task);
1968   /* finally, free n itself */
1969   GNUNET_free (n);
1970 }
1971
1972
1973 /**
1974  * Add an entry for each of our transport plugins
1975  * (that are able to send) to the list of plugins
1976  * for this neighbour.
1977  *
1978  * @param neighbour to initialize
1979  */
1980 static void
1981 add_plugins (struct NeighbourList *neighbour)
1982 {
1983   struct TransportPlugin *tp;
1984   struct ReadyList *rl;
1985
1986   neighbour->retry_plugins_time
1987     = GNUNET_TIME_relative_to_absolute (PLUGIN_RETRY_FREQUENCY);
1988   tp = plugins;
1989   while (tp != NULL)
1990     {
1991       if (tp->api->send != NULL)
1992         {
1993           rl = GNUNET_malloc (sizeof (struct ReadyList));
1994           rl->next = neighbour->plugins;
1995           neighbour->plugins = rl;
1996           rl->plugin = tp;
1997           rl->neighbour = neighbour;
1998           rl->transmit_ready = GNUNET_YES;
1999         }
2000       tp = tp->next;
2001     }
2002 }
2003
2004
2005 static void
2006 neighbour_timeout_task (void *cls,
2007                         const struct GNUNET_SCHEDULER_TaskContext *tc)
2008 {
2009   struct NeighbourList *n = cls;
2010
2011 #if DEBUG_TRANSPORT
2012   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2013               "Neighbour `%4s' has timed out!\n",
2014               GNUNET_i2s(&n->id));
2015 #endif
2016   n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
2017   disconnect_neighbour (n, GNUNET_NO);
2018 }
2019
2020
2021 /**
2022  * Create a fresh entry in our neighbour list for the given peer.
2023  * Will try to transmit our current HELLO to the new neighbour.  Also
2024  * notifies our clients about the new "connection".
2025  *
2026  * @param peer the peer for which we create the entry
2027  * @return the new neighbour list entry
2028  */
2029 static struct NeighbourList *
2030 setup_new_neighbour (const struct GNUNET_PeerIdentity *peer)
2031 {
2032   struct NeighbourList *n;
2033
2034 #if DEBUG_TRANSPORT
2035   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2036               "Setting up new neighbour `%4s', sending our HELLO to introduce ourselves\n",
2037               GNUNET_i2s (peer));
2038 #endif
2039   GNUNET_assert (our_hello != NULL);
2040   n = GNUNET_malloc (sizeof (struct NeighbourList));
2041   n->next = neighbours;
2042   neighbours = n;
2043   n->id = *peer;
2044   n->last_quota_update = GNUNET_TIME_absolute_get ();
2045   n->peer_timeout =
2046     GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2047   n->quota_in = (GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT + 59999) / (60 * 1000);
2048   add_plugins (n);
2049   n->timeout_task = GNUNET_SCHEDULER_add_delayed (sched,
2050                                                   GNUNET_NO,
2051                                                   GNUNET_SCHEDULER_PRIORITY_IDLE,
2052                                                   GNUNET_SCHEDULER_NO_TASK,
2053                                                   GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
2054                                                   &neighbour_timeout_task, n);
2055   transmit_to_peer (NULL, 0,
2056                     (const struct GNUNET_MessageHeader *) our_hello,
2057                     GNUNET_YES, n);
2058   notify_clients_connect (peer, GNUNET_TIME_UNIT_FOREVER_REL);
2059   return n;
2060 }
2061
2062
2063 /**
2064  * Function called by the plugin for each received message.
2065  * Update data volumes, possibly notify plugins about
2066  * reducing the rate at which they read from the socket
2067  * and generally forward to our receive callback.
2068  *
2069  * @param plugin_context value to pass to this plugin
2070  *        to respond to the given peer (use is optional,
2071  *        but may speed up processing)
2072  * @param service_context value passed to the transport-service
2073  *        to identify the neighbour; will be NULL on the first
2074  *        call for a given peer
2075  * @param latency estimated latency for communicating with the
2076  *             given peer
2077  * @param peer (claimed) identity of the other peer
2078  * @param message the message, NULL if peer was disconnected
2079  * @return the new service_context that the plugin should use
2080  *         for future receive calls for messages from this
2081  *         particular peer
2082  */
2083 static struct ReadyList *
2084 plugin_env_receive (void *cls,
2085                     void *plugin_context,
2086                     struct ReadyList *service_context,
2087                     struct GNUNET_TIME_Relative latency,
2088                     const struct GNUNET_PeerIdentity *peer,
2089                     const struct GNUNET_MessageHeader *message)
2090 {
2091   const struct GNUNET_MessageHeader ack = {
2092     htons (sizeof (struct GNUNET_MessageHeader)),
2093     htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ACK)
2094   };
2095   struct TransportPlugin *plugin = cls;
2096   struct TransportClient *cpos;
2097   struct InboundMessage *im;
2098   uint16_t msize;
2099   struct NeighbourList *n;
2100
2101   if (service_context != NULL)
2102     {
2103       n = service_context->neighbour;
2104       GNUNET_assert (n != NULL);
2105     }
2106   else
2107     {
2108       n = find_neighbour (peer);
2109       if (n == NULL)
2110         {
2111           if (message == NULL)
2112             return NULL;        /* disconnect of peer already marked down */
2113           n = setup_new_neighbour (peer);
2114         }
2115       service_context = n->plugins;
2116       while ((service_context != NULL) && (plugin != service_context->plugin))
2117         service_context = service_context->next;
2118       GNUNET_assert ((plugin->api->send == NULL) ||
2119                      (service_context != NULL));
2120     }
2121   if (message == NULL)
2122     {
2123 #if DEBUG_TRANSPORT
2124       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2125                   "Receive failed from `%4s', triggering disconnect\n",
2126                   GNUNET_i2s(&n->id));
2127 #endif
2128       /* TODO: call stats */
2129       if ((service_context != NULL) &&
2130           (service_context->plugin_handle == plugin_context))
2131         {
2132           service_context->connected = GNUNET_NO;
2133           service_context->plugin_handle = NULL;
2134         }
2135       disconnect_neighbour (n, GNUNET_YES);
2136       return NULL;
2137     }
2138 #if DEBUG_TRANSPORT
2139   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2140               "Processing message of type `%u' received by plugin...\n",
2141               ntohs (message->type));
2142 #endif
2143   if (service_context != NULL)
2144     {
2145       if (service_context->connected == GNUNET_NO)
2146         {
2147           service_context->connected = GNUNET_YES;
2148           service_context->transmit_ready = GNUNET_YES;
2149           service_context->connect_attempts++;
2150         }
2151       service_context->timeout
2152         = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2153       service_context->plugin_handle = plugin_context;
2154       service_context->latency = latency;
2155     }
2156   /* update traffic received amount ... */
2157   msize = ntohs (message->size);
2158   n->last_received += msize;
2159   GNUNET_SCHEDULER_cancel (sched, n->timeout_task);
2160   n->peer_timeout =
2161     GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
2162   n->timeout_task =
2163     GNUNET_SCHEDULER_add_delayed (sched, GNUNET_NO,
2164                                   GNUNET_SCHEDULER_PRIORITY_IDLE,
2165                                   GNUNET_SCHEDULER_NO_TASK,
2166                                   GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
2167                                   &neighbour_timeout_task, n);
2168   update_quota (n);
2169   if (n->quota_violation_count > QUOTA_VIOLATION_DROP_THRESHOLD)
2170     {
2171       /* dropping message due to frequent inbound volume violations! */
2172       GNUNET_log (GNUNET_ERROR_TYPE_WARNING |
2173                   GNUNET_ERROR_TYPE_BULK,
2174                   _
2175                   ("Dropping incoming message due to repeated bandwidth quota violations.\n"));
2176       /* TODO: call stats */
2177       GNUNET_assert ( (service_context == NULL) ||
2178                       (NULL != service_context->neighbour) );
2179       return service_context;
2180     }
2181   switch (ntohs (message->type))
2182     {
2183     case GNUNET_MESSAGE_TYPE_HELLO:
2184 #if DEBUG_TRANSPORT
2185       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2186                   "Receiving `%s' message from `%4s'.\n", "HELLO",
2187                   GNUNET_i2s(peer));
2188 #endif
2189       process_hello (plugin, message);
2190 #if DEBUG_TRANSPORT
2191       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2192                   "Sending `%s' message to connecting peer `%4s'.\n", "ACK",
2193                   GNUNET_i2s(peer));
2194 #endif
2195       transmit_to_peer (NULL, 0, &ack, GNUNET_YES, n);
2196       break;
2197     case GNUNET_MESSAGE_TYPE_TRANSPORT_ACK:
2198       n->saw_ack = GNUNET_YES;
2199       /* intentional fall-through! */
2200     default:
2201 #if DEBUG_TRANSPORT
2202       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2203                   "Received message of type %u from `%4s', sending to all clients.\n",
2204                   ntohs (message->type),
2205                   GNUNET_i2s(peer));
2206 #endif
2207       /* transmit message to all clients */
2208       im = GNUNET_malloc (sizeof (struct InboundMessage) + msize);
2209       im->header.size = htons (sizeof (struct InboundMessage) + msize);
2210       im->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
2211       im->latency = GNUNET_TIME_relative_hton (latency);
2212       im->peer = *peer;
2213       memcpy (&im[1], message, msize);
2214
2215       cpos = clients;
2216       while (cpos != NULL)
2217         {
2218           transmit_to_client (cpos, &im->header, GNUNET_YES);
2219           cpos = cpos->next;
2220         }
2221       GNUNET_free (im);
2222     }
2223   GNUNET_assert ( (service_context == NULL) ||
2224                   (NULL != service_context->neighbour) );
2225   return service_context;
2226 }
2227
2228
2229 /**
2230  * Handle START-message.  This is the first message sent to us
2231  * by any client which causes us to add it to our list.
2232  *
2233  * @param cls closure (always NULL)
2234  * @param client identification of the client
2235  * @param message the actual message
2236  */
2237 static void
2238 handle_start (void *cls,
2239               struct GNUNET_SERVER_Client *client,
2240               const struct GNUNET_MessageHeader *message)
2241 {
2242   struct TransportClient *c;
2243   struct ConnectInfoMessage cim;
2244   struct NeighbourList *n;
2245   struct InboundMessage *im;
2246   struct GNUNET_MessageHeader *ack;
2247
2248 #if DEBUG_TRANSPORT
2249   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2250               "Received `%s' request from client\n", "START");
2251 #endif
2252   c = clients;
2253   while (c != NULL)
2254     {
2255       if (c->client == client)
2256         {
2257           /* client already on our list! */
2258           GNUNET_break (0);
2259           GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
2260           return;
2261         }
2262       c = c->next;
2263     }
2264   c = GNUNET_malloc (sizeof (struct TransportClient));
2265   c->next = clients;
2266   clients = c;
2267   c->client = client;
2268   if (our_hello != NULL)
2269     {
2270 #if DEBUG_TRANSPORT
2271       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2272                   "Sending our own `%s' to new client\n",
2273                   "HELLO");
2274 #endif
2275       transmit_to_client (c,
2276                           (const struct GNUNET_MessageHeader *) our_hello,
2277                           GNUNET_NO);
2278       /* tell new client about all existing connections */
2279       cim.header.size = htons (sizeof (struct ConnectInfoMessage));
2280       cim.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
2281       cim.quota_out = htonl (GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT / (60 * 1000));
2282       cim.latency = GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_ZERO);  /* FIXME? */
2283       im = GNUNET_malloc (sizeof (struct InboundMessage) +
2284                           sizeof (struct GNUNET_MessageHeader));
2285       im->header.size = htons (sizeof (struct InboundMessage) +
2286                                sizeof (struct GNUNET_MessageHeader));
2287       im->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
2288       im->latency = GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_ZERO);  /* FIXME? */
2289       ack = (struct GNUNET_MessageHeader *) &im[1];
2290       ack->size = htons (sizeof (struct GNUNET_MessageHeader));
2291       ack->type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ACK);
2292       for (n = neighbours; n != NULL; n = n->next)
2293         {
2294           cim.id = n->id;
2295           transmit_to_client (c, &cim.header, GNUNET_NO);
2296           if (n->saw_ack)
2297             {
2298               im->peer = n->id;
2299               transmit_to_client (c, &im->header, GNUNET_NO);
2300             }
2301         }
2302       GNUNET_free (im);
2303     }
2304   GNUNET_SERVER_receive_done (client, GNUNET_OK);
2305 }
2306
2307
2308 /**
2309  * Handle HELLO-message.
2310  *
2311  * @param cls closure (always NULL)
2312  * @param client identification of the client
2313  * @param message the actual message
2314  */
2315 static void
2316 handle_hello (void *cls,
2317               struct GNUNET_SERVER_Client *client,
2318               const struct GNUNET_MessageHeader *message)
2319 {
2320   int ret;
2321
2322 #if DEBUG_TRANSPORT
2323   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2324               "Received `%s' request from client\n", "HELLO");
2325 #endif
2326   ret = process_hello (NULL, message);
2327   GNUNET_SERVER_receive_done (client, ret);
2328 }
2329
2330
2331 /**
2332  * Handle SEND-message.
2333  *
2334  * @param cls closure (always NULL)
2335  * @param client identification of the client
2336  * @param message the actual message
2337  */
2338 static void
2339 handle_send (void *cls,
2340              struct GNUNET_SERVER_Client *client,
2341              const struct GNUNET_MessageHeader *message)
2342 {
2343   struct TransportClient *tc;
2344   struct NeighbourList *n;
2345   const struct OutboundMessage *obm;
2346   const struct GNUNET_MessageHeader *obmm;
2347   uint16_t size;
2348   uint16_t msize;
2349
2350   size = ntohs (message->size);
2351   if (size <
2352       sizeof (struct OutboundMessage) + sizeof (struct GNUNET_MessageHeader))
2353     {
2354       GNUNET_break (0);
2355       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
2356       return;
2357     }
2358   obm = (const struct OutboundMessage *) message;
2359 #if DEBUG_TRANSPORT
2360   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2361               "Received `%s' request from client with target `%4s'\n",
2362               "SEND", GNUNET_i2s (&obm->peer));
2363 #endif
2364   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
2365   msize = ntohs (obmm->size);
2366   if (size != msize + sizeof (struct OutboundMessage))
2367     {
2368       GNUNET_break (0);
2369       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
2370       return;
2371     }
2372   n = find_neighbour (&obm->peer);
2373   if (n == NULL)
2374     n = setup_new_neighbour (&obm->peer);
2375   tc = clients;
2376   while ((tc != NULL) && (tc->client != client))
2377     tc = tc->next;
2378
2379 #if DEBUG_TRANSPORT
2380   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2381               "Client asked to transmit %u-byte message of type %u to `%4s'\n",
2382               ntohs (obmm->size),
2383               ntohs (obmm->type), GNUNET_i2s (&obm->peer));
2384 #endif
2385   transmit_to_peer (tc, ntohl(obm->priority), obmm, GNUNET_NO, n);
2386   GNUNET_SERVER_receive_done (client, GNUNET_OK);
2387 }
2388
2389
2390 /**
2391  * Handle SET_QUOTA-message.
2392  *
2393  * @param cls closure (always NULL)
2394  * @param client identification of the client
2395  * @param message the actual message
2396  */
2397 static void
2398 handle_set_quota (void *cls,
2399                   struct GNUNET_SERVER_Client *client,
2400                   const struct GNUNET_MessageHeader *message)
2401 {
2402   const struct QuotaSetMessage *qsm =
2403     (const struct QuotaSetMessage *) message;
2404   struct NeighbourList *n;
2405   struct TransportPlugin *p;
2406   struct ReadyList *rl;
2407
2408 #if DEBUG_TRANSPORT
2409   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2410               "Received `%s' request from client for peer `%4s'\n",
2411               "SET_QUOTA", GNUNET_i2s (&qsm->peer));
2412 #endif
2413   n = find_neighbour (&qsm->peer);
2414   if (n == NULL)
2415     {
2416       GNUNET_SERVER_receive_done (client, GNUNET_OK);
2417       return;
2418     }
2419   update_quota (n);
2420   if (n->quota_in < ntohl (qsm->quota_in))
2421     n->last_quota_update = GNUNET_TIME_absolute_get ();
2422   n->quota_in = ntohl (qsm->quota_in);
2423   rl = n->plugins;
2424   while (rl != NULL)
2425     {
2426       p = rl->plugin;
2427       p->api->set_receive_quota (p->api->cls,
2428                                  &qsm->peer, ntohl (qsm->quota_in));
2429       rl = rl->next;
2430     }
2431   GNUNET_SERVER_receive_done (client, GNUNET_OK);
2432 }
2433
2434
2435 /**
2436  * Handle TRY_CONNECT-message.
2437  *
2438  * @param cls closure (always NULL)
2439  * @param client identification of the client
2440  * @param message the actual message
2441  */
2442 static void
2443 handle_try_connect (void *cls,
2444                     struct GNUNET_SERVER_Client *client,
2445                     const struct GNUNET_MessageHeader *message)
2446 {
2447   const struct TryConnectMessage *tcm;
2448
2449   tcm = (const struct TryConnectMessage *) message;
2450 #if DEBUG_TRANSPORT
2451   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2452               "Received `%s' request from client %p asking to connect to `%4s'\n",
2453               "TRY_CONNECT",
2454               client,
2455               GNUNET_i2s (&tcm->peer));
2456 #endif
2457   if (NULL == find_neighbour (&tcm->peer))
2458     setup_new_neighbour (&tcm->peer);
2459 #if DEBUG_TRANSPORT
2460   else
2461     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2462                 "Client asked to connect to `%4s', but connection already exists\n",
2463                 "TRY_CONNECT",
2464                 GNUNET_i2s (&tcm->peer));
2465 #endif
2466   GNUNET_SERVER_receive_done (client, GNUNET_OK);
2467 }
2468
2469
2470 /**
2471  * List of handlers for the messages understood by this
2472  * service.
2473  */
2474 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2475   {&handle_start, NULL,
2476    GNUNET_MESSAGE_TYPE_TRANSPORT_START, 0},
2477   {&handle_hello, NULL,
2478    GNUNET_MESSAGE_TYPE_HELLO, 0},
2479   {&handle_send, NULL,
2480    GNUNET_MESSAGE_TYPE_TRANSPORT_SEND, 0},
2481   {&handle_set_quota, NULL,
2482    GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA, sizeof (struct QuotaSetMessage)},
2483   {&handle_try_connect, NULL,
2484    GNUNET_MESSAGE_TYPE_TRANSPORT_TRY_CONNECT,
2485    sizeof (struct TryConnectMessage)},
2486   {NULL, NULL, 0, 0}
2487 };
2488
2489
2490 /**
2491  * Setup the environment for this plugin.
2492  */
2493 static void
2494 create_environment (struct TransportPlugin *plug)
2495 {
2496   plug->env.cfg = cfg;
2497   plug->env.sched = sched;
2498   plug->env.my_public_key = &my_public_key;
2499   plug->env.my_private_key = my_private_key;
2500   plug->env.my_identity = &my_identity;
2501   plug->env.cls = plug;
2502   plug->env.receive = &plugin_env_receive;
2503   plug->env.lookup = &plugin_env_lookup_address;
2504   plug->env.notify_address = &plugin_env_notify_address;
2505   plug->env.notify_validation = &plugin_env_notify_validation;
2506   plug->env.default_quota_in = (GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT + 59999) / (60 * 1000);
2507   plug->env.max_connections = max_connect_per_transport;
2508 }
2509
2510
2511 /**
2512  * Start the specified transport (load the plugin).
2513  */
2514 static void
2515 start_transport (struct GNUNET_SERVER_Handle *server, const char *name)
2516 {
2517   struct TransportPlugin *plug;
2518   char *libname;
2519
2520   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2521               _("Loading `%s' transport plugin\n"), name);
2522   GNUNET_asprintf (&libname, "libgnunet_plugin_transport_%s", name);
2523   plug = GNUNET_malloc (sizeof (struct TransportPlugin));
2524   create_environment (plug);
2525   plug->short_name = GNUNET_strdup (name);
2526   plug->lib_name = libname;
2527   plug->next = plugins;
2528   plugins = plug;
2529   plug->api = GNUNET_PLUGIN_load (libname, &plug->env);
2530   if (plug->api == NULL)
2531     {
2532       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2533                   _("Failed to load transport plugin for `%s'\n"), name);
2534       GNUNET_free (plug->short_name);
2535       plugins = plug->next;
2536       GNUNET_free (libname);
2537       GNUNET_free (plug);
2538     }
2539 }
2540
2541
2542 /**
2543  * Called whenever a client is disconnected.  Frees our
2544  * resources associated with that client.
2545  *
2546  * @param cls closure
2547  * @param client identification of the client
2548  */
2549 static void
2550 client_disconnect_notification (void *cls,
2551                                 struct GNUNET_SERVER_Client *client)
2552 {
2553   struct TransportClient *pos;
2554   struct TransportClient *prev;
2555   struct ClientMessageQueueEntry *mqe;
2556
2557 #if DEBUG_TRANSPORT
2558   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2559               "Client disconnected, cleaning up.\n");
2560 #endif
2561   prev = NULL;
2562   pos = clients;
2563   while ((pos != NULL) && (pos->client != client))
2564     {
2565       prev = pos;
2566       pos = pos->next;
2567     }
2568   if (pos == NULL)
2569     return;
2570   while (NULL != (mqe = pos->message_queue_head))
2571     {
2572       pos->message_queue_head = mqe->next;
2573       GNUNET_free (mqe);
2574     }
2575   pos->message_queue_head = NULL;
2576   if (prev == NULL)
2577     clients = pos->next;
2578   else
2579     prev->next = pos->next;
2580   if (GNUNET_YES == pos->tcs_pending)
2581     {
2582       pos->client = NULL;
2583       return;
2584     }
2585   GNUNET_free (pos);
2586 }
2587
2588
2589 /**
2590  * Initiate transport service.
2591  *
2592  * @param cls closure
2593  * @param s scheduler to use
2594  * @param serv the initialized server
2595  * @param c configuration to use
2596  */
2597 static void
2598 run (void *cls,
2599      struct GNUNET_SCHEDULER_Handle *s,
2600      struct GNUNET_SERVER_Handle *serv,
2601      const struct GNUNET_CONFIGURATION_Handle *c)
2602 {
2603   char *plugs;
2604   char *pos;
2605   int no_transports;
2606   unsigned long long tneigh;
2607   char *keyfile;
2608
2609   sched = s;
2610   cfg = c;
2611   /* parse configuration */
2612   if ((GNUNET_OK !=
2613        GNUNET_CONFIGURATION_get_value_number (c,
2614                                               "TRANSPORT",
2615                                               "NEIGHBOUR_LIMIT",
2616                                               &tneigh)) ||
2617       (GNUNET_OK !=
2618        GNUNET_CONFIGURATION_get_value_filename (c,
2619                                                 "GNUNETD",
2620                                                 "HOSTKEY", &keyfile)))
2621     {
2622       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2623                   _
2624                   ("Transport service is lacking key configuration settings.  Exiting.\n"));
2625       GNUNET_SCHEDULER_shutdown (s);
2626       return;
2627     }
2628   max_connect_per_transport = (uint32_t) tneigh;
2629   my_private_key = GNUNET_CRYPTO_rsa_key_create_from_file (keyfile);
2630   GNUNET_free (keyfile);
2631   if (my_private_key == NULL)
2632     {
2633       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2634                   _
2635                   ("Transport service could not access hostkey.  Exiting.\n"));
2636       GNUNET_SCHEDULER_shutdown (s);
2637       return;
2638     }
2639   GNUNET_CRYPTO_rsa_key_get_public (my_private_key, &my_public_key);
2640   GNUNET_CRYPTO_hash (&my_public_key,
2641                       sizeof (my_public_key), &my_identity.hashPubKey);
2642   /* setup notification */
2643   server = serv;
2644   GNUNET_SERVER_disconnect_notify (server,
2645                                    &client_disconnect_notification, NULL);
2646   /* load plugins... */
2647   no_transports = 1;
2648   if (GNUNET_OK ==
2649       GNUNET_CONFIGURATION_get_value_string (c,
2650                                              "TRANSPORT", "PLUGINS", &plugs))
2651     {
2652       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2653                   _("Starting transport plugins `%s'\n"), plugs);
2654       pos = strtok (plugs, " ");
2655       while (pos != NULL)
2656         {
2657           start_transport (server, pos);
2658           no_transports = 0;
2659           pos = strtok (NULL, " ");
2660         }
2661       GNUNET_free (plugs);
2662     }
2663   if (no_transports)
2664     refresh_hello ();
2665 #if DEBUG_TRANSPORT
2666   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2667               _("Transport service ready.\n"));
2668 #endif
2669   /* process client requests */
2670   GNUNET_SERVER_add_handlers (server, handlers);
2671 }
2672
2673
2674 /**
2675  * Function called when the service shuts
2676  * down.  Unloads our plugins.
2677  *
2678  * @param cls closure
2679  * @param cfg configuration to use
2680  */
2681 static void
2682 unload_plugins (void *cls,
2683                 const struct GNUNET_CONFIGURATION_Handle *cfg)
2684 {
2685   struct TransportPlugin *plug;
2686   struct AddressList *al;
2687
2688 #if DEBUG_TRANSPORT
2689   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2690               "Transport service is unloading plugins...\n");
2691 #endif
2692   while (NULL != (plug = plugins))
2693     {
2694       plugins = plug->next;
2695       GNUNET_break (NULL == GNUNET_PLUGIN_unload (plug->lib_name, plug->api));
2696       GNUNET_free (plug->lib_name);
2697       GNUNET_free (plug->short_name);
2698       while (NULL != (al = plug->addresses))
2699         {
2700           plug->addresses = al->next;
2701           GNUNET_free (al);
2702         }
2703       GNUNET_free (plug);
2704     }
2705   if (my_private_key != NULL)
2706     GNUNET_CRYPTO_rsa_key_free (my_private_key);
2707   GNUNET_free_non_null (our_hello);
2708 }
2709
2710
2711 /**
2712  * The main function for the transport service.
2713  *
2714  * @param argc number of arguments from the command line
2715  * @param argv command line arguments
2716  * @return 0 ok, 1 on error
2717  */
2718 int
2719 main (int argc, char *const *argv)
2720 {
2721   return (GNUNET_OK ==
2722           GNUNET_SERVICE_run (argc,
2723                               argv,
2724                               "transport",
2725                               &run, NULL, &unload_plugins, NULL)) ? 0 : 1;
2726 }
2727
2728 /* end of gnunet-service-transport.c */