src/transport/transport_api2_core.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2013, 2016, 2018 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19  */
  20
  21 /**
  22  * @file transport/transport_api_core.c
  23  * @brief library to access the transport service for message exchange
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_constants.h"
  29 #include "gnunet_arm_service.h"
  30 #include "gnunet_hello_lib.h"
  31 #include "gnunet_protocols.h"
  32 #include "gnunet_transport_core_service.h"
  33 #include "transport.h"
  34
  35 #define LOG(kind, ...) GNUNET_log_from (kind, "transport-api-core", __VA_ARGS__)
  36
  37 /**
  38  * How large to start with for the hashmap of neighbours.
  39  */
  40 #define STARTING_NEIGHBOURS_SIZE 16
  41
  42 /**
  43  * Window size. How many messages to the same target do we pass
  44  * to TRANSPORT without a SEND_OK in between? Small values limit
  45  * thoughput, large values will increase latency.
  46  *
  47  * FIXME-OPTIMIZE: find out what good values are experimentally,
  48  * maybe set adaptively (i.e. to observed available bandwidth).
  49  */
  50 #define SEND_WINDOW_SIZE 4
  51
  52
  53 /**
  54  * Entry in hash table of all of our current (connected) neighbours.
  55  */
  56 struct Neighbour
  57 {
  58   /**
  59    * Identity of this neighbour.
  60    */
  61   struct GNUNET_PeerIdentity id;
  62
  63   /**
  64    * Overall transport handle.
  65    */
  66   struct GNUNET_TRANSPORT_CoreHandle *h;
  67
  68   /**
  69    * Active message queue for the peer.
  70    */
  71   struct GNUNET_MQ_Handle *mq;
  72
  73   /**
  74    * Envelope with the message we are currently transmitting (or NULL).
  75    */
  76   struct GNUNET_MQ_Envelope *env;
  77
  78   /**
  79    * Closure for @e mq handlers.
  80    */
  81   void *handlers_cls;
  82
  83   /**
  84    * How many messages can we still send to this peer before we should
  85    * throttle?
  86    */
  87   unsigned int ready_window;
  88
  89   /**
  90    * Used to indicate our status if @e env is non-NULL.  Set to
  91    * #GNUNET_YES if we did pass a message to the MQ and are waiting
  92    * for the call to #notify_send_done(). Set to #GNUNET_NO if the @e
  93    * ready_window is 0 and @e env is waiting for a
  94    * #GNUNET_MESSAGE_TYPE_TRANSPORT_RECV_OK?
  95    */
  96   int16_t awaiting_done;
  97
  98   /**
  99    * Size of the message in @e env.
 100    */
 101   uint16_t env_size;
 102 };
 103
 104
 105 /**
 106  * Handle for the transport service (includes all of the
 107  * state for the transport service).
 108  */
 109 struct GNUNET_TRANSPORT_CoreHandle
 110 {
 111   /**
 112    * Closure for the callbacks.
 113    */
 114   void *cls;
 115
 116   /**
 117    * Functions to call for received data (template for
 118    * new message queues).
 119    */
 120   struct GNUNET_MQ_MessageHandler *handlers;
 121
 122   /**
 123    * function to call on connect events
 124    */
 125   GNUNET_TRANSPORT_NotifyConnect nc_cb;
 126
 127   /**
 128    * function to call on disconnect events
 129    */
 130   GNUNET_TRANSPORT_NotifyDisconnect nd_cb;
 131
 132   /**
 133    * My client connection to the transport service.
 134    */
 135   struct GNUNET_MQ_Handle *mq;
 136
 137   /**
 138    * My configuration.
 139    */
 140   const struct GNUNET_CONFIGURATION_Handle *cfg;
 141
 142   /**
 143    * Hash map of the current connected neighbours of this peer.
 144    * Maps peer identities to `struct Neighbour` entries.
 145    */
 146   struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 147
 148   /**
 149    * Peer identity as assumed by this process, or all zeros.
 150    */
 151   struct GNUNET_PeerIdentity self;
 152
 153   /**
 154    * ID of the task trying to reconnect to the service.
 155    */
 156   struct GNUNET_SCHEDULER_Task *reconnect_task;
 157
 158   /**
 159    * Delay until we try to reconnect.
 160    */
 161   struct GNUNET_TIME_Relative reconnect_delay;
 162
 163   /**
 164    * Should we check that @e self matches what the service thinks?
 165    * (if #GNUNET_NO, then @e self is all zeros!).
 166    */
 167   int check_self;
 168 };
 169
 170
 171 /**
 172  * Function that will schedule the job that will try
 173  * to connect us again to the client.
 174  *
 175  * @param h transport service to reconnect
 176  */
 177 static void
 178 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_CoreHandle *h);
 179
 180
 181 /**
 182  * Get the neighbour list entry for the given peer
 183  *
 184  * @param h our context
 185  * @param peer peer to look up
 186  * @return NULL if no such peer entry exists
 187  */
 188 static struct Neighbour *
 189 neighbour_find (struct GNUNET_TRANSPORT_CoreHandle *h,
 190                 const struct GNUNET_PeerIdentity *peer)
 191 {
 192   return GNUNET_CONTAINER_multipeermap_get (h->neighbours, peer);
 193 }
 194
 195
 196 /**
 197  * Iterator over hash map entries, for deleting state of a neighbour.
 198  *
 199  * @param cls the `struct GNUNET_TRANSPORT_CoreHandle *`
 200  * @param key peer identity
 201  * @param value value in the hash map, the neighbour entry to delete
 202  * @return #GNUNET_YES if we should continue to
 203  *         iterate,
 204  *         #GNUNET_NO if not.
 205  */
 206 static int
 207 neighbour_delete (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
 208 {
 209   struct GNUNET_TRANSPORT_CoreHandle *handle = cls;
 210   struct Neighbour *n = value;
 211
 212   LOG (GNUNET_ERROR_TYPE_DEBUG,
 213        "Dropping entry for neighbour `%s'.\n",
 214        GNUNET_i2s (key));
 215   if (NULL != handle->nd_cb)
 216     handle->nd_cb (handle->cls, &n->id, n->handlers_cls);
 217   if (NULL != n->env)
 218   {
 219     GNUNET_MQ_send_cancel (n->env);
 220     n->env = NULL;
 221   }
 222   GNUNET_MQ_destroy (n->mq);
 223   GNUNET_assert (NULL == n->mq);
 224   GNUNET_assert (
 225     GNUNET_YES ==
 226     GNUNET_CONTAINER_multipeermap_remove (handle->neighbours, key, n));
 227   GNUNET_free (n);
 228   return GNUNET_YES;
 229 }
 230
 231
 232 /**
 233  * Generic error handler, called with the appropriate
 234  * error code and the same closure specified at the creation of
 235  * the message queue.
 236  * Not every message queue implementation supports an error handler.
 237  *
 238  * @param cls closure with the `struct GNUNET_TRANSPORT_CoreHandle *`
 239  * @param error error code
 240  */
 241 static void
 242 mq_error_handler (void *cls, enum GNUNET_MQ_Error error)
 243 {
 244   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 245
 246   LOG (GNUNET_ERROR_TYPE_DEBUG,
 247        "Error receiving from transport service, disconnecting temporarily.\n");
 248   disconnect_and_schedule_reconnect (h);
 249 }
 250
 251
 252 /**
 253  * A message from the handler's message queue to a neighbour was
 254  * transmitted.  Now trigger (possibly delayed) notification of the
 255  * neighbour's message queue that we are done and thus ready for
 256  * the next message.  Note that the MQ being ready is independent
 257  * of the send window, as we may queue many messages and simply
 258  * not pass them to TRANSPORT if the send window is insufficient.
 259  *
 260  * @param cls the `struct Neighbour` where the message was sent
 261  */
 262 static void
 263 notify_send_done (void *cls)
 264 {
 265   struct Neighbour *n = cls;
 266
 267   n->awaiting_done = GNUNET_NO;
 268   n->env = NULL;
 269   GNUNET_MQ_impl_send_continue (n->mq);
 270 }
 271
 272
 273 /**
 274  * We have an envelope waiting for transmission at @a n, and
 275  * our transmission window is positive. Perform the transmission.
 276  *
 277  * @param n neighbour to perform transmission for
 278  */
 279 static void
 280 do_send (struct Neighbour *n)
 281 {
 282   GNUNET_assert (0 < n->ready_window);
 283   GNUNET_assert (NULL != n->env);
 284   n->ready_window--;
 285   n->awaiting_done = GNUNET_YES;
 286   GNUNET_MQ_notify_sent (n->env, &notify_send_done, n);
 287   GNUNET_MQ_send (n->h->mq, n->env);
 288   LOG (GNUNET_ERROR_TYPE_DEBUG,
 289        "Passed message of type %u for neighbour `%s' to TRANSPORT.\n",
 290        ntohs (GNUNET_MQ_env_get_msg (n->env)->type),
 291        GNUNET_i2s (&n->id));
 292 }
 293
 294
 295 /**
 296  * Implement sending functionality of a message queue.
 297  * Called one message at a time. Should send the @a msg
 298  * to the transport service and then notify the queue
 299  * once we are ready for the next one.
 300  *
 301  * @param mq the message queue
 302  * @param msg the message to send
 303  * @param impl_state state of the implementation
 304  */
 305 static void
 306 mq_send_impl (struct GNUNET_MQ_Handle *mq,
 307               const struct GNUNET_MessageHeader *msg,
 308               void *impl_state)
 309 {
 310   struct Neighbour *n = impl_state;
 311   struct OutboundMessage *obm;
 312   uint16_t msize;
 313
 314   msize = ntohs (msg->size);
 315   if (msize >= GNUNET_MAX_MESSAGE_SIZE - sizeof(*obm))
 316   {
 317     GNUNET_break (0);
 318     GNUNET_MQ_impl_send_continue (mq);
 319     return;
 320   }
 321   LOG (GNUNET_ERROR_TYPE_DEBUG,
 322        "CORE requested transmission of message of type %u to neighbour `%s'.\n",
 323        ntohs (msg->type),
 324        GNUNET_i2s (&n->id));
 325
 326   GNUNET_assert (NULL == n->env);
 327   n->env =
 328     GNUNET_MQ_msg_nested_mh (obm, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND, msg);
 329   n->env_size = ntohs (msg->size);
 330   {
 331     struct GNUNET_MQ_Envelope *env;
 332
 333     env = GNUNET_MQ_get_current_envelope (mq);
 334     obm->priority = htonl ((uint32_t) GNUNET_MQ_env_get_options (env));
 335   }
 336   obm->peer = n->id;
 337   if (0 == n->ready_window)
 338   {
 339     LOG (GNUNET_ERROR_TYPE_DEBUG,
 340          "Flow control delays transmission to CORE until we see SEND_OK.\n");
 341     return;   /* can't send yet, need to wait for SEND_OK */
 342   }
 343   do_send (n);
 344 }
 345
 346
 347 /**
 348  * Handle destruction of a message queue.  Implementations must not
 349  * free @a mq, but should take care of @a impl_state.
 350  *
 351  * @param mq the message queue to destroy
 352  * @param impl_state state of the implementation
 353  */
 354 static void
 355 mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
 356 {
 357   struct Neighbour *n = impl_state;
 358
 359   GNUNET_assert (mq == n->mq);
 360   n->mq = NULL;
 361 }
 362
 363
 364 /**
 365  * Implementation function that cancels the currently sent message.
 366  * Should basically undo whatever #mq_send_impl() did.
 367  *
 368  * @param mq message queue
 369  * @param impl_state state specific to the implementation
 370  */
 371 static void
 372 mq_cancel_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
 373 {
 374   struct Neighbour *n = impl_state;
 375
 376   n->ready_window++;
 377   if (GNUNET_YES == n->awaiting_done)
 378   {
 379     GNUNET_MQ_send_cancel (n->env);
 380     n->env = NULL;
 381     n->awaiting_done = GNUNET_NO;
 382   }
 383   else
 384   {
 385     GNUNET_assert (0 == n->ready_window);
 386     n->env = NULL;
 387   }
 388 }
 389
 390
 391 /**
 392  * We had an error processing a message we forwarded from a peer to
 393  * the CORE service.  We should just complain about it but otherwise
 394  * continue processing.
 395  *
 396  * @param cls closure
 397  * @param error error code
 398  */
 399 static void
 400 peer_mq_error_handler (void *cls, enum GNUNET_MQ_Error error)
 401 {
 402   /* struct Neighbour *n = cls; */
 403
 404   GNUNET_break_op (0);
 405 }
 406
 407
 408 /**
 409  * Function we use for handling incoming connect messages.
 410  *
 411  * @param cls closure, a `struct GNUNET_TRANSPORT_Handle *`
 412  * @param cim message received
 413  */
 414 static void
 415 handle_connect (void *cls, const struct ConnectInfoMessage *cim)
 416 {
 417   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 418   struct Neighbour *n;
 419
 420   LOG (GNUNET_ERROR_TYPE_DEBUG,
 421        "Receiving CONNECT message for `%s'\n",
 422        GNUNET_i2s (&cim->id));
 423   n = neighbour_find (h, &cim->id);
 424   if (NULL != n)
 425   {
 426     GNUNET_break (0);
 427     disconnect_and_schedule_reconnect (h);
 428     return;
 429   }
 430   n = GNUNET_new (struct Neighbour);
 431   n->id = cim->id;
 432   n->h = h;
 433   n->ready_window = SEND_WINDOW_SIZE;
 434   GNUNET_assert (GNUNET_OK ==
 435                  GNUNET_CONTAINER_multipeermap_put (
 436                    h->neighbours,
 437                    &n->id,
 438                    n,
 439                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 440
 441   n->mq = GNUNET_MQ_queue_for_callbacks (&mq_send_impl,
 442                                          &mq_destroy_impl,
 443                                          &mq_cancel_impl,
 444                                          n,
 445                                          h->handlers,
 446                                          &peer_mq_error_handler,
 447                                          n);
 448   if (NULL != h->nc_cb)
 449   {
 450     n->handlers_cls = h->nc_cb (h->cls, &n->id, n->mq);
 451     GNUNET_MQ_set_handlers_closure (n->mq, n->handlers_cls);
 452   }
 453 }
 454
 455
 456 /**
 457  * Function we use for handling incoming disconnect messages.
 458  *
 459  * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *`
 460  * @param dim message received
 461  */
 462 static void
 463 handle_disconnect (void *cls, const struct DisconnectInfoMessage *dim)
 464 {
 465   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 466   struct Neighbour *n;
 467
 468   GNUNET_break (ntohl (dim->reserved) == 0);
 469   LOG (GNUNET_ERROR_TYPE_DEBUG,
 470        "Receiving DISCONNECT message for `%s'.\n",
 471        GNUNET_i2s (&dim->peer));
 472   n = neighbour_find (h, &dim->peer);
 473   if (NULL == n)
 474   {
 475     GNUNET_break (0);
 476     disconnect_and_schedule_reconnect (h);
 477     return;
 478   }
 479   GNUNET_assert (GNUNET_YES == neighbour_delete (h, &dim->peer, n));
 480 }
 481
 482
 483 /**
 484  * Function we use for handling incoming send-ok messages.
 485  *
 486  * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *`
 487  * @param okm message received
 488  */
 489 static void
 490 handle_send_ok (void *cls, const struct SendOkMessage *okm)
 491 {
 492   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 493   struct Neighbour *n;
 494
 495   LOG (GNUNET_ERROR_TYPE_DEBUG,
 496        "Receiving SEND_OK message for transmission to %s\n",
 497        GNUNET_i2s (&okm->peer));
 498   n = neighbour_find (h, &okm->peer);
 499   if (NULL == n)
 500   {
 501     /* We should never get a 'SEND_OK' for a peer that we are not
 502        connected to */
 503     GNUNET_break (0);
 504     disconnect_and_schedule_reconnect (h);
 505     return;
 506   }
 507   n->ready_window++;
 508   if ((NULL != n->env) && (1 == n->ready_window))
 509     do_send (n);
 510 }
 511
 512
 513 /**
 514  * Function we use for checking incoming "inbound" messages.
 515  *
 516  * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *`
 517  * @param im message received
 518  */
 519 static int
 520 check_recv (void *cls, const struct InboundMessage *im)
 521 {
 522   const struct GNUNET_MessageHeader *imm;
 523   uint16_t size;
 524
 525   size = ntohs (im->header.size) - sizeof(*im);
 526   if (size < sizeof(struct GNUNET_MessageHeader))
 527   {
 528     GNUNET_break (0);
 529     return GNUNET_SYSERR;
 530   }
 531   imm = (const struct GNUNET_MessageHeader *) &im[1];
 532   if (ntohs (imm->size) != size)
 533   {
 534     GNUNET_break (0);
 535     return GNUNET_SYSERR;
 536   }
 537   return GNUNET_OK;
 538 }
 539
 540
 541 /**
 542  * Function we use for handling incoming messages.
 543  *
 544  * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *`
 545  * @param im message received
 546  */
 547 static void
 548 handle_recv (void *cls, const struct InboundMessage *im)
 549 {
 550   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 551   const struct GNUNET_MessageHeader *imm =
 552     (const struct GNUNET_MessageHeader *) &im[1];
 553   struct Neighbour *n;
 554
 555   LOG (GNUNET_ERROR_TYPE_DEBUG,
 556        "Received message of type %u with %u bytes from `%s'.\n",
 557        (unsigned int) ntohs (imm->type),
 558        (unsigned int) ntohs (imm->size),
 559        GNUNET_i2s (&im->peer));
 560   n = neighbour_find (h, &im->peer);
 561   if (NULL == n)
 562   {
 563     GNUNET_break (0);
 564     disconnect_and_schedule_reconnect (h);
 565     return;
 566   }
 567   GNUNET_MQ_inject_message (n->mq, imm);
 568 }
 569
 570
 571 /**
 572  * Try again to connect to transport service.
 573  *
 574  * @param cls the handle to the transport service
 575  */
 576 static void
 577 reconnect (void *cls)
 578 {
 579   struct GNUNET_TRANSPORT_CoreHandle *h = cls;
 580   struct GNUNET_MQ_MessageHandler handlers[] =
 581   { GNUNET_MQ_hd_fixed_size (connect,
 582                              GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT,
 583                              struct ConnectInfoMessage,
 584                              h),
 585     GNUNET_MQ_hd_fixed_size (disconnect,
 586                              GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT,
 587                              struct DisconnectInfoMessage,
 588                              h),
 589     GNUNET_MQ_hd_fixed_size (send_ok,
 590                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK,
 591                              struct SendOkMessage,
 592                              h),
 593     GNUNET_MQ_hd_var_size (recv,
 594                            GNUNET_MESSAGE_TYPE_TRANSPORT_RECV,
 595                            struct InboundMessage,
 596                            h),
 597     GNUNET_MQ_handler_end () };
 598   struct GNUNET_MQ_Envelope *env;
 599   struct StartMessage *s;
 600   uint32_t options;
 601
 602   h->reconnect_task = NULL;
 603   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connecting to transport service.\n");
 604   GNUNET_assert (NULL == h->mq);
 605   h->mq =
 606     GNUNET_CLIENT_connect (h->cfg, "transport", handlers, &mq_error_handler, h);
 607   if (NULL == h->mq)
 608     return;
 609   env = GNUNET_MQ_msg (s, GNUNET_MESSAGE_TYPE_TRANSPORT_START);
 610   options = 0;
 611   if (h->check_self)
 612     options |= 1;
 613   if (NULL != h->handlers)
 614     options |= 2;
 615   s->options = htonl (options);
 616   s->self = h->self;
 617   GNUNET_MQ_send (h->mq, env);
 618 }
 619
 620
 621 /**
 622  * Disconnect from the transport service.
 623  *
 624  * @param h transport service to reconnect
 625  */
 626 static void
 627 disconnect (struct GNUNET_TRANSPORT_CoreHandle *h)
 628 {
 629   GNUNET_CONTAINER_multipeermap_iterate (h->neighbours, &neighbour_delete, h);
 630   if (NULL != h->mq)
 631   {
 632     GNUNET_MQ_destroy (h->mq);
 633     h->mq = NULL;
 634   }
 635 }
 636
 637
 638 /**
 639  * Function that will schedule the job that will try
 640  * to connect us again to the client.
 641  *
 642  * @param h transport service to reconnect
 643  */
 644 static void
 645 disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_CoreHandle *h)
 646 {
 647   GNUNET_assert (NULL == h->reconnect_task);
 648   disconnect (h);
 649   LOG (GNUNET_ERROR_TYPE_DEBUG,
 650        "Scheduling task to reconnect to transport service in %s.\n",
 651        GNUNET_STRINGS_relative_time_to_string (h->reconnect_delay, GNUNET_YES));
 652   h->reconnect_task =
 653     GNUNET_SCHEDULER_add_delayed (h->reconnect_delay, &reconnect, h);
 654   h->reconnect_delay = GNUNET_TIME_STD_BACKOFF (h->reconnect_delay);
 655 }
 656
 657
 658 /**
 659  * Checks if a given peer is connected to us and get the message queue.
 660  *
 661  * @param handle connection to transport service
 662  * @param peer the peer to check
 663  * @return NULL if disconnected, otherwise message queue for @a peer
 664  */
 665 struct GNUNET_MQ_Handle *
 666 GNUNET_TRANSPORT_core_get_mq (struct GNUNET_TRANSPORT_CoreHandle *handle,
 667                               const struct GNUNET_PeerIdentity *peer)
 668 {
 669   struct Neighbour *n;
 670
 671   n = neighbour_find (handle, peer);
 672   if (NULL == n)
 673     return NULL;
 674   return n->mq;
 675 }
 676
 677
 678 /**
 679  * Notification from the CORE service to the TRANSPORT service
 680  * that the CORE service has finished processing a message from
 681  * TRANSPORT (via the @code{handlers} of #GNUNET_TRANSPORT_core_connect())
 682  * and that it is thus now OK for TRANSPORT to send more messages
 683  * for @a pid.
 684  *
 685  * Used to provide flow control, this is our equivalent to
 686  * #GNUNET_SERVICE_client_continue() of an ordinary service.
 687  *
 688  * Note that due to the use of a window, TRANSPORT may send multiple
 689  * messages destined for the same peer even without an intermediate
 690  * call to this function. However, CORE must still call this function
 691  * once per message received, as otherwise eventually the window will
 692  * be full and TRANSPORT will stop providing messages to CORE for @a
 693  * pid.
 694  *
 695  * @param ch core handle
 696  * @param pid which peer was the message from that was fully processed by CORE
 697  */
 698 void
 699 GNUNET_TRANSPORT_core_receive_continue (struct GNUNET_TRANSPORT_CoreHandle *ch,
 700                                         const struct GNUNET_PeerIdentity *pid)
 701 {
 702   struct GNUNET_MQ_Envelope *env;
 703   struct RecvOkMessage *rok;
 704
 705   LOG (GNUNET_ERROR_TYPE_DEBUG,
 706        "Message for %s finished CORE processing, sending RECV_OK.\n",
 707        GNUNET_i2s (pid));
 708   if (NULL == ch->mq)
 709     return;
 710   env = GNUNET_MQ_msg (rok, GNUNET_MESSAGE_TYPE_TRANSPORT_RECV_OK);
 711   rok->increase_window_delta = htonl (1);
 712   rok->peer = *pid;
 713   GNUNET_MQ_send (ch->mq, env);
 714 }
 715
 716
 717 /**
 718  * Connect to the transport service.  Note that the connection may
 719  * complete (or fail) asynchronously.
 720  *
 721  * @param cfg configuration to use
 722  * @param self our own identity (API should check that it matches
 723  *             the identity found by transport), or NULL (no check)
 724  * @param cls closure for the callbacks
 725  * @param rec receive function to call
 726  * @param nc function to call on connect events
 727  * @param nd function to call on disconnect events
 728  * @return NULL on error
 729  */
 730 struct GNUNET_TRANSPORT_CoreHandle *
 731 GNUNET_TRANSPORT_core_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 732                                const struct GNUNET_PeerIdentity *self,
 733                                const struct GNUNET_MQ_MessageHandler *handlers,
 734                                void *cls,
 735                                GNUNET_TRANSPORT_NotifyConnect nc,
 736                                GNUNET_TRANSPORT_NotifyDisconnect nd)
 737 {
 738   struct GNUNET_TRANSPORT_CoreHandle *h;
 739   unsigned int i;
 740
 741   h = GNUNET_new (struct GNUNET_TRANSPORT_CoreHandle);
 742   if (NULL != self)
 743   {
 744     h->self = *self;
 745     h->check_self = GNUNET_YES;
 746   }
 747   h->cfg = cfg;
 748   h->cls = cls;
 749   h->nc_cb = nc;
 750   h->nd_cb = nd;
 751   h->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
 752   if (NULL != handlers)
 753   {
 754     for (i = 0; NULL != handlers[i].cb; i++)
 755       ;
 756     h->handlers = GNUNET_new_array (i + 1, struct GNUNET_MQ_MessageHandler);
 757     GNUNET_memcpy (h->handlers,
 758                    handlers,
 759                    i * sizeof(struct GNUNET_MQ_MessageHandler));
 760   }
 761   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connecting to transport service\n");
 762   reconnect (h);
 763   if (NULL == h->mq)
 764   {
 765     GNUNET_free_non_null (h->handlers);
 766     GNUNET_free (h);
 767     return NULL;
 768   }
 769   h->neighbours =
 770     GNUNET_CONTAINER_multipeermap_create (STARTING_NEIGHBOURS_SIZE, GNUNET_YES);
 771   return h;
 772 }
 773
 774
 775 /**
 776  * Disconnect from the transport service.
 777  *
 778  * @param handle handle to the service as returned from
 779  * #GNUNET_TRANSPORT_core_connect()
 780  */
 781 void
 782 GNUNET_TRANSPORT_core_disconnect (struct GNUNET_TRANSPORT_CoreHandle *handle)
 783 {
 784   LOG (GNUNET_ERROR_TYPE_DEBUG, "Transport disconnect called!\n");
 785   /* this disconnects all neighbours... */
 786   disconnect (handle);
 787   /* and now we stop trying to connect again... */
 788   if (NULL != handle->reconnect_task)
 789   {
 790     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 791     handle->reconnect_task = NULL;
 792   }
 793   GNUNET_CONTAINER_multipeermap_destroy (handle->neighbours);
 794   handle->neighbours = NULL;
 795   GNUNET_free_non_null (handle->handlers);
 796   handle->handlers = NULL;
 797   GNUNET_free (handle);
 798 }
 799
 800
 801 /* end of transport_api_core.c */