src/transport/gnunet-service-tng.c

   1 /*
   2  This file is part of GNUnet.
   3  Copyright (C) 2010-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 /**
  19  * @file transport/gnunet-service-transport.c
  20  * @brief main for gnunet-service-transport
  21  * @author Christian Grothoff
  22  */
  23 #include "platform.h"
  24 #include "gnunet_util_lib.h"
  25 #include "gnunet_statistics_service.h"
  26 #include "gnunet_transport_service.h"
  27 #include "gnunet_peerinfo_service.h"
  28 #include "gnunet_ats_service.h"
  29 #include "gnunet-service-transport.h"
  30 #include "transport.h"
  31
  32
  33 /**
  34  * How many messages can we have pending for a given client process
  35  * before we start to drop incoming messages?  We typically should
  36  * have only one client and so this would be the primary buffer for
  37  * messages, so the number should be chosen rather generously.
  38  *
  39  * The expectation here is that most of the time the queue is large
  40  * enough so that a drop is virtually never required.  Note that
  41  * this value must be about as large as 'TOTAL_MSGS' in the
  42  * 'test_transport_api_reliability.c', otherwise that testcase may
  43  * fail.
  44  */
  45 #define MAX_PENDING (128 * 1024)
  46
  47
  48 /**
  49  * What type of client is the `struct TransportClient` about?
  50  */
  51 enum ClientType
  52 {
  53   /**
  54    * We do not know yet (client is fresh).
  55    */
  56   CT_NONE = 0,
  57
  58   /**
  59    * Is the CORE service, we need to forward traffic to it.
  60    */
  61   CT_CORE = 1,
  62
  63   /**
  64    * It is a monitor, forward monitor data.
  65    */
  66   CT_MONITOR = 2,
  67
  68   /**
  69    * It is a communicator, use for communication.
  70    */
  71   CT_COMMUNICATOR = 3
  72 };
  73
  74
  75 /**
  76  * Client connected to the transport service.
  77  */
  78 struct TransportClient;
  79
  80
  81 /**
  82  * A neighbour that at least one communicator is connected to.
  83  */
  84 struct Neighbour;
  85
  86
  87 /**
  88  * List of available queues for a particular neighbour.
  89  */
  90 struct Queue
  91 {
  92   /**
  93    * Kept in a MDLL.
  94    */
  95   struct Queue *next_neighbour;
  96
  97   /**
  98    * Kept in a MDLL.
  99    */
 100   struct Queue *prev_neighbour;
 101
 102   /**
 103    * Kept in a MDLL.
 104    */
 105   struct Queue *prev_client;
 106
 107   /**
 108    * Kept in a MDLL.
 109    */
 110   struct Queue *next_client;
 111
 112   /**
 113    * Which neighbour is this queue for?
 114    */
 115   struct Neighbour *neighbour;
 116
 117   /**
 118    * Which communicator offers this queue?
 119    */
 120   struct TransportClient *tc;
 121
 122   /**
 123    * Unique identifier of this queue with the communicator.
 124    */
 125   uint32_t qid;
 126
 127   /**
 128    * Network type offered by this queue.
 129    */
 130   enum GNUNET_ATS_Network_Type nt;
 131
 132   /**
 133    * Address served by the queue.
 134    */
 135   const char *address;
 136 };
 137
 138
 139 /**
 140  * A neighbour that at least one communicator is connected to.
 141  */
 142 struct Neighbour
 143 {
 144
 145   /**
 146    * Which peer is this about?
 147    */
 148   struct GNUNET_PeerIdentity pid;
 149
 150   /**
 151    * Head of list of messages pending for this neighbour.
 152    */
 153   struct PendingMessage *pending_msg_head;
 154
 155   /**
 156    * Tail of list of messages pending for this neighbour.
 157    */
 158   struct PendingMessage *pending_msg_tail;
 159
 160   /**
 161    * Head of DLL of queues to this peer.
 162    */
 163   struct Queue *queue_head;
 164
 165   /**
 166    * Tail of DLL of queues to this peer.
 167    */
 168   struct Queue *queue_tail;
 169
 170 };
 171
 172
 173 /**
 174  * Transmission request from CORE that is awaiting delivery.
 175  */
 176 struct PendingMessage
 177 {
 178   /**
 179    * Kept in a MDLL of messages for this @a target.
 180    */
 181   struct PendingMessage *next_neighbour;
 182
 183   /**
 184    * Kept in a MDLL of messages for this @a target.
 185    */
 186   struct PendingMessage *prev_neighbour;
 187
 188   /**
 189    * Kept in a MDLL of messages from this @a client.
 190    */
 191   struct PendingMessage *next_client;
 192
 193   /**
 194    * Kept in a MDLL of messages from this @a client.
 195    */
 196   struct PendingMessage *prev_client;
 197
 198   /**
 199    * Target of the request.
 200    */
 201   struct Neighbour *target;
 202
 203   /**
 204    * Client that issued the transmission request.
 205    */
 206   struct TransportClient *client;
 207
 208   /**
 209    * Size of the original message.
 210    */
 211   uint32_t bytes_msg;
 212
 213 };
 214
 215
 216 /**
 217  * Client connected to the transport service.
 218  */
 219 struct TransportClient
 220 {
 221
 222   /**
 223    * Kept in a DLL.
 224    */
 225   struct TransportClient *next;
 226
 227   /**
 228    * Kept in a DLL.
 229    */
 230   struct TransportClient *prev;
 231
 232   /**
 233    * Handle to the client.
 234    */
 235   struct GNUNET_SERVICE_Client *client;
 236
 237   /**
 238    * Message queue to the client.
 239    */
 240   struct GNUNET_MQ_Handle *mq;
 241
 242   /**
 243    * What type of client is this?
 244    */
 245   enum ClientType type;
 246
 247   union
 248   {
 249
 250     /**
 251      * Information for @e type #CT_CORE.
 252      */
 253     struct {
 254
 255       /**
 256        * Head of list of messages pending for this client.
 257        */
 258       struct PendingMessage *pending_msg_head;
 259
 260       /**
 261        * Tail of list of messages pending for this client.
 262        */
 263       struct PendingMessage *pending_msg_tail;
 264
 265     } core;
 266
 267     /**
 268      * Information for @e type #CT_MONITOR.
 269      */
 270     struct {
 271
 272       /**
 273        * Peer identity to monitor the addresses of.
 274        * Zero to monitor all neighbours.  Valid if
 275        * @e type is #CT_MONITOR.
 276        */
 277       struct GNUNET_PeerIdentity peer;
 278
 279       /**
 280        * Is this a one-shot monitor?
 281        */
 282       int one_shot;
 283
 284     } monitor;
 285
 286
 287     /**
 288      * Information for @e type #CT_COMMUNICATOR.
 289      */
 290     struct {
 291       /**
 292        * If @e type is #CT_COMMUNICATOR, this communicator
 293        * supports communicating using these addresses.
 294        */
 295       char *address_prefix;
 296
 297       /**
 298        * Head of DLL of queues offered by this communicator.
 299        */
 300       struct Queue *queue_head;
 301
 302       /**
 303        * Tail of DLL of queues offered by this communicator.
 304        */
 305       struct Queue *queue_tail;
 306
 307     } communicator;
 308
 309   } details;
 310
 311 };
 312
 313
 314 /**
 315  * Head of linked list of all clients to this service.
 316  */
 317 static struct TransportClient *clients_head;
 318
 319 /**
 320  * Tail of linked list of all clients to this service.
 321  */
 322 static struct TransportClient *clients_tail;
 323
 324 /**
 325  * Statistics handle.
 326  */
 327 struct GNUNET_STATISTICS_Handle *GST_stats;
 328
 329 /**
 330  * Configuration handle.
 331  */
 332 const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
 333
 334 /**
 335  * Our public key.
 336  */
 337 struct GNUNET_PeerIdentity GST_my_identity;
 338
 339 /**
 340  * Our private key.
 341  */
 342 struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
 343
 344 /**
 345  * Map from PIDs to `struct Neighbour` entries.  A peer is
 346  * a neighbour if we have an MQ to it from some communicator.
 347  */
 348 static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 349
 350
 351 /**
 352  * Lookup neighbour record for peer @a pid.
 353  *
 354  * @param pid neighbour to look for
 355  * @return NULL if we do not have this peer as a neighbour
 356  */
 357 static struct Neighbour *
 358 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
 359 {
 360   return GNUNET_CONTAINER_multipeermap_get (neighbours,
 361                                             pid);
 362 }
 363
 364
 365 /**
 366  * Called whenever a client connects.  Allocates our
 367  * data structures associated with that client.
 368  *
 369  * @param cls closure, NULL
 370  * @param client identification of the client
 371  * @param mq message queue for the client
 372  * @return our `struct TransportClient`
 373  */
 374 static void *
 375 client_connect_cb (void *cls,
 376                    struct GNUNET_SERVICE_Client *client,
 377                    struct GNUNET_MQ_Handle *mq)
 378 {
 379   struct TransportClient *tc;
 380
 381   tc = GNUNET_new (struct TransportClient);
 382   tc->client = client;
 383   tc->mq = mq;
 384   GNUNET_CONTAINER_DLL_insert (clients_head,
 385                                clients_tail,
 386                                tc);
 387   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 388               "Client %p connected\n",
 389               tc);
 390   return tc;
 391 }
 392
 393
 394 /**
 395  * Called whenever a client is disconnected.  Frees our
 396  * resources associated with that client.
 397  *
 398  * @param cls closure, NULL
 399  * @param client identification of the client
 400  * @param app_ctx our `struct TransportClient`
 401  */
 402 static void
 403 client_disconnect_cb (void *cls,
 404                       struct GNUNET_SERVICE_Client *client,
 405                       void *app_ctx)
 406 {
 407   struct TransportClient *tc = app_ctx;
 408
 409   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 410               "Client %p disconnected, cleaning up.\n",
 411               tc);
 412   GNUNET_CONTAINER_DLL_remove (clients_head,
 413                                clients_tail,
 414                                tc);
 415   switch (tc->type)
 416   {
 417   case CT_NONE:
 418     break;
 419   case CT_CORE:
 420     {
 421       struct PendingMessage *pm;
 422
 423       while (NULL != (pm = tc->details.core.pending_msg_head))
 424       {
 425         GNUNET_CONTAINER_MDLL_remove (client,
 426                                       tc->details.core.pending_msg_head,
 427                                       tc->details.core.pending_msg_tail,
 428                                       pm);
 429         pm->client = NULL;
 430       }
 431     }
 432     break;
 433   case CT_MONITOR:
 434     break;
 435   case CT_COMMUNICATOR:
 436     GNUNET_free (tc->details.communicator.address_prefix);
 437     break;
 438   }
 439   GNUNET_free (tc);
 440 }
 441
 442
 443 /**
 444  * Initialize a "CORE" client.  We got a start message from this
 445  * client, so add it to the list of clients for broadcasting of
 446  * inbound messages.
 447  *
 448  * @param cls the client
 449  * @param start the start message that was sent
 450  */
 451 static void
 452 handle_client_start (void *cls,
 453                      const struct StartMessage *start)
 454 {
 455   struct TransportClient *tc = cls;
 456   uint32_t options;
 457
 458   options = ntohl (start->options);
 459   if ( (0 != (1 & options)) &&
 460        (0 !=
 461         memcmp (&start->self,
 462                 &GST_my_identity,
 463                 sizeof (struct GNUNET_PeerIdentity)) ) )
 464   {
 465     /* client thinks this is a different peer, reject */
 466     GNUNET_break (0);
 467     GNUNET_SERVICE_client_drop (tc->client);
 468     return;
 469   }
 470   if (CT_NONE != tc->type)
 471   {
 472     GNUNET_break (0);
 473     GNUNET_SERVICE_client_drop (tc->client);
 474     return;
 475   }
 476   tc->type = CT_CORE;
 477   GNUNET_SERVICE_client_continue (tc->client);
 478 }
 479
 480
 481 /**
 482  * Client asked for transmission to a peer.  Process the request.
 483  *
 484  * @param cls the client
 485  * @param obm the send message that was sent
 486  */
 487 static int
 488 check_client_send (void *cls,
 489                    const struct OutboundMessage *obm)
 490 {
 491   struct TransportClient *tc = cls;
 492   uint16_t size;
 493   const struct GNUNET_MessageHeader *obmm;
 494
 495   if (CT_CORE != tc->type)
 496   {
 497     GNUNET_break (0);
 498     return GNUNET_SYSERR;
 499   }
 500   size = ntohs (obm->header.size) - sizeof (struct OutboundMessage);
 501   if (size < sizeof (struct GNUNET_MessageHeader))
 502   {
 503     GNUNET_break (0);
 504     return GNUNET_SYSERR;
 505   }
 506   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
 507   if (size != ntohs (obmm->size))
 508   {
 509     GNUNET_break (0);
 510     return GNUNET_SYSERR;
 511   }
 512   return GNUNET_OK;
 513 }
 514
 515
 516 /**
 517  * Send a response to the @a pm that we have processed a
 518  * "send" request with status @a success. We
 519  * transmitted @a bytes_physical on the actual wire.
 520  * Sends a confirmation to the "core" client responsible
 521  * for the original request and free's @a pm.
 522  *
 523  * @param pm handle to the original pending message
 524  * @param success status code, #GNUNET_OK on success, #GNUNET_SYSERR
 525  *          for transmission failure
 526  * @param bytes_physical amount of bandwidth consumed
 527  */
 528 static void
 529 client_send_response (struct PendingMessage *pm,
 530                       int success,
 531                       uint32_t bytes_physical)
 532 {
 533   struct TransportClient *tc = pm->client;
 534   struct Neighbour *target = pm->target;
 535   struct GNUNET_MQ_Envelope *env;
 536   struct SendOkMessage *som;
 537
 538   if (NULL != tc)
 539   {
 540     env = GNUNET_MQ_msg (som,
 541                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
 542     som->success = htonl ((uint32_t) success);
 543     som->bytes_msg = htonl (pm->bytes_msg);
 544     som->bytes_physical = htonl (bytes_physical);
 545     som->peer = target->pid;
 546     GNUNET_MQ_send (tc->mq,
 547                     env);
 548     GNUNET_CONTAINER_MDLL_remove (client,
 549                                   tc->details.core.pending_msg_head,
 550                                   tc->details.core.pending_msg_tail,
 551                                   pm);
 552   }
 553   GNUNET_CONTAINER_MDLL_remove (neighbour,
 554                                 target->pending_msg_head,
 555                                 target->pending_msg_tail,
 556                                 pm);
 557   GNUNET_free (pm);
 558 }
 559
 560
 561 /**
 562  * Client asked for transmission to a peer.  Process the request.
 563  *
 564  * @param cls the client
 565  * @param obm the send message that was sent
 566  */
 567 static void
 568 handle_client_send (void *cls,
 569                     const struct OutboundMessage *obm)
 570 {
 571   struct TransportClient *tc = cls;
 572   struct PendingMessage *pm;
 573   const struct GNUNET_MessageHeader *obmm;
 574   struct Neighbour *target;
 575   uint32_t bytes_msg;
 576
 577   GNUNET_assert (CT_CORE == tc->type);
 578   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
 579   bytes_msg = ntohs (obmm->size);
 580   target = lookup_neighbour (&obm->peer);
 581   if (NULL == target)
 582   {
 583     /* Failure: don't have this peer as a neighbour (anymore).
 584        Might have gone down asynchronously, so this is NOT
 585        a protocol violation by CORE. Still count the event,
 586        as this should be rare. */
 587     struct GNUNET_MQ_Envelope *env;
 588     struct SendOkMessage *som;
 589
 590     env = GNUNET_MQ_msg (som,
 591                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
 592     som->success = htonl (GNUNET_SYSERR);
 593     som->bytes_msg = htonl (bytes_msg);
 594     som->bytes_physical = htonl (0);
 595     som->peer = obm->peer;
 596     GNUNET_MQ_send (tc->mq,
 597                     env);
 598     GNUNET_SERVICE_client_continue (tc->client);
 599     GNUNET_STATISTICS_update (GST_stats,
 600                               "# messages dropped (neighbour unknown)",
 601                               1,
 602                               GNUNET_NO);
 603     return;
 604   }
 605   pm = GNUNET_new (struct PendingMessage);
 606   pm->client = tc;
 607   pm->target = target;
 608   pm->bytes_msg = bytes_msg;
 609   GNUNET_CONTAINER_MDLL_insert (neighbour,
 610                                 target->pending_msg_head,
 611                                 target->pending_msg_tail,
 612                                 pm);
 613   GNUNET_CONTAINER_MDLL_insert (client,
 614                                 tc->details.core.pending_msg_head,
 615                                 tc->details.core.pending_msg_tail,
 616                                 pm);
 617   // FIXME: do the work, continuation with:
 618   client_send_response (pm,
 619                         GNUNET_NO,
 620                         0);
 621 }
 622
 623
 624 /**
 625  * Communicator started.  Test message is well-formed.
 626  *
 627  * @param cls the client
 628  * @param cam the send message that was sent
 629  */
 630 static int
 631 check_communicator_available (void *cls,
 632                               const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
 633 {
 634   struct TransportClient *tc = cls;
 635   const char *addr;
 636   uint16_t size;
 637
 638   if (CT_NONE != tc->type)
 639   {
 640     GNUNET_break (0);
 641     return GNUNET_SYSERR;
 642   }
 643   tc->type = CT_COMMUNICATOR;
 644   size = ntohs (cam->header.size) - sizeof (*cam);
 645   if (0 == size)
 646     return GNUNET_OK; /* receive-only communicator */
 647   addr = (const char *) &cam[1];
 648   if ('\0' != addr[size-1])
 649   {
 650     GNUNET_break (0);
 651     return GNUNET_SYSERR;
 652   }
 653   return GNUNET_OK;
 654 }
 655
 656
 657 /**
 658  * Communicator started.  Process the request.
 659  *
 660  * @param cls the client
 661  * @param cam the send message that was sent
 662  */
 663 static void
 664 handle_communicator_available (void *cls,
 665                                const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
 666 {
 667   struct TransportClient *tc = cls;
 668   uint16_t size;
 669
 670   size = ntohs (cam->header.size) - sizeof (*cam);
 671   if (0 == size)
 672     return; /* receive-only communicator */
 673   tc->details.communicator.address_prefix = GNUNET_strdup ((const char *) &cam[1]);
 674   GNUNET_SERVICE_client_continue (tc->client);
 675 }
 676
 677
 678 /**
 679  * Address of our peer added.  Test message is well-formed.
 680  *
 681  * @param cls the client
 682  * @param aam the send message that was sent
 683  */
 684 static int
 685 check_add_address (void *cls,
 686                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
 687 {
 688   struct TransportClient *tc = cls;
 689   const char *addr;
 690   uint16_t size;
 691
 692   if (CT_COMMUNICATOR != tc->type)
 693   {
 694     GNUNET_break (0);
 695     return GNUNET_SYSERR;
 696   }
 697   size = ntohs (aam->header.size) - sizeof (*aam);
 698   if (0 == size)
 699   {
 700     GNUNET_break (0);
 701     return GNUNET_SYSERR;
 702   }
 703   addr = (const char *) &aam[1];
 704   if ('\0' != addr[size-1])
 705   {
 706     GNUNET_break (0);
 707     return GNUNET_SYSERR;
 708   }
 709   return GNUNET_OK;
 710 }
 711
 712
 713 /**
 714  * Address of our peer added.  Process the request.
 715  *
 716  * @param cls the client
 717  * @param aam the send message that was sent
 718  */
 719 static void
 720 handle_add_address (void *cls,
 721                     const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
 722 {
 723   struct TransportClient *tc = cls;
 724
 725   GNUNET_SERVICE_client_continue (tc->client);
 726 }
 727
 728
 729 /**
 730  * Address of our peer deleted.  Process the request.
 731  *
 732  * @param cls the client
 733  * @param dam the send message that was sent
 734  */
 735 static void
 736 handle_del_address (void *cls,
 737                     const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
 738 {
 739   struct TransportClient *tc = cls;
 740
 741   if (CT_COMMUNICATOR != tc->type)
 742   {
 743     GNUNET_break (0);
 744     GNUNET_SERVICE_client_drop (tc->client);
 745     return;
 746   }
 747
 748   GNUNET_SERVICE_client_continue (tc->client);
 749 }
 750
 751
 752 /**
 753  * Client notified us about transmission from a peer.  Process the request.
 754  *
 755  * @param cls the client
 756  * @param obm the send message that was sent
 757  */
 758 static int
 759 check_incoming_msg (void *cls,
 760                     const struct GNUNET_TRANSPORT_IncomingMessage *im)
 761 {
 762   struct TransportClient *tc = cls;
 763   uint16_t size;
 764   const struct GNUNET_MessageHeader *obmm;
 765
 766   if (CT_COMMUNICATOR != tc->type)
 767   {
 768     GNUNET_break (0);
 769     return GNUNET_SYSERR;
 770   }
 771   size = ntohs (im->header.size) - sizeof (*im);
 772   if (size < sizeof (struct GNUNET_MessageHeader))
 773   {
 774     GNUNET_break (0);
 775     return GNUNET_SYSERR;
 776   }
 777   obmm = (const struct GNUNET_MessageHeader *) &im[1];
 778   if (size != ntohs (obmm->size))
 779   {
 780     GNUNET_break (0);
 781     return GNUNET_SYSERR;
 782   }
 783   return GNUNET_OK;
 784 }
 785
 786
 787 /**
 788  * Incoming meessage.  Process the request.
 789  *
 790  * @param cls the client
 791  * @param im the send message that was received
 792  */
 793 static void
 794 handle_incoming_msg (void *cls,
 795                      const struct GNUNET_TRANSPORT_IncomingMessage *im)
 796 {
 797   struct TransportClient *tc = cls;
 798
 799   GNUNET_SERVICE_client_continue (tc->client);
 800 }
 801
 802
 803 /**
 804  * New queue became available.  Check message.
 805  *
 806  * @param cls the client
 807  * @param aqm the send message that was sent
 808  */
 809 static int
 810 check_add_queue_message (void *cls,
 811                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 812 {
 813   struct TransportClient *tc = cls;
 814   const char *addr;
 815   uint16_t size;
 816
 817   if (CT_COMMUNICATOR != tc->type)
 818   {
 819     GNUNET_break (0);
 820     return GNUNET_SYSERR;
 821   }
 822   size = ntohs (aqm->header.size) - sizeof (*aqm);
 823   if (0 == size)
 824   {
 825     GNUNET_break (0);
 826     return GNUNET_SYSERR;
 827   }
 828   addr = (const char *) &aqm[1];
 829   if ('\0' != addr[size-1])
 830   {
 831     GNUNET_break (0);
 832     return GNUNET_SYSERR;
 833   }
 834   return GNUNET_OK;
 835 }
 836
 837
 838 /**
 839  * New queue became available.  Process the request.
 840  *
 841  * @param cls the client
 842  * @param aqm the send message that was sent
 843  */
 844 static void
 845 handle_add_queue_message (void *cls,
 846                           const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 847 {
 848   struct TransportClient *tc = cls;
 849   struct Queue *queue;
 850   struct Neighbour *neighbour;
 851   const char *addr;
 852   uint16_t addr_len;
 853
 854   neighbour = lookup_neighbour (&aqm->receiver);
 855   if (NULL == neighbour)
 856   {
 857     neighbour = GNUNET_new (struct Neighbour);
 858     neighbour->pid = aqm->receiver;
 859     GNUNET_assert (GNUNET_OK ==
 860                    GNUNET_CONTAINER_multipeermap_put (neighbours,
 861                                                       &neighbour->pid,
 862                                                       neighbour,
 863                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 864     // FIXME: notify cores/monitors!
 865   }
 866   addr_len = ntohs (aqm->header.size) - sizeof (*aqm);
 867   addr = (const char *) &aqm[1];
 868
 869   queue = GNUNET_malloc (sizeof (struct Queue) + addr_len);
 870   queue->qid = aqm->qid;
 871   queue->nt = (enum GNUNET_ATS_Network_Type) ntohl (aqm->nt);
 872   queue->tc = tc;
 873   queue->neighbour = neighbour;
 874   queue->address = (const char *) &queue[1];
 875   memcpy (&queue[1],
 876           addr,
 877           addr_len);
 878   GNUNET_CONTAINER_MDLL_insert (neighbour,
 879                                 neighbour->queue_head,
 880                                 neighbour->queue_tail,
 881                                 queue);
 882   GNUNET_CONTAINER_MDLL_insert (client,
 883                                 tc->details.communicator.queue_head,
 884                                 tc->details.communicator.queue_tail,
 885                                 queue);
 886   // FIXME: possibly transmit queued messages?
 887   GNUNET_SERVICE_client_continue (tc->client);
 888 }
 889
 890
 891 /**
 892  * Release memory used by @a neighbour.
 893  *
 894  * @param neighbour neighbour entry to free
 895  */
 896 static void
 897 free_neighbour (struct Neighbour *neighbour)
 898 {
 899   GNUNET_assert (NULL == neighbour->queue_head);
 900   GNUNET_assert (GNUNET_YES ==
 901                  GNUNET_CONTAINER_multipeermap_remove (neighbours,
 902                                                        &neighbour->pid,
 903                                                        neighbour));
 904   GNUNET_free (neighbour);
 905 }
 906
 907
 908 /**
 909  * Queue to a peer went down.  Process the request.
 910  *
 911  * @param cls the client
 912  * @param dqm the send message that was sent
 913  */
 914 static void
 915 handle_del_queue_message (void *cls,
 916                           const struct GNUNET_TRANSPORT_DelQueueMessage *dqm)
 917 {
 918   struct TransportClient *tc = cls;
 919
 920   if (CT_COMMUNICATOR != tc->type)
 921   {
 922     GNUNET_break (0);
 923     GNUNET_SERVICE_client_drop (tc->client);
 924     return;
 925   }
 926   for (struct Queue *queue = tc->details.communicator.queue_head;
 927        NULL != queue;
 928        queue = queue->next_client)
 929   {
 930     struct Neighbour *neighbour = queue->neighbour;
 931
 932     if ( (dqm->qid != queue->qid) ||
 933          (0 != memcmp (&dqm->receiver,
 934                        &neighbour->pid,
 935                        sizeof (struct GNUNET_PeerIdentity))) )
 936       continue;
 937     GNUNET_CONTAINER_MDLL_remove (neighbour,
 938                                   neighbour->queue_head,
 939                                   neighbour->queue_tail,
 940                                   queue);
 941     GNUNET_CONTAINER_MDLL_remove (client,
 942                                   tc->details.communicator.queue_head,
 943                                   tc->details.communicator.queue_tail,
 944                                   queue);
 945     GNUNET_free (queue);
 946     if (NULL == neighbour->queue_head)
 947     {
 948       // FIXME: notify cores/monitors!
 949       free_neighbour (neighbour);
 950     }
 951     GNUNET_SERVICE_client_continue (tc->client);
 952     return;
 953   }
 954   GNUNET_break (0);
 955   GNUNET_SERVICE_client_drop (tc->client);
 956 }
 957
 958
 959 /**
 960  * Message was transmitted.  Process the request.
 961  *
 962  * @param cls the client
 963  * @param sma the send message that was sent
 964  */
 965 static void
 966 handle_send_message_ack (void *cls,
 967                          const struct GNUNET_TRANSPORT_SendMessageToAck *sma)
 968 {
 969   struct TransportClient *tc = cls;
 970
 971   if (CT_COMMUNICATOR != tc->type)
 972   {
 973     GNUNET_break (0);
 974     GNUNET_SERVICE_client_drop (tc->client);
 975     return;
 976   }
 977   GNUNET_SERVICE_client_continue (tc->client);
 978 }
 979
 980
 981 /**
 982  * Initialize a monitor client.
 983  *
 984  * @param cls the client
 985  * @param start the start message that was sent
 986  */
 987 static void
 988 handle_monitor_start (void *cls,
 989                      const struct GNUNET_TRANSPORT_MonitorStart *start)
 990 {
 991   struct TransportClient *tc = cls;
 992
 993   if (CT_NONE != tc->type)
 994   {
 995     GNUNET_break (0);
 996     GNUNET_SERVICE_client_drop (tc->client);
 997     return;
 998   }
 999   tc->type = CT_MONITOR;
1000   tc->details.monitor.peer = start->peer;
1001   tc->details.monitor.one_shot = ntohl (start->one_shot);
1002   // FIXME: do work!
1003   GNUNET_SERVICE_client_continue (tc->client);
1004 }
1005
1006
1007 /**
1008  * Free neighbour entry.
1009  *
1010  * @param cls NULL
1011  * @param pid unused
1012  * @param value a `struct Neighbour`
1013  * @return #GNUNET_OK (always)
1014  */
1015 static int
1016 free_neighbour_cb (void *cls,
1017                    const struct GNUNET_PeerIdentity *pid,
1018                    void *value)
1019 {
1020   struct Neighbour *neighbour = value;
1021
1022   (void) cls;
1023   (void) pid;
1024   GNUNET_break (0); // should this ever happen?
1025   free_neighbour (neighbour);
1026
1027   return GNUNET_OK;
1028 }
1029
1030
1031 /**
1032  * Function called when the service shuts down.  Unloads our plugins
1033  * and cancels pending validations.
1034  *
1035  * @param cls closure, unused
1036  */
1037 static void
1038 do_shutdown (void *cls)
1039 {
1040   (void) cls;
1041
1042   if (NULL != GST_stats)
1043   {
1044     GNUNET_STATISTICS_destroy (GST_stats,
1045                                GNUNET_NO);
1046     GST_stats = NULL;
1047   }
1048   if (NULL != GST_my_private_key)
1049   {
1050     GNUNET_free (GST_my_private_key);
1051     GST_my_private_key = NULL;
1052   }
1053   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
1054                                          &free_neighbour_cb,
1055                                          NULL);
1056   GNUNET_CONTAINER_multipeermap_destroy (neighbours);
1057 }
1058
1059
1060 /**
1061  * Initiate transport service.
1062  *
1063  * @param cls closure
1064  * @param c configuration to use
1065  * @param service the initialized service
1066  */
1067 static void
1068 run (void *cls,
1069      const struct GNUNET_CONFIGURATION_Handle *c,
1070      struct GNUNET_SERVICE_Handle *service)
1071 {
1072   (void) cls;
1073   /* setup globals */
1074   GST_cfg = c;
1075   neighbours = GNUNET_CONTAINER_multipeermap_create (1024,
1076                                                      GNUNET_YES);
1077   GST_my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (GST_cfg);
1078   if (NULL == GST_my_private_key)
1079   {
1080     GNUNET_log(GNUNET_ERROR_TYPE_ERROR,
1081         _("Transport service is lacking key configuration settings. Exiting.\n"));
1082     GNUNET_SCHEDULER_shutdown ();
1083     return;
1084   }
1085   GNUNET_CRYPTO_eddsa_key_get_public (GST_my_private_key,
1086                                       &GST_my_identity.public_key);
1087   GNUNET_log(GNUNET_ERROR_TYPE_INFO,
1088              "My identity is `%s'\n",
1089              GNUNET_i2s_full (&GST_my_identity));
1090
1091   GST_stats = GNUNET_STATISTICS_create ("transport",
1092                                         GST_cfg);
1093   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
1094                                  NULL);
1095   /* start subsystems */
1096 }
1097
1098
1099 /**
1100  * Define "main" method using service macro.
1101  */
1102 GNUNET_SERVICE_MAIN
1103 ("transport",
1104  GNUNET_SERVICE_OPTION_NONE,
1105  &run,
1106  &client_connect_cb,
1107  &client_disconnect_cb,
1108  NULL,
1109  /* communication with core */
1110  GNUNET_MQ_hd_fixed_size (client_start,
1111                           GNUNET_MESSAGE_TYPE_TRANSPORT_START,
1112                           struct StartMessage,
1113                           NULL),
1114  GNUNET_MQ_hd_var_size (client_send,
1115                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND,
1116                         struct OutboundMessage,
1117                         NULL),
1118  /* communication with communicators */
1119  GNUNET_MQ_hd_var_size (communicator_available,
1120                         GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
1121                         struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
1122                         NULL),
1123  GNUNET_MQ_hd_var_size (add_address,
1124                         GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
1125                         struct GNUNET_TRANSPORT_AddAddressMessage,
1126                         NULL),
1127  GNUNET_MQ_hd_fixed_size (del_address,
1128                           GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
1129                           struct GNUNET_TRANSPORT_DelAddressMessage,
1130                           NULL),
1131  GNUNET_MQ_hd_var_size (incoming_msg,
1132                         GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
1133                         struct GNUNET_TRANSPORT_IncomingMessage,
1134                         NULL),
1135  GNUNET_MQ_hd_var_size (add_queue_message,
1136                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
1137                         struct GNUNET_TRANSPORT_AddQueueMessage,
1138                         NULL),
1139  GNUNET_MQ_hd_fixed_size (del_queue_message,
1140                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
1141                           struct GNUNET_TRANSPORT_DelQueueMessage,
1142                           NULL),
1143  GNUNET_MQ_hd_fixed_size (send_message_ack,
1144                           GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
1145                           struct GNUNET_TRANSPORT_SendMessageToAck,
1146                           NULL),
1147  /* communication with monitors */
1148  GNUNET_MQ_hd_fixed_size (monitor_start,
1149                           GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_START,
1150                           struct GNUNET_TRANSPORT_MonitorStart,
1151                           NULL),
1152  GNUNET_MQ_handler_end ());
1153
1154
1155 /* end of file gnunet-service-transport.c */