src/transport/transport_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2013 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file transport/transport_api.c
  23  * @brief library to access the low-level P2P IO service
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - test test test
  28  */
  29 #include "platform.h"
  30 #include "gnunet_util_lib.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_arm_service.h"
  33 #include "gnunet_hello_lib.h"
  34 #include "gnunet_protocols.h"
  35 #include "gnunet_transport_service.h"
  36 #include "transport.h"
  37
  38 #define LOG(kind,...) GNUNET_log_from (kind, "transport-api",__VA_ARGS__)
  39
  40 /**
  41  * How large to start with for the hashmap of neighbours.
  42  */
  43 #define STARTING_NEIGHBOURS_SIZE 16
  44
  45 /**
  46  * Handle for a message that should be transmitted to the service.
  47  * Used for both control messages and normal messages.
  48  */
  49 struct GNUNET_TRANSPORT_TransmitHandle
  50 {
  51
  52   /**
  53    * We keep all requests in a DLL.
  54    */
  55   struct GNUNET_TRANSPORT_TransmitHandle *next;
  56
  57   /**
  58    * We keep all requests in a DLL.
  59    */
  60   struct GNUNET_TRANSPORT_TransmitHandle *prev;
  61
  62   /**
  63    * Neighbour for this handle, NULL for control messages.
  64    */
  65   struct Neighbour *neighbour;
  66
  67   /**
  68    * Function to call when @e notify_size bytes are available
  69    * for transmission.
  70    */
  71   GNUNET_TRANSPORT_TransmitReadyNotify notify;
  72
  73   /**
  74    * Closure for @e notify.
  75    */
  76   void *notify_cls;
  77
  78   /**
  79    * Timeout for this request, 0 for control messages.
  80    */
  81   struct GNUNET_TIME_Absolute timeout;
  82
  83   /**
  84    * Task to trigger request timeout if the request is stalled due to
  85    * congestion.
  86    */
  87   struct GNUNET_SCHEDULER_Task * timeout_task;
  88
  89   /**
  90    * How many bytes is our notify callback waiting for?
  91    */
  92   size_t notify_size;
  93
  94 };
  95
  96
  97 /**
  98  * Entry in hash table of all of our current (connected) neighbours.
  99  */
 100 struct Neighbour
 101 {
 102   /**
 103    * Overall transport handle.
 104    */
 105   struct GNUNET_TRANSPORT_Handle *h;
 106
 107   /**
 108    * Active transmit handle or NULL.
 109    */
 110   struct GNUNET_TRANSPORT_TransmitHandle *th;
 111
 112   /**
 113    * Identity of this neighbour.
 114    */
 115   struct GNUNET_PeerIdentity id;
 116
 117   /**
 118    * Outbound bandwidh tracker.
 119    */
 120   struct GNUNET_BANDWIDTH_Tracker out_tracker;
 121
 122   /**
 123    * Entry in our readyness heap (which is sorted by @e next_ready
 124    * value).  NULL if there is no pending transmission request for
 125    * this neighbour or if we're waiting for @e is_ready to become
 126    * true AFTER the @e out_tracker suggested that this peer's quota
 127    * has been satisfied (so once @e is_ready goes to #GNUNET_YES,
 128    * we should immediately go back into the heap).
 129    */
 130   struct GNUNET_CONTAINER_HeapNode *hn;
 131
 132   /**
 133    * Is this peer currently ready to receive a message?
 134    */
 135   int is_ready;
 136
 137   /**
 138    * Sending consumed more bytes on wire than payload was announced
 139    * This overhead is added to the delay of next sending operation
 140    */
 141   size_t traffic_overhead;
 142 };
 143
 144
 145 /**
 146  * Linked list of functions to call whenever our HELLO is updated.
 147  */
 148 struct GNUNET_TRANSPORT_GetHelloHandle
 149 {
 150
 151   /**
 152    * This is a doubly linked list.
 153    */
 154   struct GNUNET_TRANSPORT_GetHelloHandle *next;
 155
 156   /**
 157    * This is a doubly linked list.
 158    */
 159   struct GNUNET_TRANSPORT_GetHelloHandle *prev;
 160
 161   /**
 162    * Transport handle.
 163    */
 164   struct GNUNET_TRANSPORT_Handle *handle;
 165
 166   /**
 167    * Callback to call once we got our HELLO.
 168    */
 169   GNUNET_TRANSPORT_HelloUpdateCallback rec;
 170
 171   /**
 172    * Task for calling the HelloUpdateCallback when we already have a HELLO
 173    */
 174   struct GNUNET_SCHEDULER_Task * notify_task;
 175
 176   /**
 177    * Closure for @e rec.
 178    */
 179   void *rec_cls;
 180
 181 };
 182
 183
 184 /**
 185  * Entry in linked list for a try-connect request.
 186  */
 187 struct GNUNET_TRANSPORT_TryConnectHandle
 188 {
 189   /**
 190    * For the DLL.
 191    */
 192   struct GNUNET_TRANSPORT_TryConnectHandle *prev;
 193
 194   /**
 195    * For the DLL.
 196    */
 197   struct GNUNET_TRANSPORT_TryConnectHandle *next;
 198
 199   /**
 200    * Peer we should try to connect to.
 201    */
 202   struct GNUNET_PeerIdentity pid;
 203
 204   /**
 205    * Transport service handle this request is part of.
 206    */
 207   struct GNUNET_TRANSPORT_Handle *th;
 208
 209   /**
 210    * Message transmission request to communicate to service.
 211    */
 212   struct GNUNET_TRANSPORT_TransmitHandle *tth;
 213
 214   /**
 215    * Function to call upon completion (of request transmission).
 216    */
 217   GNUNET_TRANSPORT_TryConnectCallback cb;
 218
 219   /**
 220    * Closure for @e cb.
 221    */
 222   void *cb_cls;
 223
 224 };
 225
 226
 227 /**
 228  * Entry in linked list for all try-disconnect requests
 229  */
 230 struct GNUNET_TRANSPORT_TryDisconnectHandle
 231 {
 232   /**
 233    * For the DLL.
 234    */
 235   struct GNUNET_TRANSPORT_TryDisconnectHandle *prev;
 236
 237   /**
 238    * For the DLL.
 239    */
 240   struct GNUNET_TRANSPORT_TryDisconnectHandle *next;
 241
 242   /**
 243    * Peer we should try to connect to.
 244    */
 245   struct GNUNET_PeerIdentity pid;
 246
 247   /**
 248    * Transport service handle this request is part of.
 249    */
 250   struct GNUNET_TRANSPORT_Handle *th;
 251
 252   /**
 253    * Message transmission request to communicate to service.
 254    */
 255   struct GNUNET_TRANSPORT_TransmitHandle *tth;
 256
 257   /**
 258    * Function to call upon completion (of request transmission).
 259    */
 260   GNUNET_TRANSPORT_TryDisconnectCallback cb;
 261
 262   /**
 263    * Closure for @e cb.
 264    */
 265   void *cb_cls;
 266
 267 };
 268
 269
 270 /**
 271  * Entry in linked list for all offer-HELLO requests.
 272  */
 273 struct GNUNET_TRANSPORT_OfferHelloHandle
 274 {
 275   /**
 276    * For the DLL.
 277    */
 278   struct GNUNET_TRANSPORT_OfferHelloHandle *prev;
 279
 280   /**
 281    * For the DLL.
 282    */
 283   struct GNUNET_TRANSPORT_OfferHelloHandle *next;
 284
 285   /**
 286    * Transport service handle we use for transmission.
 287    */
 288   struct GNUNET_TRANSPORT_Handle *th;
 289
 290   /**
 291    * Transmission handle for this request.
 292    */
 293   struct GNUNET_TRANSPORT_TransmitHandle *tth;
 294
 295   /**
 296    * Function to call once we are done.
 297    */
 298   GNUNET_SCHEDULER_TaskCallback cont;
 299
 300   /**
 301    * Closure for @e cont
 302    */
 303   void *cls;
 304
 305   /**
 306    * The HELLO message to be transmitted.
 307    */
 308   struct GNUNET_MessageHeader *msg;
 309 };
 310
 311
 312 /**
 313  * Handle for the transport service (includes all of the
 314  * state for the transport service).
 315  */
 316 struct GNUNET_TRANSPORT_Handle
 317 {
 318
 319   /**
 320    * Closure for the callbacks.
 321    */
 322   void *cls;
 323
 324   /**
 325    * Function to call for received data.
 326    */
 327   GNUNET_TRANSPORT_ReceiveCallback rec;
 328
 329   /**
 330    * function to call on connect events
 331    */
 332   GNUNET_TRANSPORT_NotifyConnect nc_cb;
 333
 334   /**
 335    * function to call on disconnect events
 336    */
 337   GNUNET_TRANSPORT_NotifyDisconnect nd_cb;
 338
 339   /**
 340    * function to call on excess bandwidth events
 341    */
 342   GNUNET_TRANSPORT_NotifyExcessBandwidth neb_cb;
 343
 344   /**
 345    * Head of DLL of control messages.
 346    */
 347   struct GNUNET_TRANSPORT_TransmitHandle *control_head;
 348
 349   /**
 350    * Tail of DLL of control messages.
 351    */
 352   struct GNUNET_TRANSPORT_TransmitHandle *control_tail;
 353
 354   /**
 355    * The current HELLO message for this peer.  Updated
 356    * whenever transports change their addresses.
 357    */
 358   struct GNUNET_MessageHeader *my_hello;
 359
 360   /**
 361    * My client connection to the transport service.
 362    */
 363   struct GNUNET_CLIENT_Connection *client;
 364
 365   /**
 366    * Handle to our registration with the client for notification.
 367    */
 368   struct GNUNET_CLIENT_TransmitHandle *cth;
 369
 370   /**
 371    * Linked list of pending requests for our HELLO.
 372    */
 373   struct GNUNET_TRANSPORT_GetHelloHandle *hwl_head;
 374
 375   /**
 376    * Linked list of pending requests for our HELLO.
 377    */
 378   struct GNUNET_TRANSPORT_GetHelloHandle *hwl_tail;
 379
 380   /**
 381    * Linked list of pending try connect requests head
 382    */
 383   struct GNUNET_TRANSPORT_TryConnectHandle *tc_head;
 384
 385   /**
 386    * Linked list of pending try connect requests tail
 387    */
 388   struct GNUNET_TRANSPORT_TryConnectHandle *tc_tail;
 389
 390   /**
 391    * Linked list of pending try disconnect requests head
 392    */
 393   struct GNUNET_TRANSPORT_TryDisconnectHandle *td_head;
 394
 395   /**
 396    * Linked list of pending try connect requests tail
 397    */
 398   struct GNUNET_TRANSPORT_TryDisconnectHandle *td_tail;
 399
 400   /**
 401    * Linked list of pending offer HELLO requests head
 402    */
 403   struct GNUNET_TRANSPORT_OfferHelloHandle *oh_head;
 404
 405   /**
 406    * Linked list of pending offer HELLO requests tail
 407    */
 408   struct GNUNET_TRANSPORT_OfferHelloHandle *oh_tail;
 409
 410   /**
 411    * My configuration.
 412    */
 413   const struct GNUNET_CONFIGURATION_Handle *cfg;
 414
 415   /**
 416    * Hash map of the current connected neighbours of this peer.
 417    * Maps peer identities to `struct Neighbour` entries.
 418    */
 419   struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 420
 421   /**
 422    * Heap sorting peers with pending messages by the timestamps that
 423    * specify when we could next send a message to the respective peer.
 424    * Excludes control messages (which can always go out immediately).
 425    * Maps time stamps to 'struct Neighbour' entries.
 426    */
 427   struct GNUNET_CONTAINER_Heap *ready_heap;
 428
 429   /**
 430    * Peer identity as assumed by this process, or all zeros.
 431    */
 432   struct GNUNET_PeerIdentity self;
 433
 434   /**
 435    * ID of the task trying to reconnect to the service.
 436    */
 437   struct GNUNET_SCHEDULER_Task * reconnect_task;
 438
 439   /**
 440    * ID of the task trying to trigger transmission for a peer while
 441    * maintaining bandwidth quotas.  In use if there are no control
 442    * messages and the smallest entry in the 'ready_heap' has a time
 443    * stamp in the future.
 444    */
 445   struct GNUNET_SCHEDULER_Task * quota_task;
 446
 447   /**
 448    * Delay until we try to reconnect.
 449    */
 450   struct GNUNET_TIME_Relative reconnect_delay;
 451
 452   /**
 453    * Should we check that @e self matches what the service thinks?
 454    * (if #GNUNET_NO, then @e self is all zeros!).
 455    */
 456   int check_self;
 457
 458   /**
 459    * Reconnect in progress
 460    */
 461   int reconnecting;
 462 };
 463
 464
 465 /**
 466  * Schedule the task to send one message, either from the control
 467  * list or the peer message queues  to the service.
 468  *
 469  * @param h transport service to schedule a transmission for
 470  */
 471 static void
 472 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h);
 473
 474
 475 /**
 476  * Function that will schedule the job that will try
 477  * to connect us again to the client.
 478  *
 479  * @param h transport service to reconnect
 480  */
 481 static void
 482 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h);
 483
 484
 485 /**
 486  * Get the neighbour list entry for the given peer
 487  *
 488  * @param h our context
 489  * @param peer peer to look up
 490  * @return NULL if no such peer entry exists
 491  */
 492 static struct Neighbour *
 493 neighbour_find (struct GNUNET_TRANSPORT_Handle *h,
 494                 const struct GNUNET_PeerIdentity *peer)
 495 {
 496   return GNUNET_CONTAINER_multipeermap_get (h->neighbours,
 497                                             peer);
 498 }
 499
 500
 501 /**
 502  * The outbound quota has changed in a way that may require
 503  * us to reset the timeout.  Update the timeout.
 504  *
 505  * @param cls the `struct Neighbour` for which the timeout changed
 506  */
 507 static void
 508 outbound_bw_tracker_update (void *cls)
 509 {
 510   struct Neighbour *n = cls;
 511   struct GNUNET_TIME_Relative delay;
 512
 513   if (NULL == n->hn)
 514     return;
 515   delay = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker,
 516                                               n->th->notify_size + n->traffic_overhead);
 517   LOG (GNUNET_ERROR_TYPE_DEBUG,
 518        "New outbound delay %llu us\n",
 519        GNUNET_STRINGS_relative_time_to_string (delay,
 520                                                GNUNET_NO));
 521   GNUNET_CONTAINER_heap_update_cost (n->h->ready_heap,
 522       n->hn, delay.rel_value_us);
 523   schedule_transmission (n->h);
 524 }
 525
 526
 527 /**
 528  * Function called by the bandwidth tracker if we have excess
 529  * bandwidth.
 530  *
 531  * @param cls the `struct Neighbour` that has excess bandwidth
 532  */
 533 static void
 534 notify_excess_cb (void *cls)
 535 {
 536   struct Neighbour *n = cls;
 537   struct GNUNET_TRANSPORT_Handle *h = n->h;
 538
 539   if (NULL != h->neb_cb)
 540     h->neb_cb (h->cls,
 541                &n->id);
 542 }
 543
 544
 545 /**
 546  * Add neighbour to our list
 547  *
 548  * @return NULL if this API is currently disconnecting from the service
 549  */
 550 static struct Neighbour *
 551 neighbour_add (struct GNUNET_TRANSPORT_Handle *h,
 552                const struct GNUNET_PeerIdentity *pid)
 553 {
 554   struct Neighbour *n;
 555
 556   LOG (GNUNET_ERROR_TYPE_DEBUG,
 557        "Creating entry for neighbour `%4s'.\n",
 558        GNUNET_i2s (pid));
 559   n = GNUNET_new (struct Neighbour);
 560   n->id = *pid;
 561   n->h = h;
 562   n->is_ready = GNUNET_YES;
 563   n->traffic_overhead = 0;
 564   GNUNET_BANDWIDTH_tracker_init2 (&n->out_tracker,
 565                                   &outbound_bw_tracker_update,
 566                                   n,
 567                                   GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
 568                                   MAX_BANDWIDTH_CARRY_S,
 569                                   &notify_excess_cb,
 570                                   n);
 571   GNUNET_assert (GNUNET_OK ==
 572                  GNUNET_CONTAINER_multipeermap_put (h->neighbours,
 573                                                     &n->id,
 574                                                     n,
 575                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 576   return n;
 577 }
 578
 579
 580 /**
 581  * Iterator over hash map entries, for deleting state of a neighbour.
 582  *
 583  * @param cls the `struct GNUNET_TRANSPORT_Handle *`
 584  * @param key peer identity
 585  * @param value value in the hash map, the neighbour entry to delete
 586  * @return #GNUNET_YES if we should continue to
 587  *         iterate,
 588  *         #GNUNET_NO if not.
 589  */
 590 static int
 591 neighbour_delete (void *cls,
 592                   const struct GNUNET_PeerIdentity *key,
 593                   void *value)
 594 {
 595   struct GNUNET_TRANSPORT_Handle *handle = cls;
 596   struct Neighbour *n = value;
 597
 598   GNUNET_BANDWIDTH_tracker_notification_stop (&n->out_tracker);
 599   if (NULL != handle->nd_cb)
 600     handle->nd_cb (handle->cls,
 601                    &n->id);
 602   GNUNET_assert (NULL == n->th);
 603   GNUNET_assert (NULL == n->hn);
 604   GNUNET_assert (GNUNET_YES ==
 605                  GNUNET_CONTAINER_multipeermap_remove (handle->neighbours,
 606                                                        key,
 607                                                        n));
 608   GNUNET_free (n);
 609   return GNUNET_YES;
 610 }
 611
 612
 613 /**
 614  * Function we use for handling incoming messages.
 615  *
 616  * @param cls closure, a `struct GNUNET_TRANSPORT_Handle *`
 617  * @param msg message received, NULL on timeout or fatal error
 618  */
 619 static void
 620 demultiplexer (void *cls,
 621                const struct GNUNET_MessageHeader *msg)
 622 {
 623   struct GNUNET_TRANSPORT_Handle *h = cls;
 624   const struct DisconnectInfoMessage *dim;
 625   const struct ConnectInfoMessage *cim;
 626   const struct InboundMessage *im;
 627   const struct GNUNET_MessageHeader *imm;
 628   const struct SendOkMessage *okm;
 629   const struct QuotaSetMessage *qm;
 630   struct GNUNET_TRANSPORT_GetHelloHandle *hwl;
 631   struct GNUNET_TRANSPORT_GetHelloHandle *next_hwl;
 632   struct Neighbour *n;
 633   struct GNUNET_PeerIdentity me;
 634   uint16_t size;
 635   uint32_t bytes_msg;
 636   uint32_t bytes_physical;
 637
 638   GNUNET_assert (NULL != h->client);
 639   if (GNUNET_YES == h->reconnecting)
 640   {
 641     return;
 642   }
 643   if (NULL == msg)
 644   {
 645     LOG (GNUNET_ERROR_TYPE_DEBUG,
 646          "Error receiving from transport service, disconnecting temporarily.\n");
 647     h->reconnecting = GNUNET_YES;
 648     disconnect_and_schedule_reconnect (h);
 649     return;
 650   }
 651   GNUNET_CLIENT_receive (h->client,
 652                          &demultiplexer,
 653                          h,
 654                          GNUNET_TIME_UNIT_FOREVER_REL);
 655   size = ntohs (msg->size);
 656   switch (ntohs (msg->type))
 657   {
 658   case GNUNET_MESSAGE_TYPE_HELLO:
 659     if (GNUNET_OK !=
 660         GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) msg,
 661                              &me))
 662     {
 663       GNUNET_break (0);
 664       break;
 665     }
 666     LOG (GNUNET_ERROR_TYPE_DEBUG,
 667          "Receiving (my own) HELLO message (%u bytes), I am `%4s'.\n",
 668          (unsigned int) size,
 669          GNUNET_i2s (&me));
 670     GNUNET_free_non_null (h->my_hello);
 671     h->my_hello = NULL;
 672     if (size < sizeof (struct GNUNET_MessageHeader))
 673     {
 674       GNUNET_break (0);
 675       break;
 676     }
 677     h->my_hello = GNUNET_copy_message (msg);
 678     hwl = h->hwl_head;
 679     while (NULL != hwl)
 680     {
 681       next_hwl = hwl->next;
 682       hwl->rec (hwl->rec_cls,
 683                 h->my_hello);
 684       hwl = next_hwl;
 685     }
 686     break;
 687   case GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT:
 688     if (size < sizeof (struct ConnectInfoMessage))
 689     {
 690       GNUNET_break (0);
 691       break;
 692     }
 693     cim = (const struct ConnectInfoMessage *) msg;
 694     if (size !=
 695         sizeof (struct ConnectInfoMessage))
 696     {
 697       GNUNET_break (0);
 698       break;
 699     }
 700     LOG (GNUNET_ERROR_TYPE_DEBUG,
 701          "Receiving CONNECT message for `%4s'.\n",
 702          GNUNET_i2s (&cim->id));
 703     n = neighbour_find (h, &cim->id);
 704     if (NULL != n)
 705     {
 706       GNUNET_break (0);
 707       break;
 708     }
 709     n = neighbour_add (h,
 710                        &cim->id);
 711     LOG (GNUNET_ERROR_TYPE_DEBUG,
 712          "Receiving CONNECT message for `%4s' with quota %u\n",
 713          GNUNET_i2s (&cim->id),
 714          ntohl (cim->quota_out.value__));
 715     GNUNET_BANDWIDTH_tracker_update_quota (&n->out_tracker,
 716                                            cim->quota_out);
 717     if (NULL != h->nc_cb)
 718       h->nc_cb (h->cls,
 719                 &n->id);
 720     break;
 721   case GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT:
 722     if (size != sizeof (struct DisconnectInfoMessage))
 723     {
 724       GNUNET_break (0);
 725       break;
 726     }
 727     dim = (const struct DisconnectInfoMessage *) msg;
 728     GNUNET_break (ntohl (dim->reserved) == 0);
 729     LOG (GNUNET_ERROR_TYPE_DEBUG,
 730          "Receiving DISCONNECT message for `%4s'.\n",
 731          GNUNET_i2s (&dim->peer));
 732     n = neighbour_find (h, &dim->peer);
 733     if (NULL == n)
 734     {
 735       GNUNET_break (0);
 736       break;
 737     }
 738     neighbour_delete (h,
 739                       &dim->peer,
 740                       n);
 741     break;
 742   case GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK:
 743     if (size != sizeof (struct SendOkMessage))
 744     {
 745       GNUNET_break (0);
 746       break;
 747     }
 748     okm = (const struct SendOkMessage *) msg;
 749     bytes_msg = ntohl (okm->bytes_msg);
 750     bytes_physical = ntohl (okm->bytes_physical);
 751     LOG (GNUNET_ERROR_TYPE_DEBUG,
 752          "Receiving SEND_OK message, transmission %s.\n",
 753          ntohl (okm->success) == GNUNET_OK ? "succeeded" : "failed");
 754
 755     n = neighbour_find (h, &okm->peer);
 756     if (NULL == n)
 757       break;
 758
 759     if (bytes_physical >= bytes_msg)
 760     {
 761       LOG (GNUNET_ERROR_TYPE_DEBUG,
 762            "Overhead for %u byte message: %u\n",
 763            bytes_msg,
 764            bytes_physical - bytes_msg);
 765       n->traffic_overhead += bytes_physical - bytes_msg;
 766     }
 767     GNUNET_break (GNUNET_NO == n->is_ready);
 768     n->is_ready = GNUNET_YES;
 769     if ((NULL != n->th) && (NULL == n->hn))
 770     {
 771       GNUNET_assert (NULL != n->th->timeout_task);
 772       GNUNET_SCHEDULER_cancel (n->th->timeout_task);
 773       n->th->timeout_task = NULL;
 774       /* we've been waiting for this (congestion, not quota,
 775        * caused delayed transmission) */
 776       n->hn = GNUNET_CONTAINER_heap_insert (h->ready_heap, n, 0);
 777       schedule_transmission (h);
 778     }
 779     break;
 780   case GNUNET_MESSAGE_TYPE_TRANSPORT_RECV:
 781     if (size <
 782         sizeof (struct InboundMessage) + sizeof (struct GNUNET_MessageHeader))
 783     {
 784       GNUNET_break (0);
 785       break;
 786     }
 787     im = (const struct InboundMessage *) msg;
 788     imm = (const struct GNUNET_MessageHeader *) &im[1];
 789     if (ntohs (imm->size) + sizeof (struct InboundMessage) != size)
 790     {
 791       GNUNET_break (0);
 792       break;
 793     }
 794     LOG (GNUNET_ERROR_TYPE_DEBUG,
 795          "Received message of type %u from `%4s'.\n",
 796          ntohs (imm->type), GNUNET_i2s (&im->peer));
 797     n = neighbour_find (h, &im->peer);
 798     if (NULL == n)
 799     {
 800       GNUNET_break (0);
 801       break;
 802     }
 803     if (NULL != h->rec)
 804       h->rec (h->cls,
 805               &im->peer,
 806               imm);
 807     break;
 808   case GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA:
 809     if (size != sizeof (struct QuotaSetMessage))
 810     {
 811       GNUNET_break (0);
 812       break;
 813     }
 814     qm = (const struct QuotaSetMessage *) msg;
 815     n = neighbour_find (h, &qm->peer);
 816     if (NULL == n)
 817       break;
 818     LOG (GNUNET_ERROR_TYPE_DEBUG,
 819          "Receiving SET_QUOTA message for `%4s' with quota %u\n",
 820          GNUNET_i2s (&qm->peer),
 821          ntohl (qm->quota.value__));
 822     GNUNET_BANDWIDTH_tracker_update_quota (&n->out_tracker,
 823                                            qm->quota);
 824     break;
 825   default:
 826     LOG (GNUNET_ERROR_TYPE_ERROR,
 827          _("Received unexpected message of type %u in %s:%u\n"),
 828          ntohs (msg->type),
 829          __FILE__,
 830          __LINE__);
 831     GNUNET_break (0);
 832     break;
 833   }
 834 }
 835
 836
 837 /**
 838  * A transmission request could not be satisfied because of
 839  * network congestion.  Notify the initiator and clean up.
 840  *
 841  * @param cls the `struct GNUNET_TRANSPORT_TransmitHandle`
 842  * @param tc scheduler context
 843  */
 844 static void
 845 timeout_request_due_to_congestion (void *cls,
 846                                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 847 {
 848   struct GNUNET_TRANSPORT_TransmitHandle *th = cls;
 849   struct Neighbour *n = th->neighbour;
 850
 851   n->th->timeout_task = NULL;
 852   GNUNET_assert (th == n->th);
 853   GNUNET_assert (NULL == n->hn);
 854   n->th = NULL;
 855   th->notify (th->notify_cls,
 856               0,
 857               NULL);
 858   GNUNET_free (th);
 859 }
 860
 861
 862 /**
 863  * Transmit message(s) to service.
 864  *
 865  * @param cls handle to transport
 866  * @param size number of bytes available in @a buf
 867  * @param buf where to copy the message
 868  * @return number of bytes copied to @a buf
 869  */
 870 static size_t
 871 transport_notify_ready (void *cls,
 872                         size_t size,
 873                         void *buf)
 874 {
 875   struct GNUNET_TRANSPORT_Handle *h = cls;
 876   struct GNUNET_TRANSPORT_TransmitHandle *th;
 877   struct Neighbour *n;
 878   char *cbuf;
 879   struct OutboundMessage obm;
 880   size_t ret;
 881   size_t nret;
 882   size_t mret;
 883
 884   GNUNET_assert (NULL != h->client);
 885   h->cth = NULL;
 886   if (NULL == buf)
 887   {
 888     /* transmission failed */
 889     disconnect_and_schedule_reconnect (h);
 890     return 0;
 891   }
 892
 893   cbuf = buf;
 894   ret = 0;
 895   /* first send control messages */
 896   while ( (NULL != (th = h->control_head)) &&
 897           (th->notify_size <= size) )
 898   {
 899     GNUNET_CONTAINER_DLL_remove (h->control_head,
 900                                  h->control_tail,
 901                                  th);
 902     nret = th->notify (th->notify_cls, size, &cbuf[ret]);
 903     LOG (GNUNET_ERROR_TYPE_DEBUG,
 904          "Added %u bytes of control message at %u\n",
 905          nret,
 906          ret);
 907     GNUNET_free (th);
 908     ret += nret;
 909     size -= nret;
 910   }
 911
 912   /* then, if possible and no control messages pending, send data messages */
 913   while ((NULL == h->control_head) &&
 914          (NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap))))
 915   {
 916     if (GNUNET_YES != n->is_ready)
 917     {
 918       /* peer not ready, wait for notification! */
 919       GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 920       n->hn = NULL;
 921       GNUNET_assert (NULL == n->th->timeout_task);
 922       n->th->timeout_task =
 923           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
 924                                         (n->th->timeout),
 925                                         &timeout_request_due_to_congestion,
 926                                         n->th);
 927       continue;
 928     }
 929     th = n->th;
 930     if (th->notify_size + sizeof (struct OutboundMessage) > size)
 931       break;                    /* does not fit */
 932     if (GNUNET_BANDWIDTH_tracker_get_delay
 933         (&n->out_tracker, th->notify_size).rel_value_us > 0)
 934       break;                    /* too early */
 935     GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 936     n->hn = NULL;
 937     n->th = NULL;
 938     n->is_ready = GNUNET_NO;
 939     GNUNET_assert (size >= sizeof (struct OutboundMessage));
 940     mret =
 941         th->notify (th->notify_cls, size - sizeof (struct OutboundMessage),
 942                     &cbuf[ret + sizeof (struct OutboundMessage)]);
 943     GNUNET_assert (mret <= size - sizeof (struct OutboundMessage));
 944     if (mret != 0)
 945     {
 946       GNUNET_assert (mret + sizeof (struct OutboundMessage) <
 947                      GNUNET_SERVER_MAX_MESSAGE_SIZE);
 948       obm.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SEND);
 949       obm.header.size = htons (mret + sizeof (struct OutboundMessage));
 950       obm.reserved = htonl (0);
 951       obm.timeout =
 952           GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
 953                                      (th->timeout));
 954       obm.peer = n->id;
 955       memcpy (&cbuf[ret], &obm, sizeof (struct OutboundMessage));
 956       ret += (mret + sizeof (struct OutboundMessage));
 957       size -= (mret + sizeof (struct OutboundMessage));
 958       GNUNET_BANDWIDTH_tracker_consume (&n->out_tracker, mret);
 959     }
 960     GNUNET_free (th);
 961   }
 962   /* if there are more pending messages, try to schedule those */
 963   schedule_transmission (h);
 964   LOG (GNUNET_ERROR_TYPE_DEBUG,
 965        "Transmitting %u bytes to transport service\n",
 966        ret);
 967   return ret;
 968 }
 969
 970
 971 /**
 972  * Schedule the task to send one message, either from the control
 973  * list or the peer message queues  to the service.
 974  *
 975  * @param cls transport service to schedule a transmission for
 976  * @param tc scheduler context
 977  */
 978 static void
 979 schedule_transmission_task (void *cls,
 980                             const struct GNUNET_SCHEDULER_TaskContext *tc)
 981 {
 982   struct GNUNET_TRANSPORT_Handle *h = cls;
 983   size_t size;
 984   struct GNUNET_TRANSPORT_TransmitHandle *th;
 985   struct Neighbour *n;
 986
 987   h->quota_task = NULL;
 988   GNUNET_assert (NULL != h->client);
 989   /* destroy all requests that have timed out */
 990   while ((NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap))) &&
 991          (0 == GNUNET_TIME_absolute_get_remaining (n->th->timeout).rel_value_us))
 992   {
 993     /* notify client that the request could not be satisfied within
 994      * the given time constraints */
 995     th = n->th;
 996     n->th = NULL;
 997     GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 998     n->hn = NULL;
 999     LOG (GNUNET_ERROR_TYPE_DEBUG,
1000          "Signalling timeout for transmission to peer %s due to congestion\n",
1001          GNUNET_i2s (&n->id));
1002     GNUNET_assert (0 == th->notify (th->notify_cls, 0, NULL));
1003     GNUNET_free (th);
1004   }
1005   if (NULL != h->cth)
1006     return;
1007   if (NULL != h->control_head)
1008   {
1009     size = h->control_head->notify_size;
1010   }
1011   else
1012   {
1013     n = GNUNET_CONTAINER_heap_peek (h->ready_heap);
1014     if (NULL == n)
1015       return;                   /* no pending messages */
1016     size = n->th->notify_size + sizeof (struct OutboundMessage);
1017   }
1018   LOG (GNUNET_ERROR_TYPE_DEBUG,
1019        "Calling notify_transmit_ready\n");
1020   h->cth =
1021       GNUNET_CLIENT_notify_transmit_ready (h->client, size,
1022                                            GNUNET_TIME_UNIT_FOREVER_REL,
1023                                            GNUNET_NO, &transport_notify_ready,
1024                                            h);
1025   GNUNET_assert (NULL != h->cth);
1026 }
1027
1028
1029 /**
1030  * Schedule the task to send one message, either from the control
1031  * list or the peer message queues  to the service.
1032  *
1033  * @param h transport service to schedule a transmission for
1034  */
1035 static void
1036 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h)
1037 {
1038   struct GNUNET_TIME_Relative delay;
1039   struct Neighbour *n;
1040
1041   GNUNET_assert (NULL != h->client);
1042   if (h->quota_task != NULL)
1043   {
1044     GNUNET_SCHEDULER_cancel (h->quota_task);
1045     h->quota_task = NULL;
1046   }
1047   if (NULL != h->control_head)
1048     delay = GNUNET_TIME_UNIT_ZERO;
1049   else if (NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap)))
1050   {
1051     delay =
1052         GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker,
1053                                             n->th->notify_size + n->traffic_overhead);
1054     n->traffic_overhead = 0;
1055   }
1056   else
1057     return;                     /* no work to be done */
1058   LOG (GNUNET_ERROR_TYPE_DEBUG,
1059        "Scheduling next transmission to service in %s\n",
1060        GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
1061   h->quota_task =
1062       GNUNET_SCHEDULER_add_delayed (delay, &schedule_transmission_task, h);
1063 }
1064
1065
1066 /**
1067  * Queue control request for transmission to the transport
1068  * service.
1069  *
1070  * @param h handle to the transport service
1071  * @param size number of bytes to be transmitted
1072  * @param notify function to call to get the content
1073  * @param notify_cls closure for @a notify
1074  * @return a `struct GNUNET_TRANSPORT_TransmitHandle`
1075  */
1076 static struct GNUNET_TRANSPORT_TransmitHandle *
1077 schedule_control_transmit (struct GNUNET_TRANSPORT_Handle *h,
1078                            size_t size,
1079                            GNUNET_TRANSPORT_TransmitReadyNotify notify,
1080                            void *notify_cls)
1081 {
1082   struct GNUNET_TRANSPORT_TransmitHandle *th;
1083
1084   LOG (GNUNET_ERROR_TYPE_DEBUG,
1085        "Control transmit of %u bytes requested\n",
1086        size);
1087   th = GNUNET_new (struct GNUNET_TRANSPORT_TransmitHandle);
1088   th->notify = notify;
1089   th->notify_cls = notify_cls;
1090   th->notify_size = size;
1091   GNUNET_CONTAINER_DLL_insert_tail (h->control_head, h->control_tail, th);
1092   schedule_transmission (h);
1093   return th;
1094 }
1095
1096
1097 /**
1098  * Transmit START message to service.
1099  *
1100  * @param cls unused
1101  * @param size number of bytes available in @a buf
1102  * @param buf where to copy the message
1103  * @return number of bytes copied to @a buf
1104  */
1105 static size_t
1106 send_start (void *cls, size_t size, void *buf)
1107 {
1108   struct GNUNET_TRANSPORT_Handle *h = cls;
1109   struct StartMessage s;
1110   uint32_t options;
1111
1112   if (NULL == buf)
1113   {
1114     /* Can only be shutdown, just give up */
1115     LOG (GNUNET_ERROR_TYPE_DEBUG,
1116          "Shutdown while trying to transmit START request.\n");
1117     return 0;
1118   }
1119   LOG (GNUNET_ERROR_TYPE_DEBUG,
1120        "Transmitting START request.\n");
1121   GNUNET_assert (size >= sizeof (struct StartMessage));
1122   s.header.size = htons (sizeof (struct StartMessage));
1123   s.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_START);
1124   options = 0;
1125   if (h->check_self)
1126     options |= 1;
1127   if (h->rec != NULL)
1128     options |= 2;
1129   s.options = htonl (options);
1130   s.self = h->self;
1131   memcpy (buf, &s, sizeof (struct StartMessage));
1132   GNUNET_CLIENT_receive (h->client, &demultiplexer, h,
1133                          GNUNET_TIME_UNIT_FOREVER_REL);
1134   return sizeof (struct StartMessage);
1135 }
1136
1137
1138 /**
1139  * Try again to connect to transport service.
1140  *
1141  * @param cls the handle to the transport service
1142  * @param tc scheduler context
1143  */
1144 static void
1145 reconnect (void *cls,
1146            const struct GNUNET_SCHEDULER_TaskContext *tc)
1147 {
1148   struct GNUNET_TRANSPORT_Handle *h = cls;
1149
1150   h->reconnect_task = NULL;
1151   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1152   {
1153     /* shutdown, just give up */
1154     return;
1155   }
1156   LOG (GNUNET_ERROR_TYPE_DEBUG,
1157        "Connecting to transport service.\n");
1158   GNUNET_assert (NULL == h->client);
1159   GNUNET_assert (NULL == h->control_head);
1160   GNUNET_assert (NULL == h->control_tail);
1161   h->reconnecting = GNUNET_NO;
1162   h->client = GNUNET_CLIENT_connect ("transport", h->cfg);
1163
1164   GNUNET_assert (NULL != h->client);
1165   schedule_control_transmit (h, sizeof (struct StartMessage),
1166                              &send_start, h);
1167 }
1168
1169
1170 /**
1171  * Function that will schedule the job that will try
1172  * to connect us again to the client.
1173  *
1174  * @param h transport service to reconnect
1175  */
1176 static void
1177 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h)
1178 {
1179   struct GNUNET_TRANSPORT_TransmitHandle *th;
1180
1181   GNUNET_assert (h->reconnect_task == NULL);
1182   if (NULL != h->cth)
1183   {
1184     GNUNET_CLIENT_notify_transmit_ready_cancel (h->cth);
1185     h->cth = NULL;
1186   }
1187   if (NULL != h->client)
1188   {
1189     GNUNET_CLIENT_disconnect (h->client);
1190     h->client = NULL;
1191 /*    LOG (GNUNET_ERROR_TYPE_ERROR,
1192          "Client disconnect done \n");*/
1193   }
1194   /* Forget about all neighbours that we used to be connected to */
1195   GNUNET_CONTAINER_multipeermap_iterate (h->neighbours,
1196                                          &neighbour_delete,
1197                                          h);
1198   if (NULL != h->quota_task)
1199   {
1200     GNUNET_SCHEDULER_cancel (h->quota_task);
1201     h->quota_task = NULL;
1202   }
1203   while ((NULL != (th = h->control_head)))
1204   {
1205     GNUNET_CONTAINER_DLL_remove (h->control_head,
1206                                  h->control_tail,
1207                                  th);
1208     th->notify (th->notify_cls,
1209                 0,
1210                 NULL);
1211     GNUNET_free (th);
1212   }
1213   LOG (GNUNET_ERROR_TYPE_DEBUG,
1214        "Scheduling task to reconnect to transport service in %s.\n",
1215        GNUNET_STRINGS_relative_time_to_string (h->reconnect_delay,
1216                                                GNUNET_YES));
1217   h->reconnect_task =
1218       GNUNET_SCHEDULER_add_delayed (h->reconnect_delay,
1219                                     &reconnect,
1220                                     h);
1221   h->reconnect_delay = GNUNET_TIME_STD_BACKOFF (h->reconnect_delay);
1222 }
1223
1224
1225 /**
1226  * Cancel control request for transmission to the transport service.
1227  *
1228  * @param th handle to the transport service
1229  * @param tth transmit handle to cancel
1230  */
1231 static void
1232 cancel_control_transmit (struct GNUNET_TRANSPORT_Handle *th,
1233                          struct GNUNET_TRANSPORT_TransmitHandle *tth)
1234 {
1235   LOG (GNUNET_ERROR_TYPE_DEBUG,
1236        "Canceling transmit of contral transmission requested\n");
1237   GNUNET_CONTAINER_DLL_remove (th->control_head,
1238                                th->control_tail,
1239                                tth);
1240   GNUNET_free (tth);
1241 }
1242
1243
1244 /**
1245  * Send #GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_CONNECT message to the
1246  * service.
1247  *
1248  * @param cls the `struct GNUNET_TRANSPORT_TryConnectHandle`
1249  * @param size number of bytes available in @a buf
1250  * @param buf where to copy the message
1251  * @return number of bytes copied to @a buf
1252  */
1253 static size_t
1254 send_try_connect (void *cls,
1255                   size_t size,
1256                   void *buf)
1257 {
1258   struct GNUNET_TRANSPORT_TryConnectHandle *tch = cls;
1259   struct TransportRequestConnectMessage msg;
1260
1261   tch->tth = NULL;
1262   if (NULL == buf)
1263   {
1264     LOG (GNUNET_ERROR_TYPE_DEBUG,
1265          "Discarding  `%s' request to `%4s' due to error in transport service connection.\n",
1266          "REQUEST_CONNECT",
1267          GNUNET_i2s (&tch->pid));
1268     if (NULL != tch->cb)
1269       tch->cb (tch->cb_cls,
1270                GNUNET_SYSERR);
1271     GNUNET_TRANSPORT_try_connect_cancel (tch);
1272     return 0;
1273   }
1274   LOG (GNUNET_ERROR_TYPE_DEBUG,
1275        "Transmitting `%s' request with respect to `%4s'.\n",
1276        "REQUEST_CONNECT",
1277        GNUNET_i2s (&tch->pid));
1278   GNUNET_assert (size >= sizeof (struct TransportRequestConnectMessage));
1279   msg.header.size = htons (sizeof (struct TransportRequestConnectMessage));
1280   msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_CONNECT);
1281   msg.reserved = htonl (0);
1282   msg.peer = tch->pid;
1283   memcpy (buf, &msg, sizeof (msg));
1284   if (NULL != tch->cb)
1285     tch->cb (tch->cb_cls, GNUNET_OK);
1286   GNUNET_TRANSPORT_try_connect_cancel (tch);
1287   return sizeof (struct TransportRequestConnectMessage);
1288 }
1289
1290
1291 /**
1292  * Ask the transport service to establish a connection to
1293  * the given peer.
1294  *
1295  * @param handle connection to transport service
1296  * @param target who we should try to connect to
1297  * @param cb callback to be called when request was transmitted to transport
1298  *         service
1299  * @param cb_cls closure for the callback
1300  * @return a `struct GNUNET_TRANSPORT_TryConnectHandle` handle or
1301  *         NULL on failure (cb will not be called)
1302  */
1303 struct GNUNET_TRANSPORT_TryConnectHandle *
1304 GNUNET_TRANSPORT_try_connect (struct GNUNET_TRANSPORT_Handle *handle,
1305                               const struct GNUNET_PeerIdentity *target,
1306                               GNUNET_TRANSPORT_TryConnectCallback cb,
1307                               void *cb_cls)
1308 {
1309   struct GNUNET_TRANSPORT_TryConnectHandle *tch;
1310
1311   if (NULL == handle->client)
1312     return NULL;
1313   tch = GNUNET_new (struct GNUNET_TRANSPORT_TryConnectHandle);
1314   tch->th = handle;
1315   tch->pid = *target;
1316   tch->cb = cb;
1317   tch->cb_cls = cb_cls;
1318   tch->tth = schedule_control_transmit (handle,
1319                                         sizeof (struct TransportRequestConnectMessage),
1320                                         &send_try_connect, tch);
1321   GNUNET_CONTAINER_DLL_insert (handle->tc_head,
1322                                handle->tc_tail,
1323                                tch);
1324   return tch;
1325 }
1326
1327
1328 /**
1329  * Cancel the request to transport to try a connect
1330  * Callback will not be called
1331  *
1332  * @param tch the handle to cancel
1333  */
1334 void
1335 GNUNET_TRANSPORT_try_connect_cancel (struct GNUNET_TRANSPORT_TryConnectHandle *tch)
1336 {
1337   struct GNUNET_TRANSPORT_Handle *th;
1338
1339   th = tch->th;
1340   if (NULL != tch->tth)
1341     cancel_control_transmit (th, tch->tth);
1342   GNUNET_CONTAINER_DLL_remove (th->tc_head,
1343                                th->tc_tail,
1344                                tch);
1345   GNUNET_free (tch);
1346 }
1347
1348
1349 /**
1350  * Send #GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_DISCONNECT message to the
1351  * service.
1352  *
1353  * @param cls the `struct GNUNET_TRANSPORT_TryDisconnectHandle`
1354  * @param size number of bytes available in @a buf
1355  * @param buf where to copy the message
1356  * @return number of bytes copied to @a buf
1357  */
1358 static size_t
1359 send_try_disconnect (void *cls,
1360                      size_t size,
1361                      void *buf)
1362 {
1363   struct GNUNET_TRANSPORT_TryDisconnectHandle *tdh = cls;
1364   struct TransportRequestConnectMessage msg;
1365
1366   tdh->th = NULL;
1367   if (NULL == buf)
1368   {
1369     LOG (GNUNET_ERROR_TYPE_DEBUG,
1370          "Discarding  `%s' request to `%4s' due to error in transport service connection.\n",
1371          "REQUEST_DISCONNECT",
1372          GNUNET_i2s (&tdh->pid));
1373     if (NULL != tdh->cb)
1374       tdh->cb (tdh->cb_cls,
1375                GNUNET_SYSERR);
1376     GNUNET_TRANSPORT_try_disconnect_cancel (tdh);
1377     return 0;
1378   }
1379   LOG (GNUNET_ERROR_TYPE_DEBUG,
1380        "Transmitting `%s' request with respect to `%4s'.\n",
1381        "REQUEST_DISCONNECT",
1382        GNUNET_i2s (&tdh->pid));
1383   GNUNET_assert (size >= sizeof (struct TransportRequestDisconnectMessage));
1384   msg.header.size = htons (sizeof (struct TransportRequestDisconnectMessage));
1385   msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_DISCONNECT);
1386   msg.reserved = htonl (0);
1387   msg.peer = tdh->pid;
1388   memcpy (buf, &msg, sizeof (msg));
1389   if (NULL != tdh->cb)
1390     tdh->cb (tdh->cb_cls, GNUNET_OK);
1391   GNUNET_TRANSPORT_try_disconnect_cancel (tdh);
1392   return sizeof (struct TransportRequestDisconnectMessage);
1393 }
1394
1395
1396 /**
1397  * Ask the transport service to shutdown a connection to
1398  * the given peer.
1399  *
1400  * @param handle connection to transport service
1401  * @param target who we should try to connect to
1402  * @param cb callback to be called when request was transmitted to transport
1403  *         service
1404  * @param cb_cls closure for the callback @a cb
1405  * @return a `struct GNUNET_TRANSPORT_TryDisconnectHandle` handle or
1406  *         NULL on failure (cb will not be called)
1407  */
1408 struct GNUNET_TRANSPORT_TryDisconnectHandle *
1409 GNUNET_TRANSPORT_try_disconnect (struct GNUNET_TRANSPORT_Handle *handle,
1410                                  const struct GNUNET_PeerIdentity *target,
1411                                  GNUNET_TRANSPORT_TryDisconnectCallback cb,
1412                                  void *cb_cls)
1413 {
1414   struct GNUNET_TRANSPORT_TryDisconnectHandle *tdh;
1415
1416   if (NULL == handle->client)
1417     return NULL;
1418   tdh = GNUNET_new (struct GNUNET_TRANSPORT_TryDisconnectHandle);
1419   tdh->th = handle;
1420   tdh->pid = *target;
1421   tdh->cb = cb;
1422   tdh->cb_cls = cb_cls;
1423   tdh->tth = schedule_control_transmit (handle,
1424                                         sizeof (struct TransportRequestDisconnectMessage),
1425                                         &send_try_disconnect, tdh);
1426   GNUNET_CONTAINER_DLL_insert (handle->td_head,
1427                                handle->td_tail,
1428                                tdh);
1429   return tdh;
1430 }
1431
1432
1433 /**
1434  * Cancel the request to transport to try a disconnect
1435  * Callback will not be called
1436  *
1437  * @param tdh the handle to cancel
1438  */
1439 void
1440 GNUNET_TRANSPORT_try_disconnect_cancel (struct GNUNET_TRANSPORT_TryDisconnectHandle *tdh)
1441 {
1442   struct GNUNET_TRANSPORT_Handle *th;
1443
1444   th = tdh->th;
1445   if (NULL != tdh->tth)
1446     cancel_control_transmit (th, tdh->tth);
1447   GNUNET_CONTAINER_DLL_remove (th->td_head,
1448                                th->td_tail,
1449                                tdh);
1450   GNUNET_free (tdh);
1451 }
1452
1453
1454 /**
1455  * Send HELLO message to the service.
1456  *
1457  * @param cls the HELLO message to send
1458  * @param size number of bytes available in @a buf
1459  * @param buf where to copy the message
1460  * @return number of bytes copied to @a buf
1461  */
1462 static size_t
1463 send_hello (void *cls,
1464             size_t size,
1465             void *buf)
1466 {
1467   struct GNUNET_TRANSPORT_OfferHelloHandle *ohh = cls;
1468   struct GNUNET_MessageHeader *msg = ohh->msg;
1469   uint16_t ssize;
1470   struct GNUNET_SCHEDULER_TaskContext tc;
1471
1472   tc.read_ready = NULL;
1473   tc.write_ready = NULL;
1474   tc.reason = GNUNET_SCHEDULER_REASON_TIMEOUT;
1475   if (NULL == buf)
1476   {
1477     LOG (GNUNET_ERROR_TYPE_DEBUG,
1478          "Timeout while trying to transmit `%s' request.\n",
1479          "HELLO");
1480     if (NULL != ohh->cont)
1481       ohh->cont (ohh->cls,
1482                  &tc);
1483     GNUNET_free (msg);
1484     GNUNET_CONTAINER_DLL_remove (ohh->th->oh_head,
1485                                  ohh->th->oh_tail,
1486                                  ohh);
1487     GNUNET_free (ohh);
1488     return 0;
1489   }
1490   LOG (GNUNET_ERROR_TYPE_DEBUG,
1491        "Transmitting `%s' request.\n",
1492        "HELLO");
1493   ssize = ntohs (msg->size);
1494   GNUNET_assert (size >= ssize);
1495   memcpy (buf,
1496           msg,
1497           ssize);
1498   GNUNET_free (msg);
1499   tc.reason = GNUNET_SCHEDULER_REASON_READ_READY;
1500   if (NULL != ohh->cont)
1501     ohh->cont (ohh->cls,
1502                &tc);
1503   GNUNET_CONTAINER_DLL_remove (ohh->th->oh_head,
1504                                ohh->th->oh_tail,
1505                                ohh);
1506   GNUNET_free (ohh);
1507   return ssize;
1508 }
1509
1510
1511 /**
1512  * Send traffic metric message to the service.
1513  *
1514  * @param cls the message to send
1515  * @param size number of bytes available in @a buf
1516  * @param buf where to copy the message
1517  * @return number of bytes copied to @a buf
1518  */
1519 static size_t
1520 send_metric (void *cls,
1521              size_t size,
1522              void *buf)
1523 {
1524   struct TrafficMetricMessage *msg = cls;
1525   uint16_t ssize;
1526
1527   if (NULL == buf)
1528   {
1529     LOG (GNUNET_ERROR_TYPE_DEBUG,
1530          "Timeout while trying to transmit TRAFFIC_METRIC request.\n");
1531     GNUNET_free (msg);
1532     return 0;
1533   }
1534   LOG (GNUNET_ERROR_TYPE_DEBUG,
1535        "Transmitting TRAFFIC_METRIC request.\n");
1536   ssize = ntohs (msg->header.size);
1537   GNUNET_assert (size >= ssize);
1538   memcpy (buf, msg, ssize);
1539   GNUNET_free (msg);
1540   return ssize;
1541 }
1542
1543
1544 /**
1545  * Set transport metrics for a peer and a direction
1546  *
1547  * @param handle transport handle
1548  * @param peer the peer to set the metric for
1549  * @param prop the performance metrics to set
1550  * @param delay_in inbound delay to introduce
1551  * @param delay_out outbound delay to introduce
1552  *
1553  * Note: Delay restrictions in receiving direction will be enforced
1554  * with one message delay.
1555  */
1556 void
1557 GNUNET_TRANSPORT_set_traffic_metric (struct GNUNET_TRANSPORT_Handle *handle,
1558                                      const struct GNUNET_PeerIdentity *peer,
1559                                      const struct GNUNET_ATS_Properties *prop,
1560                                      struct GNUNET_TIME_Relative delay_in,
1561                                      struct GNUNET_TIME_Relative delay_out)
1562 {
1563   struct TrafficMetricMessage *msg;
1564
1565   msg = GNUNET_new (struct TrafficMetricMessage);
1566   msg->header.size = htons (sizeof (struct TrafficMetricMessage));
1567   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_TRAFFIC_METRIC);
1568   msg->reserved = htonl (0);
1569   msg->peer = *peer;
1570   GNUNET_ATS_properties_hton (&msg->properties,
1571                               prop);
1572   msg->delay_in = GNUNET_TIME_relative_hton (delay_in);
1573   msg->delay_out = GNUNET_TIME_relative_hton (delay_out);
1574   schedule_control_transmit (handle,
1575                              sizeof (struct TrafficMetricMessage),
1576                              &send_metric,
1577                              msg);
1578 }
1579
1580
1581 /**
1582  * Offer the transport service the HELLO of another peer.  Note that
1583  * the transport service may just ignore this message if the HELLO is
1584  * malformed or useless due to our local configuration.
1585  *
1586  * @param handle connection to transport service
1587  * @param hello the hello message
1588  * @param cont continuation to call when HELLO has been sent,
1589  *      tc reason #GNUNET_SCHEDULER_REASON_TIMEOUT for fail
1590  *      tc reasong #GNUNET_SCHEDULER_REASON_READ_READY for success
1591  * @param cls closure for continuation
1592  * @return a `struct GNUNET_TRANSPORT_OfferHelloHandle` handle or NULL on failure,
1593  *      in case of failure cont will not be called
1594  *
1595  */
1596 struct GNUNET_TRANSPORT_OfferHelloHandle *
1597 GNUNET_TRANSPORT_offer_hello (struct GNUNET_TRANSPORT_Handle *handle,
1598                               const struct GNUNET_MessageHeader *hello,
1599                               GNUNET_SCHEDULER_TaskCallback cont,
1600                               void *cls)
1601 {
1602   struct GNUNET_TRANSPORT_OfferHelloHandle *ohh;
1603   struct GNUNET_MessageHeader *msg;
1604   struct GNUNET_PeerIdentity peer;
1605   uint16_t size;
1606
1607   if (NULL == handle->client)
1608     return NULL;
1609   GNUNET_break (ntohs (hello->type) == GNUNET_MESSAGE_TYPE_HELLO);
1610   size = ntohs (hello->size);
1611   GNUNET_break (size >= sizeof (struct GNUNET_MessageHeader));
1612   if (GNUNET_OK !=
1613       GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) hello,
1614                            &peer))
1615   {
1616     GNUNET_break (0);
1617     return NULL;
1618   }
1619
1620   msg = GNUNET_malloc (size);
1621   memcpy (msg, hello, size);
1622   LOG (GNUNET_ERROR_TYPE_DEBUG,
1623        "Offering `%s' message of `%4s' to transport for validation.\n", "HELLO",
1624        GNUNET_i2s (&peer));
1625
1626   ohh = GNUNET_new (struct GNUNET_TRANSPORT_OfferHelloHandle);
1627   ohh->th = handle;
1628   ohh->cont = cont;
1629   ohh->cls = cls;
1630   ohh->msg = msg;
1631   ohh->tth = schedule_control_transmit (handle, size,
1632                                         &send_hello, ohh);
1633   GNUNET_CONTAINER_DLL_insert (handle->oh_head, handle->oh_tail, ohh);
1634   return ohh;
1635 }
1636
1637
1638 /**
1639  * Cancel the request to transport to offer the HELLO message
1640  *
1641  * @param ohh the GNUNET_TRANSPORT_OfferHelloHandle to cancel
1642  */
1643 void
1644 GNUNET_TRANSPORT_offer_hello_cancel (struct GNUNET_TRANSPORT_OfferHelloHandle *ohh)
1645 {
1646   struct GNUNET_TRANSPORT_Handle *th = ohh->th;
1647
1648   cancel_control_transmit (ohh->th, ohh->tth);
1649   GNUNET_CONTAINER_DLL_remove (th->oh_head,
1650                                th->oh_tail,
1651                                ohh);
1652   GNUNET_free (ohh->msg);
1653   GNUNET_free (ohh);
1654 }
1655
1656
1657 /**
1658  * Checks if a given peer is connected to us
1659  *
1660  * @param handle connection to transport service
1661  * @param peer the peer to check
1662  * @return #GNUNET_YES (connected) or #GNUNET_NO (disconnected)
1663  */
1664 int
1665 GNUNET_TRANSPORT_check_peer_connected (struct GNUNET_TRANSPORT_Handle *handle,
1666                                        const struct GNUNET_PeerIdentity *peer)
1667 {
1668   if (GNUNET_YES ==
1669       GNUNET_CONTAINER_multipeermap_contains (handle->neighbours,
1670                                               peer))
1671     return GNUNET_YES;
1672   return GNUNET_NO;
1673 }
1674
1675
1676 /**
1677  * Task to call the HelloUpdateCallback of the GetHelloHandle
1678  *
1679  * @param cls the `struct GNUNET_TRANSPORT_GetHelloHandle`
1680  * @param tc the scheduler task context
1681  */
1682 static void
1683 call_hello_update_cb_async (void *cls,
1684                             const struct GNUNET_SCHEDULER_TaskContext *tc)
1685 {
1686   struct GNUNET_TRANSPORT_GetHelloHandle *ghh = cls;
1687
1688   GNUNET_assert (NULL != ghh->handle->my_hello);
1689   GNUNET_assert (NULL != ghh->notify_task);
1690   ghh->notify_task = NULL;
1691   ghh->rec (ghh->rec_cls,
1692             ghh->handle->my_hello);
1693 }
1694
1695
1696 /**
1697  * Obtain the HELLO message for this peer.  The callback given in this function
1698  * is never called synchronously.
1699  *
1700  * @param handle connection to transport service
1701  * @param rec function to call with the HELLO, sender will be our peer
1702  *            identity; message and sender will be NULL on timeout
1703  *            (handshake with transport service pending/failed).
1704  *             cost estimate will be 0.
1705  * @param rec_cls closure for @a rec
1706  * @return handle to cancel the operation
1707  */
1708 struct GNUNET_TRANSPORT_GetHelloHandle *
1709 GNUNET_TRANSPORT_get_hello (struct GNUNET_TRANSPORT_Handle *handle,
1710                             GNUNET_TRANSPORT_HelloUpdateCallback rec,
1711                             void *rec_cls)
1712 {
1713   struct GNUNET_TRANSPORT_GetHelloHandle *hwl;
1714
1715   hwl = GNUNET_new (struct GNUNET_TRANSPORT_GetHelloHandle);
1716   hwl->rec = rec;
1717   hwl->rec_cls = rec_cls;
1718   hwl->handle = handle;
1719   GNUNET_CONTAINER_DLL_insert (handle->hwl_head, handle->hwl_tail, hwl);
1720   if (NULL != handle->my_hello)
1721     hwl->notify_task = GNUNET_SCHEDULER_add_now (&call_hello_update_cb_async,
1722                                                  hwl);
1723   return hwl;
1724 }
1725
1726
1727 /**
1728  * Stop receiving updates about changes to our HELLO message.
1729  *
1730  * @param ghh handle to cancel
1731  */
1732 void
1733 GNUNET_TRANSPORT_get_hello_cancel (struct GNUNET_TRANSPORT_GetHelloHandle *ghh)
1734 {
1735   struct GNUNET_TRANSPORT_Handle *handle = ghh->handle;
1736
1737   if (NULL != ghh->notify_task)
1738     GNUNET_SCHEDULER_cancel (ghh->notify_task);
1739   GNUNET_CONTAINER_DLL_remove (handle->hwl_head, handle->hwl_tail, ghh);
1740   GNUNET_free (ghh);
1741 }
1742
1743
1744 /**
1745  * Connect to the transport service.  Note that the connection may
1746  * complete (or fail) asynchronously.
1747  *
1748  * @param cfg configuration to use
1749  * @param self our own identity (API should check that it matches
1750  *             the identity found by transport), or NULL (no check)
1751  * @param cls closure for the callbacks
1752  * @param rec receive function to call
1753  * @param nc function to call on connect events
1754  * @param nd function to call on disconnect events
1755  * @return NULL on error
1756  */
1757 struct GNUNET_TRANSPORT_Handle *
1758 GNUNET_TRANSPORT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
1759                           const struct GNUNET_PeerIdentity *self,
1760                           void *cls,
1761                           GNUNET_TRANSPORT_ReceiveCallback rec,
1762                           GNUNET_TRANSPORT_NotifyConnect nc,
1763                           GNUNET_TRANSPORT_NotifyDisconnect nd)
1764 {
1765   return GNUNET_TRANSPORT_connect2 (cfg,
1766                                     self,
1767                                     cls,
1768                                     rec,
1769                                     nc,
1770                                     nd,
1771                                     NULL);
1772 }
1773
1774
1775 /**
1776  * Connect to the transport service.  Note that the connection may
1777  * complete (or fail) asynchronously.
1778  *
1779  * @param cfg configuration to use
1780  * @param self our own identity (API should check that it matches
1781  *             the identity found by transport), or NULL (no check)
1782  * @param cls closure for the callbacks
1783  * @param rec receive function to call
1784  * @param nc function to call on connect events
1785  * @param nd function to call on disconnect events
1786  * @param neb function to call if we have excess bandwidth to a peer
1787  * @return NULL on error
1788  */
1789 struct GNUNET_TRANSPORT_Handle *
1790 GNUNET_TRANSPORT_connect2 (const struct GNUNET_CONFIGURATION_Handle *cfg,
1791                            const struct GNUNET_PeerIdentity *self,
1792                            void *cls,
1793                            GNUNET_TRANSPORT_ReceiveCallback rec,
1794                            GNUNET_TRANSPORT_NotifyConnect nc,
1795                            GNUNET_TRANSPORT_NotifyDisconnect nd,
1796                            GNUNET_TRANSPORT_NotifyExcessBandwidth neb)
1797 {
1798   struct GNUNET_TRANSPORT_Handle *ret;
1799
1800   ret = GNUNET_new (struct GNUNET_TRANSPORT_Handle);
1801   if (NULL != self)
1802   {
1803     ret->self = *self;
1804     ret->check_self = GNUNET_YES;
1805   }
1806   ret->cfg = cfg;
1807   ret->cls = cls;
1808   ret->rec = rec;
1809   ret->nc_cb = nc;
1810   ret->nd_cb = nd;
1811   ret->neb_cb = neb;
1812   ret->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
1813   LOG (GNUNET_ERROR_TYPE_DEBUG,
1814        "Connecting to transport service.\n");
1815   ret->client = GNUNET_CLIENT_connect ("transport",
1816                                        cfg);
1817   if (NULL == ret->client)
1818   {
1819     GNUNET_free (ret);
1820     return NULL;
1821   }
1822   ret->neighbours =
1823     GNUNET_CONTAINER_multipeermap_create (STARTING_NEIGHBOURS_SIZE,
1824                                           GNUNET_YES);
1825   ret->ready_heap =
1826       GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1827   schedule_control_transmit (ret,
1828                              sizeof (struct StartMessage),
1829                              &send_start,
1830                              ret);
1831   return ret;
1832 }
1833
1834
1835 /**
1836  * Disconnect from the transport service.
1837  *
1838  * @param handle handle to the service as returned from #GNUNET_TRANSPORT_connect()
1839  */
1840 void
1841 GNUNET_TRANSPORT_disconnect (struct GNUNET_TRANSPORT_Handle *handle)
1842 {
1843   LOG (GNUNET_ERROR_TYPE_DEBUG,
1844        "Transport disconnect called!\n");
1845   /* this disconnects all neighbours... */
1846   if (handle->reconnect_task == NULL)
1847     disconnect_and_schedule_reconnect (handle);
1848   /* and now we stop trying to connect again... */
1849   if (handle->reconnect_task != NULL)
1850   {
1851     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
1852     handle->reconnect_task = NULL;
1853   }
1854   GNUNET_CONTAINER_multipeermap_destroy (handle->neighbours);
1855   handle->neighbours = NULL;
1856   if (handle->quota_task != NULL)
1857   {
1858     GNUNET_SCHEDULER_cancel (handle->quota_task);
1859     handle->quota_task = NULL;
1860   }
1861   GNUNET_free_non_null (handle->my_hello);
1862   handle->my_hello = NULL;
1863   GNUNET_assert (NULL == handle->tc_head);
1864   GNUNET_assert (NULL == handle->tc_tail);
1865   GNUNET_assert (NULL == handle->hwl_head);
1866   GNUNET_assert (NULL == handle->hwl_tail);
1867   GNUNET_CONTAINER_heap_destroy (handle->ready_heap);
1868   handle->ready_heap = NULL;
1869   GNUNET_free (handle);
1870 }
1871
1872
1873 /**
1874  * Check if we could queue a message of the given size for
1875  * transmission.  The transport service will take both its
1876  * internal buffers and bandwidth limits imposed by the
1877  * other peer into consideration when answering this query.
1878  *
1879  * @param handle connection to transport service
1880  * @param target who should receive the message
1881  * @param size how big is the message we want to transmit?
1882  * @param timeout after how long should we give up (and call
1883  *        notify with buf NULL and size 0)?
1884  * @param notify function to call when we are ready to
1885  *        send such a message
1886  * @param notify_cls closure for @a notify
1887  * @return NULL if someone else is already waiting to be notified
1888  *         non-NULL if the notify callback was queued (can be used to cancel
1889  *         using #GNUNET_TRANSPORT_notify_transmit_ready_cancel)
1890  */
1891 struct GNUNET_TRANSPORT_TransmitHandle *
1892 GNUNET_TRANSPORT_notify_transmit_ready (struct GNUNET_TRANSPORT_Handle *handle,
1893                                         const struct GNUNET_PeerIdentity *target,
1894                                         size_t size,
1895                                         struct GNUNET_TIME_Relative timeout,
1896                                         GNUNET_TRANSPORT_TransmitReadyNotify notify,
1897                                         void *notify_cls)
1898 {
1899   struct Neighbour *n;
1900   struct GNUNET_TRANSPORT_TransmitHandle *th;
1901   struct GNUNET_TIME_Relative delay;
1902
1903   n = neighbour_find (handle, target);
1904   if (NULL == n)
1905   {
1906     /* use GNUNET_TRANSPORT_try_connect first, only use this function
1907      * once a connection has been established */
1908     GNUNET_assert (0);
1909     return NULL;
1910   }
1911   if (NULL != n->th)
1912   {
1913     /* attempt to send two messages at the same time to the same peer */
1914     GNUNET_assert (0);
1915     return NULL;
1916   }
1917   GNUNET_assert (NULL == n->hn);
1918   th = GNUNET_new (struct GNUNET_TRANSPORT_TransmitHandle);
1919   th->neighbour = n;
1920   th->notify = notify;
1921   th->notify_cls = notify_cls;
1922   th->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1923   th->notify_size = size;
1924   n->th = th;
1925   /* calculate when our transmission should be ready */
1926   delay = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker,
1927                                               size + n->traffic_overhead);
1928   n->traffic_overhead = 0;
1929   if (delay.rel_value_us > timeout.rel_value_us)
1930     delay.rel_value_us = 0;        /* notify immediately (with failure) */
1931   LOG (GNUNET_ERROR_TYPE_DEBUG,
1932        "Bandwidth tracker allows next transmission to peer %s in %s\n",
1933        GNUNET_i2s (target),
1934        GNUNET_STRINGS_relative_time_to_string (delay,
1935                                                GNUNET_YES));
1936   n->hn = GNUNET_CONTAINER_heap_insert (handle->ready_heap,
1937                                         n,
1938                                         delay.rel_value_us);
1939   schedule_transmission (handle);
1940   return th;
1941 }
1942
1943
1944 /**
1945  * Cancel the specified transmission-ready notification.
1946  *
1947  * @param th handle returned from #GNUNET_TRANSPORT_notify_transmit_ready()
1948  */
1949 void
1950 GNUNET_TRANSPORT_notify_transmit_ready_cancel (struct GNUNET_TRANSPORT_TransmitHandle *th)
1951 {
1952   struct Neighbour *n;
1953
1954   GNUNET_assert (NULL == th->next);
1955   GNUNET_assert (NULL == th->prev);
1956   n = th->neighbour;
1957   GNUNET_assert (th == n->th);
1958   n->th = NULL;
1959   if (NULL != n->hn)
1960   {
1961     GNUNET_CONTAINER_heap_remove_node (n->hn);
1962     n->hn = NULL;
1963   }
1964   else
1965   {
1966     GNUNET_assert (NULL != th->timeout_task);
1967     GNUNET_SCHEDULER_cancel (th->timeout_task);
1968     th->timeout_task = NULL;
1969   }
1970   GNUNET_free (th);
1971 }
1972
1973
1974 /* end of transport_api.c */