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