src/transport/transport_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file transport/transport_api.c
  23  * @brief library to access the low-level P2P IO service
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - adjust testcases to use new 'try connect' style (should be easy, breaks API compatibility!)
  28  * - adjust core service to use new 'try connect' style (should be MUCH nicer there as well!)
  29  * - test test test
  30  */
  31 #include "platform.h"
  32 #include "gnunet_bandwidth_lib.h"
  33 #include "gnunet_client_lib.h"
  34 #include "gnunet_constants.h"
  35 #include "gnunet_container_lib.h"
  36 #include "gnunet_arm_service.h"
  37 #include "gnunet_hello_lib.h"
  38 #include "gnunet_protocols.h"
  39 #include "gnunet_server_lib.h"
  40 #include "gnunet_time_lib.h"
  41 #include "gnunet_transport_service.h"
  42 #include "transport.h"
  43
  44 /**
  45  * How large to start with for the hashmap of neighbours.
  46  */
  47 #define STARTING_NEIGHBOURS_SIZE 16
  48
  49
  50 /**
  51  * Handle for a message that should be transmitted to the service.
  52  * Used for both control messages and normal messages.
  53  */
  54 struct GNUNET_TRANSPORT_TransmitHandle
  55 {
  56
  57   /**
  58    * We keep all requests in a DLL.
  59    */
  60   struct GNUNET_TRANSPORT_TransmitHandle *next;
  61
  62   /**
  63    * We keep all requests in a DLL.
  64    */
  65   struct GNUNET_TRANSPORT_TransmitHandle *prev;
  66
  67   /**
  68    * Neighbour for this handle, NULL for control messages.
  69    */
  70   struct Neighbour *neighbour;
  71
  72   /**
  73    * Function to call when notify_size bytes are available
  74    * for transmission.
  75    */
  76   GNUNET_CONNECTION_TransmitReadyNotify notify;
  77
  78   /**
  79    * Closure for notify.
  80    */
  81   void *notify_cls;
  82
  83   /**
  84    * Timeout for this request, 0 for control messages.
  85    */
  86   struct GNUNET_TIME_Absolute timeout;
  87
  88   /**
  89    * Task to trigger request timeout if the request is stalled due to
  90    * congestion.
  91    */
  92   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
  93
  94   /**
  95    * How many bytes is our notify callback waiting for?
  96    */
  97   size_t notify_size;
  98
  99   /**
 100    * How important is this message? Not used for control messages.
 101    */
 102   uint32_t priority;
 103
 104 };
 105
 106
 107 /**
 108  * Entry in hash table of all of our current neighbours.
 109  */
 110 struct Neighbour
 111 {
 112   /**
 113    * Overall transport handle.
 114    */
 115   struct GNUNET_TRANSPORT_Handle *h;
 116
 117   /**
 118    * Active transmit handle or NULL.
 119    */
 120   struct GNUNET_TRANSPORT_TransmitHandle *th;
 121
 122   /**
 123    * Identity of this neighbour.
 124    */
 125   struct GNUNET_PeerIdentity id;
 126
 127   /**
 128    * Outbound bandwidh tracker.
 129    */
 130   struct GNUNET_BANDWIDTH_Tracker out_tracker;
 131
 132   /**
 133    * Entry in our readyness heap (which is sorted by 'next_ready'
 134    * value).  NULL if there is no pending transmission request for
 135    * this neighbour or if we're waiting for 'is_ready' to become
 136    * true AFTER the 'out_tracker' suggested that this peer's quota
 137    * has been satisfied (so once 'is_ready' goes to GNUNET_YES,
 138    * we should immediately go back into the heap).
 139    */
 140   struct GNUNET_CONTAINER_HeapNode *hn;
 141
 142   /**
 143    * Is this peer currently ready to receive a message?
 144    */
 145   int is_ready;
 146
 147 };
 148
 149
 150 /**
 151  * Linked list of functions to call whenever our HELLO is updated.
 152  */
 153 struct HelloWaitList
 154 {
 155
 156   /**
 157    * This is a doubly linked list.
 158    */
 159   struct HelloWaitList *next;
 160
 161   /**
 162    * This is a doubly linked list.
 163    */
 164   struct HelloWaitList *prev;
 165
 166   /**
 167    * Callback to call once we got our HELLO.
 168    */
 169   GNUNET_TRANSPORT_HelloUpdateCallback rec;
 170
 171   /**
 172    * Closure for rec.
 173    */
 174   void *rec_cls;
 175
 176 };
 177
 178
 179 /**
 180  * Handle for the transport service (includes all of the
 181  * state for the transport service).
 182  */
 183 struct GNUNET_TRANSPORT_Handle
 184 {
 185
 186   /**
 187    * Closure for the callbacks.
 188    */
 189   void *cls;
 190
 191   /**
 192    * Function to call for received data.
 193    */
 194   GNUNET_TRANSPORT_ReceiveCallback rec;
 195
 196   /**
 197    * function to call on connect events
 198    */
 199   GNUNET_TRANSPORT_NotifyConnect nc_cb;
 200
 201   /**
 202    * function to call on disconnect events
 203    */
 204   GNUNET_TRANSPORT_NotifyDisconnect nd_cb;
 205
 206   /**
 207    * Head of DLL of control messages.
 208    */
 209   struct GNUNET_TRANSPORT_TransmitHandle *control_head;
 210
 211   /**
 212    * Tail of DLL of control messages.
 213    */
 214   struct GNUNET_TRANSPORT_TransmitHandle *control_tail;
 215
 216   /**
 217    * The current HELLO message for this peer.  Updated
 218    * whenever transports change their addresses.
 219    */
 220   struct GNUNET_HELLO_Message *my_hello;
 221
 222   /**
 223    * My client connection to the transport service.
 224    */
 225   struct GNUNET_CLIENT_Connection *client;
 226
 227   /**
 228    * Handle to our registration with the client for notification.
 229    */
 230   struct GNUNET_CLIENT_TransmitHandle *cth;
 231
 232   /**
 233    * Linked list of pending requests for our HELLO.
 234    */
 235   struct HelloWaitList *hwl_head;
 236
 237   /**
 238    * Linked list of pending requests for our HELLO.
 239    */
 240   struct HelloWaitList *hwl_tail;
 241
 242   /**
 243    * My configuration.
 244    */
 245   const struct GNUNET_CONFIGURATION_Handle *cfg;
 246
 247   /**
 248    * Hash map of the current connected neighbours of this peer.
 249    * Maps peer identities to 'struct Neighbour' entries.
 250    */
 251   struct GNUNET_CONTAINER_MultiHashMap *neighbours;
 252
 253   /**
 254    * Heap sorting peers with pending messages by the timestamps that
 255    * specify when we could next send a message to the respective peer.
 256    * Excludes control messages (which can always go out immediately).
 257    * Maps time stamps to 'struct Neighbour' entries.
 258    */
 259   struct GNUNET_CONTAINER_Heap *ready_heap;
 260
 261   /**
 262    * Peer identity as assumed by this process, or all zeros.
 263    */
 264   struct GNUNET_PeerIdentity self;
 265
 266   /**
 267    * ID of the task trying to reconnect to the service.
 268    */
 269   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 270
 271   /**
 272    * ID of the task trying to trigger transmission for a peer while
 273    * maintaining bandwidth quotas.  In use if there are no control
 274    * messages and the smallest entry in the 'ready_heap' has a time
 275    * stamp in the future.
 276    */
 277   GNUNET_SCHEDULER_TaskIdentifier quota_task;
 278
 279   /**
 280    * Delay until we try to reconnect.
 281    */
 282   struct GNUNET_TIME_Relative reconnect_delay;
 283
 284   /**
 285    * Should we check that 'self' matches what the service thinks?
 286    * (if GNUNET_NO, then 'self' is all zeros!).
 287    */
 288   int check_self;
 289 };
 290
 291
 292 /**
 293  * Schedule the task to send one message, either from the control
 294  * list or the peer message queues  to the service.
 295  *
 296  * @param h transport service to schedule a transmission for
 297  */
 298 static void
 299 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h);
 300
 301
 302 /**
 303  * Function that will schedule the job that will try
 304  * to connect us again to the client.
 305  *
 306  * @param h transport service to reconnect
 307  */
 308 static void
 309 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h);
 310
 311
 312 /**
 313  * Get the neighbour list entry for the given peer
 314  *
 315  * @param h our context
 316  * @param peer peer to look up
 317  * @return NULL if no such peer entry exists
 318  */
 319 static struct Neighbour *
 320 neighbour_find (struct GNUNET_TRANSPORT_Handle *h,
 321                 const struct GNUNET_PeerIdentity *peer)
 322 {
 323   return GNUNET_CONTAINER_multihashmap_get(h->neighbours, &peer->hashPubKey);
 324 }
 325
 326
 327 /**
 328  * Add neighbour to our list
 329  *
 330  * @return NULL if this API is currently disconnecting from the service
 331  */
 332 static struct Neighbour *
 333 neighbour_add (struct GNUNET_TRANSPORT_Handle *h,
 334                const struct GNUNET_PeerIdentity *pid)
 335 {
 336   struct Neighbour *n;
 337
 338 #if DEBUG_TRANSPORT
 339   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 340               "Creating entry for neighbour `%4s'.\n",
 341               GNUNET_i2s (pid));
 342 #endif
 343   n = GNUNET_malloc (sizeof (struct Neighbour));
 344   n->id = *pid;
 345   n->h = h;
 346   n->is_ready = GNUNET_YES;
 347   GNUNET_BANDWIDTH_tracker_init (&n->out_tracker,
 348                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
 349                                  MAX_BANDWIDTH_CARRY_S);
 350   GNUNET_assert (GNUNET_OK ==
 351                  GNUNET_CONTAINER_multihashmap_put (h->neighbours,
 352                                                     &pid->hashPubKey,
 353                                                     n,
 354                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 355   return n;
 356 }
 357
 358
 359 /**
 360  * Iterator over hash map entries, for deleting state of a neighbour.
 361  *
 362  * @param cls the 'struct GNUNET_TRANSPORT_Handle*'
 363  * @param key peer identity
 364  * @param value value in the hash map, the neighbour entry to delete
 365  * @return GNUNET_YES if we should continue to
 366  *         iterate,
 367  *         GNUNET_NO if not.
 368  */
 369 static int
 370 neighbour_delete (void *cls,
 371                   const GNUNET_HashCode * key,
 372                   void *value)
 373 {
 374   struct GNUNET_TRANSPORT_Handle *handle = cls;
 375   struct Neighbour *n = value;
 376
 377   if (NULL != handle->nd_cb)
 378     handle->nd_cb (handle->cls,
 379                    &n->id);
 380   GNUNET_assert (NULL == n->th);
 381   GNUNET_assert (NULL == n->hn);
 382   GNUNET_assert (GNUNET_YES ==
 383                  GNUNET_CONTAINER_multihashmap_remove (handle->neighbours,
 384                                                        key,
 385                                                        n));
 386   GNUNET_free (n);
 387   return GNUNET_YES;
 388 }
 389
 390
 391 /**
 392  * Function we use for handling incoming messages.
 393  *
 394  * @param cls closure (struct GNUNET_TRANSPORT_Handle *)
 395  * @param msg message received, NULL on timeout or fatal error
 396  */
 397 static void
 398 demultiplexer (void *cls,
 399                const struct GNUNET_MessageHeader *msg)
 400 {
 401   struct GNUNET_TRANSPORT_Handle *h = cls;
 402   const struct DisconnectInfoMessage *dim;
 403   const struct ConnectInfoMessage *cim;
 404   const struct InboundMessage *im;
 405   const struct GNUNET_MessageHeader *imm;
 406   const struct SendOkMessage *okm;
 407   struct HelloWaitList *hwl;
 408   struct HelloWaitList *next_hwl;
 409   struct Neighbour *n;
 410   struct GNUNET_PeerIdentity me;
 411   uint16_t size;
 412   uint32_t ats_count;
 413
 414   GNUNET_assert (h->client != NULL);
 415   if (msg == NULL)
 416     {
 417 #if DEBUG_TRANSPORT
 418       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 419                   "Error receiving from transport service, disconnecting temporarily.\n");
 420 #endif
 421       disconnect_and_schedule_reconnect (h);
 422       return;
 423     }
 424   GNUNET_CLIENT_receive (h->client,
 425                          &demultiplexer, h,
 426                          GNUNET_TIME_UNIT_FOREVER_REL);
 427   size = ntohs (msg->size);
 428   switch (ntohs (msg->type))
 429     {
 430     case GNUNET_MESSAGE_TYPE_HELLO:
 431       if (GNUNET_OK !=
 432           GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) msg,
 433                                &me))
 434         {
 435           GNUNET_break (0);
 436           break;
 437         }
 438 #if DEBUG_TRANSPORT
 439       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 440                   "Receiving (my own) `%s' message, I am `%4s'.\n",
 441                   "HELLO", GNUNET_i2s (&me));
 442 #endif
 443       GNUNET_free_non_null (h->my_hello);
 444       h->my_hello = NULL;
 445       if (size < sizeof (struct GNUNET_MessageHeader))
 446         {
 447           GNUNET_break (0);
 448           break;
 449         }
 450       h->my_hello = GNUNET_malloc (size);
 451       memcpy (h->my_hello, msg, size);
 452       hwl = h->hwl_head;
 453       while (NULL != hwl)
 454         {
 455           next_hwl = hwl->next;
 456           hwl->rec (hwl->rec_cls,
 457                     (const struct GNUNET_MessageHeader *) h->my_hello);
 458           hwl = next_hwl;
 459         }
 460       break;
 461     case GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT:
 462       if (size < sizeof (struct ConnectInfoMessage))
 463         {
 464           GNUNET_break (0);
 465           break;
 466         }
 467       cim = (const struct ConnectInfoMessage *) msg;
 468       ats_count = ntohl (cim->ats_count);
 469       if (size != sizeof (struct ConnectInfoMessage) + ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information))
 470         {
 471           GNUNET_break (0);
 472           break;
 473         }
 474 #if DEBUG_TRANSPORT
 475       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 476                   "Receiving `%s' message for `%4s'.\n",
 477                   "CONNECT", GNUNET_i2s (&cim->id));
 478 #endif
 479       n = neighbour_find (h, &cim->id);
 480       if (n != NULL)
 481         {
 482           GNUNET_break (0);
 483           break;
 484         }
 485       n = neighbour_add (h, &cim->id);
 486       if (h->nc_cb != NULL)
 487         h->nc_cb (h->cls, &n->id,
 488                   &cim->ats, ats_count);
 489       break;
 490     case GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT:
 491       if (size != sizeof (struct DisconnectInfoMessage))
 492         {
 493           GNUNET_break (0);
 494           break;
 495         }
 496       dim = (const struct DisconnectInfoMessage *) msg;
 497       GNUNET_break (ntohl (dim->reserved) == 0);
 498 #if DEBUG_TRANSPORT_DISCONNECT
 499       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 500                   "Receiving `%s' message for `%4s'.\n",
 501                   "DISCONNECT",
 502                   GNUNET_i2s (&dim->peer));
 503 #endif
 504       n = neighbour_find (h, &dim->peer);
 505       if (n == NULL)
 506         {
 507           GNUNET_break (0);
 508           break;
 509         }
 510       neighbour_delete (h, &dim->peer.hashPubKey, n);
 511       break;
 512     case GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK:
 513       if (size != sizeof (struct SendOkMessage))
 514         {
 515           GNUNET_break (0);
 516           break;
 517         }
 518       okm = (const struct SendOkMessage *) msg;
 519 #if DEBUG_TRANSPORT
 520       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 521                   "Receiving `%s' message, transmission %s.\n", "SEND_OK",
 522                   ntohl (okm->success) == GNUNET_OK ? "succeeded" : "failed");
 523 #endif
 524       n = neighbour_find (h, &okm->peer);
 525       if (n == NULL)
 526         break;
 527       GNUNET_break (GNUNET_NO == n->is_ready);
 528       n->is_ready = GNUNET_YES;
 529       if ( (n->th != NULL) &&
 530            (n->hn == NULL) )
 531         {
 532           GNUNET_assert (GNUNET_SCHEDULER_NO_TASK != n->th->timeout_task);
 533           GNUNET_SCHEDULER_cancel (n->th->timeout_task);
 534           n->th->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 535           /* we've been waiting for this (congestion, not quota,
 536              caused delayed transmission) */
 537           n->hn = GNUNET_CONTAINER_heap_insert (h->ready_heap,
 538                                                 n, 0);
 539           schedule_transmission (h);
 540         }
 541       break;
 542     case GNUNET_MESSAGE_TYPE_TRANSPORT_RECV:
 543 #if DEBUG_TRANSPORT
 544       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 545                   "Receiving `%s' message.\n", "RECV");
 546 #endif
 547       if (size <
 548           sizeof (struct InboundMessage) +
 549           sizeof (struct GNUNET_MessageHeader))
 550         {
 551           GNUNET_break (0);
 552           break;
 553         }
 554       im = (const struct InboundMessage *) msg;
 555       GNUNET_break (0 == ntohl (im->reserved));
 556       ats_count = ntohl(im->ats_count);
 557       imm = (const struct GNUNET_MessageHeader *) &((&(im->ats))[ats_count+1]);
 558
 559       if (ntohs (imm->size) + sizeof (struct InboundMessage) + ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information) != size)
 560         {
 561           GNUNET_break (0);
 562           break;
 563         }
 564 #if DEBUG_TRANSPORT
 565       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 566                   "Received message of type %u from `%4s'.\n",
 567                   ntohs (imm->type), GNUNET_i2s (&im->peer));
 568 #endif
 569       n = neighbour_find (h, &im->peer);
 570       if (n == NULL)
 571         {
 572           GNUNET_break (0);
 573           break;
 574         }
 575       if (h->rec != NULL)
 576         h->rec (h->cls, &im->peer, imm,
 577                 &im->ats, ats_count);
 578       break;
 579     default:
 580       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 581                   _
 582                   ("Received unexpected message of type %u in %s:%u\n"),
 583                   ntohs (msg->type), __FILE__, __LINE__);
 584       GNUNET_break (0);
 585       break;
 586     }
 587 }
 588
 589
 590 /**
 591  * A transmission request could not be satisfied because of
 592  * network congestion.  Notify the initiator and clean up.
 593  *
 594  * @param cls the 'struct GNUNET_TRANSPORT_TransmitHandle'
 595  * @param tc scheduler context
 596  */
 597 static void
 598 timeout_request_due_to_congestion (void *cls,
 599                                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 600 {
 601   struct GNUNET_TRANSPORT_TransmitHandle *th = cls;
 602   struct Neighbour *n = th->neighbour;
 603
 604   n->th->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 605   GNUNET_assert (th == n->th);
 606   GNUNET_assert (NULL == n->hn);
 607   n->th = NULL;
 608   th->notify (th->notify_cls, 0, NULL);
 609   GNUNET_free (th);
 610 }
 611
 612
 613 /**
 614  * Transmit message(s) to service.
 615  *
 616  * @param cls handle to transport
 617  * @param size number of bytes available in buf
 618  * @param buf where to copy the message
 619  * @return number of bytes copied to buf
 620  */
 621 static size_t
 622 transport_notify_ready (void *cls, size_t size, void *buf)
 623 {
 624   struct GNUNET_TRANSPORT_Handle *h = cls;
 625   struct GNUNET_TRANSPORT_TransmitHandle *th;
 626   struct Neighbour *n;
 627   char *cbuf;
 628   struct OutboundMessage obm;
 629   size_t ret;
 630   size_t nret;
 631   size_t mret;
 632
 633   GNUNET_assert (NULL != h->client);
 634   h->cth = NULL;
 635   if (NULL == buf)
 636     {
 637       /* transmission failed */
 638       disconnect_and_schedule_reconnect (h);
 639       return 0;
 640     }
 641
 642   cbuf = buf;
 643   ret = 0;
 644   /* first send control messages */
 645   while ( (NULL != (th = h->control_head)) &&
 646           (th->notify_size <= size) )
 647     {
 648       GNUNET_CONTAINER_DLL_remove (h->control_head,
 649                                    h->control_tail,
 650                                    th);
 651       nret = th->notify (th->notify_cls, size, &cbuf[ret]);
 652 #if DEBUG_TRANSPORT
 653       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 654                   "Added %u bytes of control message at %u\n",
 655                   nret,
 656                   ret);
 657 #endif
 658       GNUNET_free (th);
 659       ret += nret;
 660       size -= nret;
 661     }
 662
 663   /* then, if possible and no control messages pending, send data messages */
 664   while ( (NULL == h->control_head) &&
 665           (NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap))) )
 666     {
 667        if (GNUNET_YES != n->is_ready)
 668         {
 669           /* peer not ready, wait for notification! */
 670           GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 671           n->hn = NULL;
 672           GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == n->th->timeout_task);
 673           n->th->timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (n->th->timeout),
 674                                                               &timeout_request_due_to_congestion,
 675                                                               n->th);
 676           continue;
 677         }
 678       th = n->th;
 679       if (th->notify_size + sizeof (struct OutboundMessage) > size)
 680         break; /* does not fit */
 681       if (GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker, th->notify_size).rel_value > 0)
 682         break; /* too early */
 683       GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 684       n->hn = NULL;
 685       n->th = NULL;
 686       n->is_ready = GNUNET_NO;
 687       GNUNET_assert (size >= sizeof (struct OutboundMessage));
 688       mret = th->notify (th->notify_cls,
 689                          size - sizeof (struct OutboundMessage),
 690                          &cbuf[ret + sizeof (struct OutboundMessage)]);
 691       GNUNET_assert (mret <= size - sizeof (struct OutboundMessage));
 692       if (mret != 0)
 693         {
 694           GNUNET_assert (mret + sizeof (struct OutboundMessage) < GNUNET_SERVER_MAX_MESSAGE_SIZE);
 695           obm.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SEND);
 696           obm.header.size = htons (mret + sizeof (struct OutboundMessage));
 697           obm.priority = htonl (th->priority);
 698           obm.timeout = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (th->timeout));
 699           obm.peer = n->id;
 700           memcpy (&cbuf[ret], &obm, sizeof (struct OutboundMessage));
 701           ret += (mret + sizeof (struct OutboundMessage));
 702           size -= (mret + sizeof (struct OutboundMessage));
 703           GNUNET_BANDWIDTH_tracker_consume (&n->out_tracker, mret);
 704         }
 705       GNUNET_free (th);
 706     }
 707   /* if there are more pending messages, try to schedule those */
 708   schedule_transmission (h);
 709 #if DEBUG_TRANSPORT
 710   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 711               "Transmitting %u bytes to transport service\n", ret);
 712 #endif
 713   return ret;
 714 }
 715
 716
 717 /**
 718  * Schedule the task to send one message, either from the control
 719  * list or the peer message queues  to the service.
 720  *
 721  * @param cls transport service to schedule a transmission for
 722  * @param tc scheduler context
 723  */
 724 static void
 725 schedule_transmission_task (void *cls,
 726                             const struct GNUNET_SCHEDULER_TaskContext *tc)
 727 {
 728   struct GNUNET_TRANSPORT_Handle *h = cls;
 729   size_t size;
 730   struct GNUNET_TRANSPORT_TransmitHandle *th;
 731   struct Neighbour *n;
 732
 733   h->quota_task = GNUNET_SCHEDULER_NO_TASK;
 734   GNUNET_assert (NULL != h->client);
 735   /* destroy all requests that have timed out */
 736   while ( (NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap))) &&
 737           (GNUNET_TIME_absolute_get_remaining (n->th->timeout).rel_value == 0) )
 738     {
 739       /* notify client that the request could not be satisfied within
 740          the given time constraints */
 741       th = n->th;
 742       n->th = NULL;
 743       GNUNET_assert (n == GNUNET_CONTAINER_heap_remove_root (h->ready_heap));
 744       n->hn = NULL;
 745 #if DEBUG_TRANSPORT
 746       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 747                   "Signalling timeout for transmission to peer %s due to congestion\n",
 748                   GNUNET_i2s (&n->id));
 749 #endif
 750       GNUNET_assert (0 ==
 751                      th->notify (th->notify_cls, 0, NULL));
 752       GNUNET_free (th);
 753     }
 754   if (NULL != h->cth)
 755     return;
 756   if (NULL != h->control_head)
 757     {
 758       size = h->control_head->notify_size;
 759     }
 760   else
 761     {
 762       n = GNUNET_CONTAINER_heap_peek (h->ready_heap);
 763       if (NULL == n)
 764         return; /* no pending messages */
 765       size = n->th->notify_size + sizeof (struct OutboundMessage);
 766     }
 767 #if DEBUG_TRANSPORT
 768   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 769               "Calling notify_transmit_ready\n");
 770 #endif
 771   h->cth =
 772     GNUNET_CLIENT_notify_transmit_ready (h->client,
 773                                          size,
 774                                          GNUNET_TIME_UNIT_FOREVER_REL,
 775                                          GNUNET_NO,
 776                                          &transport_notify_ready,
 777                                          h);
 778   GNUNET_assert (NULL != h->cth);
 779 }
 780
 781
 782 /**
 783  * Schedule the task to send one message, either from the control
 784  * list or the peer message queues  to the service.
 785  *
 786  * @param h transport service to schedule a transmission for
 787  */
 788 static void
 789 schedule_transmission (struct GNUNET_TRANSPORT_Handle *h)
 790 {
 791   struct GNUNET_TIME_Relative delay;
 792   struct Neighbour *n;
 793
 794   GNUNET_assert (NULL != h->client);
 795   if (h->quota_task != GNUNET_SCHEDULER_NO_TASK)
 796     {
 797       GNUNET_SCHEDULER_cancel (h->quota_task);
 798       h->quota_task = GNUNET_SCHEDULER_NO_TASK;
 799     }
 800   if (NULL != h->control_head)
 801     delay = GNUNET_TIME_UNIT_ZERO;
 802   else if (NULL != (n = GNUNET_CONTAINER_heap_peek (h->ready_heap)))
 803     delay = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker, n->th->notify_size);
 804   else
 805     return; /* no work to be done */
 806   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 807               "Scheduling next transmission to service in %llu ms\n",
 808               (unsigned long long) delay.rel_value);
 809   h->quota_task = GNUNET_SCHEDULER_add_delayed (delay,
 810                                                 &schedule_transmission_task,
 811                                                 h);
 812 }
 813
 814
 815 /**
 816  * Queue control request for transmission to the transport
 817  * service.
 818  *
 819  * @param h handle to the transport service
 820  * @param size number of bytes to be transmitted
 821  * @param notify function to call to get the content
 822  * @param notify_cls closure for notify
 823  */
 824 static void
 825 schedule_control_transmit (struct GNUNET_TRANSPORT_Handle *h,
 826                            size_t size,
 827                            GNUNET_CONNECTION_TransmitReadyNotify notify,
 828                            void *notify_cls)
 829 {
 830   struct GNUNET_TRANSPORT_TransmitHandle *th;
 831
 832 #if DEBUG_TRANSPORT
 833   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 834               "Control transmit of %u bytes requested\n",
 835               size);
 836 #endif
 837   th = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_TransmitHandle));
 838   th->notify = notify;
 839   th->notify_cls = notify_cls;
 840   th->notify_size = size;
 841   GNUNET_CONTAINER_DLL_insert_tail (h->control_head,
 842                                     h->control_tail,
 843                                     th);
 844   schedule_transmission (h);
 845 }
 846
 847
 848 /**
 849  * Transmit START message to service.
 850  *
 851  * @param cls unused
 852  * @param size number of bytes available in buf
 853  * @param buf where to copy the message
 854  * @return number of bytes copied to buf
 855  */
 856 static size_t
 857 send_start (void *cls, size_t size, void *buf)
 858 {
 859   struct GNUNET_TRANSPORT_Handle *h = cls;
 860   struct StartMessage s;
 861
 862   if (buf == NULL)
 863     {
 864       /* Can only be shutdown, just give up */
 865 #if DEBUG_TRANSPORT
 866       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 867                   "Shutdown while trying to transmit `%s' request.\n",
 868                   "START");
 869 #endif
 870       return 0;
 871     }
 872 #if DEBUG_TRANSPORT
 873   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 874               "Transmitting `%s' request.\n", "START");
 875 #endif
 876   GNUNET_assert (size >= sizeof (struct StartMessage));
 877   s.header.size = htons (sizeof (struct StartMessage));
 878   s.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_START);
 879   s.do_check = htonl (h->check_self);
 880   s.self = h->self;
 881   memcpy (buf, &s, sizeof (struct StartMessage));
 882   GNUNET_CLIENT_receive (h->client,
 883                          &demultiplexer, h, GNUNET_TIME_UNIT_FOREVER_REL);
 884   return sizeof (struct StartMessage);
 885 }
 886
 887
 888 /**
 889  * Try again to connect to transport service.
 890  *
 891  * @param cls the handle to the transport service
 892  * @param tc scheduler context
 893  */
 894 static void
 895 reconnect (void *cls,
 896            const struct GNUNET_SCHEDULER_TaskContext *tc)
 897 {
 898   struct GNUNET_TRANSPORT_Handle *h = cls;
 899
 900   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 901   if ( (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 902     {
 903       /* shutdown, just give up */
 904       return;
 905     }
 906 #if DEBUG_TRANSPORT
 907   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 908               "Connecting to transport service.\n");
 909 #endif
 910   GNUNET_assert (h->client == NULL);
 911   GNUNET_assert (h->control_head == NULL);
 912   GNUNET_assert (h->control_tail == NULL);
 913   h->client = GNUNET_CLIENT_connect ("transport", h->cfg);
 914   GNUNET_assert (h->client != NULL);
 915   schedule_control_transmit (h,
 916                              sizeof (struct StartMessage),
 917                              &send_start, h);
 918 }
 919
 920
 921 /**
 922  * Function that will schedule the job that will try
 923  * to connect us again to the client.
 924  *
 925  * @param h transport service to reconnect
 926  */
 927 static void
 928 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_Handle *h)
 929 {
 930   struct GNUNET_TRANSPORT_TransmitHandle *th;
 931
 932   GNUNET_assert (h->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
 933   /* Forget about all neighbours that we used to be connected to */
 934   GNUNET_CONTAINER_multihashmap_iterate(h->neighbours,
 935                                         &neighbour_delete,
 936                                         h);
 937   if (NULL != h->cth)
 938     {
 939       GNUNET_CLIENT_notify_transmit_ready_cancel (h->cth);
 940       h->cth = NULL;
 941     }
 942   if (NULL != h->client)
 943     {
 944       GNUNET_CLIENT_disconnect (h->client, GNUNET_YES);
 945       h->client = NULL;
 946     }
 947   if (h->quota_task != GNUNET_SCHEDULER_NO_TASK)
 948     {
 949       GNUNET_SCHEDULER_cancel (h->quota_task);
 950       h->quota_task = GNUNET_SCHEDULER_NO_TASK;
 951     }
 952   while ( (NULL != (th = h->control_head)))
 953     {
 954       GNUNET_CONTAINER_DLL_remove (h->control_head,
 955                                    h->control_tail,
 956                                    th);
 957       th->notify (th->notify_cls, 0, NULL);
 958       GNUNET_free (th);
 959     }
 960 #if DEBUG_TRANSPORT
 961   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 962               "Scheduling task to reconnect to transport service in %llu ms.\n",
 963               h->reconnect_delay.rel_value);
 964 #endif
 965   h->reconnect_task
 966     = GNUNET_SCHEDULER_add_delayed (h->reconnect_delay,
 967                                     &reconnect, h);
 968   if (h->reconnect_delay.rel_value == 0)
 969     {
 970       h->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
 971     }
 972   else
 973     {
 974       h->reconnect_delay = GNUNET_TIME_relative_multiply (h->reconnect_delay, 2);
 975       h->reconnect_delay = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
 976                                                      h->reconnect_delay);
 977     }
 978 }
 979
 980
 981 /**
 982  * Closure for 'send_set_quota'.
 983  */
 984 struct SetQuotaContext
 985 {
 986
 987   /**
 988    * Identity of the peer impacted by the quota change.
 989    */
 990   struct GNUNET_PeerIdentity target;
 991
 992   /**
 993    * Quota to transmit.
 994    */
 995   struct GNUNET_BANDWIDTH_Value32NBO quota_in;
 996 };
 997
 998
 999 /**
1000  * Send SET_QUOTA message to the service.
1001  *
1002  * @param cls the 'struct SetQuotaContext'
1003  * @param size number of bytes available in buf
1004  * @param buf where to copy the message
1005  * @return number of bytes copied to buf
1006  */
1007 static size_t
1008 send_set_quota (void *cls, size_t size, void *buf)
1009 {
1010   struct SetQuotaContext *sqc = cls;
1011   struct QuotaSetMessage msg;
1012
1013   if (buf == NULL)
1014     {
1015       GNUNET_free (sqc);
1016       return 0;
1017     }
1018 #if DEBUG_TRANSPORT
1019   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1020               "Transmitting `%s' request with respect to `%4s'.\n",
1021               "SET_QUOTA",
1022               GNUNET_i2s (&sqc->target));
1023 #endif
1024   GNUNET_assert (size >= sizeof (struct QuotaSetMessage));
1025   msg.header.size = htons (sizeof (struct QuotaSetMessage));
1026   msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_SET_QUOTA);
1027   msg.quota = sqc->quota_in;
1028   msg.peer = sqc->target;
1029   memcpy (buf, &msg, sizeof (msg));
1030   GNUNET_free (sqc);
1031   return sizeof (struct QuotaSetMessage);
1032 }
1033
1034
1035 /**
1036  * Set the share of incoming bandwidth for the given
1037  * peer to the specified amount.
1038  *
1039  * @param handle connection to transport service
1040  * @param target who's bandwidth quota is being changed
1041  * @param quota_in incoming bandwidth quota in bytes per ms
1042  * @param quota_out outgoing bandwidth quota in bytes per ms
1043  */
1044 void
1045 GNUNET_TRANSPORT_set_quota (struct GNUNET_TRANSPORT_Handle *handle,
1046                             const struct GNUNET_PeerIdentity *target,
1047                             struct GNUNET_BANDWIDTH_Value32NBO quota_in,
1048                             struct GNUNET_BANDWIDTH_Value32NBO quota_out)
1049 {
1050   struct Neighbour *n;
1051   struct SetQuotaContext *sqc;
1052
1053   n = neighbour_find (handle, target);
1054   if (NULL == n)
1055     {
1056       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1057                   "Quota changed to %u for peer `%s', but I have no such neighbour!\n",
1058                   (unsigned int) ntohl (quota_out.value__),
1059                   GNUNET_i2s (target));
1060       return;
1061     }
1062   GNUNET_assert (NULL != handle->client);
1063 #if DEBUG_TRANSPORT
1064   if (ntohl (quota_out.value__) != n->out_tracker.available_bytes_per_s__)
1065     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1066                 "Quota changed from %u to %u for peer `%s'\n",
1067                 (unsigned int) n->out_tracker.available_bytes_per_s__,
1068                 (unsigned int) ntohl (quota_out.value__),
1069                 GNUNET_i2s (target));
1070   else
1071     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1072                 "Quota remains at %u for peer `%s'\n",
1073                 (unsigned int) n->out_tracker.available_bytes_per_s__,
1074                 GNUNET_i2s (target));
1075 #endif
1076   GNUNET_BANDWIDTH_tracker_update_quota (&n->out_tracker,
1077                                          quota_out);
1078   sqc = GNUNET_malloc (sizeof (struct SetQuotaContext));
1079   sqc->target = *target;
1080   sqc->quota_in = quota_in;
1081   schedule_control_transmit (handle,
1082                              sizeof (struct QuotaSetMessage),
1083                              &send_set_quota, sqc);
1084 }
1085
1086
1087 /**
1088  * Send REQUEST_CONNECT message to the service.
1089  *
1090  * @param cls the 'struct GNUNET_PeerIdentity'
1091  * @param size number of bytes available in buf
1092  * @param buf where to copy the message
1093  * @return number of bytes copied to buf
1094  */
1095 static size_t
1096 send_try_connect (void *cls, size_t size, void *buf)
1097 {
1098   struct GNUNET_PeerIdentity *pid = cls;
1099   struct TransportRequestConnectMessage msg;
1100
1101   if (buf == NULL)
1102     {
1103       GNUNET_free (pid);
1104       return 0;
1105     }
1106 #if DEBUG_TRANSPORT
1107   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1108               "Transmitting `%s' request with respect to `%4s'.\n",
1109               "REQUEST_CONNECT",
1110               GNUNET_i2s (pid));
1111 #endif
1112   GNUNET_assert (size >= sizeof (struct TransportRequestConnectMessage));
1113   msg.header.size = htons (sizeof (struct TransportRequestConnectMessage));
1114   msg.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_CONNECT);
1115   msg.reserved = htonl (0);
1116   msg.peer = *pid;
1117   memcpy (buf, &msg, sizeof (msg));
1118   GNUNET_free (pid);
1119   return sizeof (struct TransportRequestConnectMessage);
1120 }
1121
1122
1123 /**
1124  * Ask the transport service to establish a connection to
1125  * the given peer.
1126  *
1127  * @param handle connection to transport service
1128  * @param target who we should try to connect to
1129  */
1130 void
1131 GNUNET_TRANSPORT_try_connect (struct GNUNET_TRANSPORT_Handle *handle,
1132                               const struct GNUNET_PeerIdentity *target)
1133 {
1134   struct GNUNET_PeerIdentity *pid;
1135
1136   if (NULL == handle->client)
1137     return;
1138   pid = GNUNET_malloc (sizeof (struct GNUNET_PeerIdentity));
1139   *pid = *target;
1140   schedule_control_transmit (handle,
1141                              sizeof (struct TransportRequestConnectMessage),
1142                              &send_try_connect, pid);
1143 }
1144
1145
1146 /**
1147  * Send HELLO message to the service.
1148  *
1149  * @param cls the HELLO message to send
1150  * @param size number of bytes available in buf
1151  * @param buf where to copy the message
1152  * @return number of bytes copied to buf
1153  */
1154 static size_t
1155 send_hello (void *cls, size_t size, void *buf)
1156 {
1157   struct GNUNET_MessageHeader *msg = cls;
1158   uint16_t ssize;
1159
1160   if (buf == NULL)
1161     {
1162 #if DEBUG_TRANSPORT_TIMEOUT
1163       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1164                   "Timeout while trying to transmit `%s' request.\n",
1165                   "HELLO");
1166 #endif
1167       GNUNET_free (msg);
1168       return 0;
1169     }
1170 #if DEBUG_TRANSPORT
1171   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1172               "Transmitting `%s' request.\n", "HELLO");
1173 #endif
1174   ssize = ntohs (msg->size);
1175   GNUNET_assert (size >= ssize);
1176   memcpy (buf, msg, ssize);
1177   GNUNET_free (msg);
1178   return ssize;
1179 }
1180
1181
1182 /**
1183  * Offer the transport service the HELLO of another peer.  Note that
1184  * the transport service may just ignore this message if the HELLO is
1185  * malformed or useless due to our local configuration.
1186  *
1187  * @param handle connection to transport service
1188  * @param hello the hello message
1189  * @param cont continuation to call when HELLO has been sent
1190  * @param cls closure for continuation
1191  *
1192  */
1193 void
1194 GNUNET_TRANSPORT_offer_hello (struct GNUNET_TRANSPORT_Handle *handle,
1195                               const struct GNUNET_MessageHeader *hello,
1196                               GNUNET_SCHEDULER_Task cont,
1197                               void *cls)
1198 {
1199   uint16_t size;
1200   struct GNUNET_PeerIdentity peer;
1201   struct GNUNET_MessageHeader *msg;
1202
1203   if (NULL == handle->client)
1204     return;
1205   GNUNET_break (ntohs (hello->type) == GNUNET_MESSAGE_TYPE_HELLO);
1206   size = ntohs (hello->size);
1207   GNUNET_break (size >= sizeof (struct GNUNET_MessageHeader));
1208   if (GNUNET_OK != GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message*) hello,
1209                                         &peer))
1210     {
1211       GNUNET_break (0);
1212       return;
1213     }
1214   msg = GNUNET_malloc(size);
1215   memcpy (msg, hello, size);
1216 #if DEBUG_TRANSPORT
1217   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1218               "Offering `%s' message of `%4s' to transport for validation.\n",
1219               "HELLO",
1220               GNUNET_i2s (&peer));
1221 #endif
1222   schedule_control_transmit (handle,
1223                              size,
1224                              &send_hello, msg);
1225 }
1226
1227
1228 /**
1229  * Obtain the HELLO message for this peer.
1230  *
1231  * @param handle connection to transport service
1232  * @param rec function to call with the HELLO, sender will be our peer
1233  *            identity; message and sender will be NULL on timeout
1234  *            (handshake with transport service pending/failed).
1235  *             cost estimate will be 0.
1236  * @param rec_cls closure for rec
1237  */
1238 void
1239 GNUNET_TRANSPORT_get_hello (struct GNUNET_TRANSPORT_Handle *handle,
1240                             GNUNET_TRANSPORT_HelloUpdateCallback rec,
1241                             void *rec_cls)
1242 {
1243   struct HelloWaitList *hwl;
1244
1245   hwl = GNUNET_malloc (sizeof (struct HelloWaitList));
1246   hwl->rec = rec;
1247   hwl->rec_cls = rec_cls;
1248   GNUNET_CONTAINER_DLL_insert (handle->hwl_head,
1249                                handle->hwl_tail,
1250                                hwl);
1251   if (handle->my_hello == NULL)
1252     return;
1253   rec (rec_cls, (const struct GNUNET_MessageHeader *) handle->my_hello);
1254 }
1255
1256
1257 /**
1258  * Stop receiving updates about changes to our HELLO message.
1259  *
1260  * @param handle connection to transport service
1261  * @param rec function previously registered to be called with the HELLOs
1262  * @param rec_cls closure for rec
1263  */
1264 void
1265 GNUNET_TRANSPORT_get_hello_cancel (struct GNUNET_TRANSPORT_Handle *handle,
1266                                    GNUNET_TRANSPORT_HelloUpdateCallback rec,
1267                                    void *rec_cls)
1268 {
1269   struct HelloWaitList *pos;
1270
1271   pos = handle->hwl_head;
1272   while (pos != NULL)
1273     {
1274       if ( (pos->rec == rec) &&
1275            (pos->rec_cls == rec_cls) )
1276         break;
1277       pos = pos->next;
1278     }
1279   GNUNET_break (pos != NULL);
1280   if (pos == NULL)
1281     return;
1282   GNUNET_CONTAINER_DLL_remove (handle->hwl_head,
1283                                handle->hwl_tail,
1284                                pos);
1285   GNUNET_free (pos);
1286 }
1287
1288
1289 /**
1290  * Connect to the transport service.  Note that the connection may
1291  * complete (or fail) asynchronously.
1292  *
1293  * @param cfg configuration to use
1294  * @param self our own identity (API should check that it matches
1295  *             the identity found by transport), or NULL (no check)
1296  * @param cls closure for the callbacks
1297  * @param rec receive function to call
1298  * @param nc function to call on connect events
1299  * @param nd function to call on disconnect events
1300  */
1301 struct GNUNET_TRANSPORT_Handle *
1302 GNUNET_TRANSPORT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
1303                           const struct GNUNET_PeerIdentity *self,
1304                           void *cls,
1305                           GNUNET_TRANSPORT_ReceiveCallback rec,
1306                           GNUNET_TRANSPORT_NotifyConnect nc,
1307                           GNUNET_TRANSPORT_NotifyDisconnect nd)
1308 {
1309   struct GNUNET_TRANSPORT_Handle *ret;
1310
1311   ret = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_Handle));
1312   if (self != NULL)
1313     {
1314       ret->self = *self;
1315       ret->check_self = GNUNET_YES;
1316     }
1317   ret->cfg = cfg;
1318   ret->cls = cls;
1319   ret->rec = rec;
1320   ret->nc_cb = nc;
1321   ret->nd_cb = nd;
1322   ret->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
1323   ret->neighbours = GNUNET_CONTAINER_multihashmap_create(STARTING_NEIGHBOURS_SIZE);
1324   ret->ready_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1325   ret->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, ret);
1326   return ret;
1327 }
1328
1329
1330 /**
1331  * Disconnect from the transport service.
1332  *
1333  * @param handle handle to the service as returned from GNUNET_TRANSPORT_connect
1334  */
1335 void
1336 GNUNET_TRANSPORT_disconnect (struct GNUNET_TRANSPORT_Handle *handle)
1337 {
1338 #if DEBUG_TRANSPORT
1339   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1340               "Transport disconnect called!\n");
1341 #endif
1342   /* this disconnects all neighbours... */
1343   if (handle->reconnect_task == GNUNET_SCHEDULER_NO_TASK)
1344     disconnect_and_schedule_reconnect (handle);
1345   /* and now we stop trying to connect again... */
1346   if (handle->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
1347     {
1348       GNUNET_SCHEDULER_cancel (handle->reconnect_task);
1349       handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1350     }
1351   GNUNET_CONTAINER_multihashmap_destroy (handle->neighbours);
1352   handle->neighbours = NULL;
1353   if (handle->quota_task != GNUNET_SCHEDULER_NO_TASK)
1354     {
1355       GNUNET_SCHEDULER_cancel (handle->quota_task);
1356       handle->quota_task = GNUNET_SCHEDULER_NO_TASK;
1357     }
1358   GNUNET_free_non_null (handle->my_hello);
1359   handle->my_hello = NULL;
1360   GNUNET_assert (handle->hwl_head == NULL);
1361   GNUNET_assert (handle->hwl_tail == NULL);
1362   GNUNET_CONTAINER_heap_destroy (handle->ready_heap);
1363   handle->ready_heap = NULL;
1364   GNUNET_free (handle);
1365 }
1366
1367
1368 /**
1369  * Check if we could queue a message of the given size for
1370  * transmission.  The transport service will take both its
1371  * internal buffers and bandwidth limits imposed by the
1372  * other peer into consideration when answering this query.
1373  *
1374  * @param handle connection to transport service
1375  * @param target who should receive the message
1376  * @param size how big is the message we want to transmit?
1377  * @param priority how important is the message?
1378  * @param timeout after how long should we give up (and call
1379  *        notify with buf NULL and size 0)?
1380  * @param notify function to call when we are ready to
1381  *        send such a message
1382  * @param notify_cls closure for notify
1383  * @return NULL if someone else is already waiting to be notified
1384  *         non-NULL if the notify callback was queued (can be used to cancel
1385  *         using GNUNET_TRANSPORT_notify_transmit_ready_cancel)
1386  */
1387 struct GNUNET_TRANSPORT_TransmitHandle *
1388 GNUNET_TRANSPORT_notify_transmit_ready (struct GNUNET_TRANSPORT_Handle *handle,
1389                                         const struct GNUNET_PeerIdentity *target,
1390                                         size_t size,
1391                                         uint32_t priority,
1392                                         struct GNUNET_TIME_Relative timeout,
1393                                         GNUNET_CONNECTION_TransmitReadyNotify notify,
1394                                         void *notify_cls)
1395 {
1396   struct Neighbour *n;
1397   struct GNUNET_TRANSPORT_TransmitHandle *th;
1398   struct GNUNET_TIME_Relative delay;
1399
1400   n = neighbour_find (handle, target);
1401   if (NULL == n)
1402     {
1403       /* use GNUNET_TRANSPORT_try_connect first, only use this function
1404          once a connection has been established */
1405       GNUNET_assert (0);
1406       return NULL;
1407     }
1408   if (NULL != n->th)
1409     {
1410       /* attempt to send two messages at the same time to the same peer */
1411       GNUNET_assert (0);
1412       return NULL;
1413     }
1414   GNUNET_assert (NULL == n->hn);
1415   th = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_TransmitHandle));
1416   th->neighbour = n;
1417   th->notify = notify;
1418   th->notify_cls = notify_cls;
1419   th->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1420   th->notify_size = size;
1421   th->priority = priority;
1422   n->th = th;
1423   /* calculate when our transmission should be ready */
1424   delay = GNUNET_BANDWIDTH_tracker_get_delay (&n->out_tracker, size);
1425   if (delay.rel_value > timeout.rel_value)
1426     delay.rel_value = 0; /* notify immediately (with failure) */
1427 #if DEBUG_TRANSPORT
1428   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1429               "Bandwidth tracker allows next transmission to peer %s in %llu ms\n",
1430               GNUNET_i2s (target),
1431               (unsigned long long) delay.rel_value);
1432 #endif
1433   n->hn = GNUNET_CONTAINER_heap_insert (handle->ready_heap,
1434                                         n,
1435                                         delay.rel_value);
1436   schedule_transmission (handle);
1437   return th;
1438 }
1439
1440
1441 /**
1442  * Cancel the specified transmission-ready notification.
1443  *
1444  * @param th handle returned from GNUNET_TRANSPORT_notify_transmit_ready
1445  */
1446 void
1447 GNUNET_TRANSPORT_notify_transmit_ready_cancel (struct GNUNET_TRANSPORT_TransmitHandle *th)
1448 {
1449   struct Neighbour *n;
1450
1451   GNUNET_assert (NULL == th->next);
1452   GNUNET_assert (NULL == th->prev);
1453   n = th->neighbour;
1454   GNUNET_assert (th == n->th);
1455   n->th = NULL;
1456   if (n->hn != NULL)
1457     {
1458       GNUNET_CONTAINER_heap_remove_node (n->hn);
1459       n->hn = NULL;
1460     }
1461   else
1462     {
1463       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK != th->timeout_task);
1464       GNUNET_SCHEDULER_cancel (th->timeout_task);
1465       th->timeout_task = GNUNET_SCHEDULER_NO_TASK;
1466     }
1467   GNUNET_free (th);
1468 }
1469
1470
1471 /* end of transport_api.c */