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