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