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    * Address served by the queue.
 124    */
 125   const char *address;
 126
 127   /**
 128    * Unique identifier of this queue with the communicator.
 129    */
 130   uint32_t qid;
 131
 132   /**
 133    * Network type offered by this queue.
 134    */
 135   enum GNUNET_ATS_Network_Type nt;
 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  * One of the addresses of this peer.
 218  */
 219 struct AddressListEntry
 220 {
 221
 222   /**
 223    * Kept in a DLL.
 224    */
 225   struct AddressListEntry *next;
 226
 227   /**
 228    * Kept in a DLL.
 229    */
 230   struct AddressListEntry *prev;
 231
 232   /**
 233    * Which communicator provides this address?
 234    */
 235   struct TransportClient *tc;
 236
 237   /**
 238    * The actual address.
 239    */
 240   const char *address;
 241
 242   /**
 243    * What is a typical lifetime the communicator expects this
 244    * address to have? (Always from now.)
 245    */
 246   struct GNUNET_TIME_Relative expiration;
 247
 248   /**
 249    * Address identifier used by the communicator.
 250    */
 251   uint32_t aid;
 252
 253   /**
 254    * Network type offered by this address.
 255    */
 256   enum GNUNET_ATS_Network_Type nt;
 257
 258 };
 259
 260
 261 /**
 262  * Client connected to the transport service.
 263  */
 264 struct TransportClient
 265 {
 266
 267   /**
 268    * Kept in a DLL.
 269    */
 270   struct TransportClient *next;
 271
 272   /**
 273    * Kept in a DLL.
 274    */
 275   struct TransportClient *prev;
 276
 277   /**
 278    * Handle to the client.
 279    */
 280   struct GNUNET_SERVICE_Client *client;
 281
 282   /**
 283    * Message queue to the client.
 284    */
 285   struct GNUNET_MQ_Handle *mq;
 286
 287   /**
 288    * What type of client is this?
 289    */
 290   enum ClientType type;
 291
 292   union
 293   {
 294
 295     /**
 296      * Information for @e type #CT_CORE.
 297      */
 298     struct {
 299
 300       /**
 301        * Head of list of messages pending for this client.
 302        */
 303       struct PendingMessage *pending_msg_head;
 304
 305       /**
 306        * Tail of list of messages pending for this client.
 307        */
 308       struct PendingMessage *pending_msg_tail;
 309
 310     } core;
 311
 312     /**
 313      * Information for @e type #CT_MONITOR.
 314      */
 315     struct {
 316
 317       /**
 318        * Peer identity to monitor the addresses of.
 319        * Zero to monitor all neighbours.  Valid if
 320        * @e type is #CT_MONITOR.
 321        */
 322       struct GNUNET_PeerIdentity peer;
 323
 324       /**
 325        * Is this a one-shot monitor?
 326        */
 327       int one_shot;
 328
 329     } monitor;
 330
 331
 332     /**
 333      * Information for @e type #CT_COMMUNICATOR.
 334      */
 335     struct {
 336       /**
 337        * If @e type is #CT_COMMUNICATOR, this communicator
 338        * supports communicating using these addresses.
 339        */
 340       char *address_prefix;
 341
 342       /**
 343        * Head of DLL of queues offered by this communicator.
 344        */
 345       struct Queue *queue_head;
 346
 347       /**
 348        * Tail of DLL of queues offered by this communicator.
 349        */
 350       struct Queue *queue_tail;
 351
 352       /**
 353        * Head of list of the addresses of this peer offered by this communicator.
 354        */
 355       struct AddressListEntry *addr_head;
 356
 357       /**
 358        * Tail of list of the addresses of this peer offered by this communicator.
 359        */
 360       struct AddressListEntry *addr_tail;
 361
 362     } communicator;
 363
 364   } details;
 365
 366 };
 367
 368
 369 /**
 370  * Head of linked list of all clients to this service.
 371  */
 372 static struct TransportClient *clients_head;
 373
 374 /**
 375  * Tail of linked list of all clients to this service.
 376  */
 377 static struct TransportClient *clients_tail;
 378
 379 /**
 380  * Statistics handle.
 381  */
 382 struct GNUNET_STATISTICS_Handle *GST_stats;
 383
 384 /**
 385  * Configuration handle.
 386  */
 387 const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
 388
 389 /**
 390  * Our public key.
 391  */
 392 struct GNUNET_PeerIdentity GST_my_identity;
 393
 394 /**
 395  * Our private key.
 396  */
 397 struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
 398
 399 /**
 400  * Map from PIDs to `struct Neighbour` entries.  A peer is
 401  * a neighbour if we have an MQ to it from some communicator.
 402  */
 403 static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
 404
 405
 406 /**
 407  * Lookup neighbour record for peer @a pid.
 408  *
 409  * @param pid neighbour to look for
 410  * @return NULL if we do not have this peer as a neighbour
 411  */
 412 static struct Neighbour *
 413 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
 414 {
 415   return GNUNET_CONTAINER_multipeermap_get (neighbours,
 416                                             pid);
 417 }
 418
 419
 420 /**
 421  * Called whenever a client connects.  Allocates our
 422  * data structures associated with that client.
 423  *
 424  * @param cls closure, NULL
 425  * @param client identification of the client
 426  * @param mq message queue for the client
 427  * @return our `struct TransportClient`
 428  */
 429 static void *
 430 client_connect_cb (void *cls,
 431                    struct GNUNET_SERVICE_Client *client,
 432                    struct GNUNET_MQ_Handle *mq)
 433 {
 434   struct TransportClient *tc;
 435
 436   tc = GNUNET_new (struct TransportClient);
 437   tc->client = client;
 438   tc->mq = mq;
 439   GNUNET_CONTAINER_DLL_insert (clients_head,
 440                                clients_tail,
 441                                tc);
 442   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 443               "Client %p connected\n",
 444               tc);
 445   return tc;
 446 }
 447
 448
 449 /**
 450  * Called whenever a client is disconnected.  Frees our
 451  * resources associated with that client.
 452  *
 453  * @param cls closure, NULL
 454  * @param client identification of the client
 455  * @param app_ctx our `struct TransportClient`
 456  */
 457 static void
 458 client_disconnect_cb (void *cls,
 459                       struct GNUNET_SERVICE_Client *client,
 460                       void *app_ctx)
 461 {
 462   struct TransportClient *tc = app_ctx;
 463
 464   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 465               "Client %p disconnected, cleaning up.\n",
 466               tc);
 467   GNUNET_CONTAINER_DLL_remove (clients_head,
 468                                clients_tail,
 469                                tc);
 470   switch (tc->type)
 471   {
 472   case CT_NONE:
 473     break;
 474   case CT_CORE:
 475     {
 476       struct PendingMessage *pm;
 477
 478       while (NULL != (pm = tc->details.core.pending_msg_head))
 479       {
 480         GNUNET_CONTAINER_MDLL_remove (client,
 481                                       tc->details.core.pending_msg_head,
 482                                       tc->details.core.pending_msg_tail,
 483                                       pm);
 484         pm->client = NULL;
 485       }
 486     }
 487     break;
 488   case CT_MONITOR:
 489     break;
 490   case CT_COMMUNICATOR:
 491     GNUNET_free (tc->details.communicator.address_prefix);
 492     break;
 493   }
 494   GNUNET_free (tc);
 495 }
 496
 497
 498 /**
 499  * Initialize a "CORE" client.  We got a start message from this
 500  * client, so add it to the list of clients for broadcasting of
 501  * inbound messages.
 502  *
 503  * @param cls the client
 504  * @param start the start message that was sent
 505  */
 506 static void
 507 handle_client_start (void *cls,
 508                      const struct StartMessage *start)
 509 {
 510   struct TransportClient *tc = cls;
 511   uint32_t options;
 512
 513   options = ntohl (start->options);
 514   if ( (0 != (1 & options)) &&
 515        (0 !=
 516         memcmp (&start->self,
 517                 &GST_my_identity,
 518                 sizeof (struct GNUNET_PeerIdentity)) ) )
 519   {
 520     /* client thinks this is a different peer, reject */
 521     GNUNET_break (0);
 522     GNUNET_SERVICE_client_drop (tc->client);
 523     return;
 524   }
 525   if (CT_NONE != tc->type)
 526   {
 527     GNUNET_break (0);
 528     GNUNET_SERVICE_client_drop (tc->client);
 529     return;
 530   }
 531   tc->type = CT_CORE;
 532   GNUNET_SERVICE_client_continue (tc->client);
 533 }
 534
 535
 536 /**
 537  * Client asked for transmission to a peer.  Process the request.
 538  *
 539  * @param cls the client
 540  * @param obm the send message that was sent
 541  */
 542 static int
 543 check_client_send (void *cls,
 544                    const struct OutboundMessage *obm)
 545 {
 546   struct TransportClient *tc = cls;
 547   uint16_t size;
 548   const struct GNUNET_MessageHeader *obmm;
 549
 550   if (CT_CORE != tc->type)
 551   {
 552     GNUNET_break (0);
 553     return GNUNET_SYSERR;
 554   }
 555   size = ntohs (obm->header.size) - sizeof (struct OutboundMessage);
 556   if (size < sizeof (struct GNUNET_MessageHeader))
 557   {
 558     GNUNET_break (0);
 559     return GNUNET_SYSERR;
 560   }
 561   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
 562   if (size != ntohs (obmm->size))
 563   {
 564     GNUNET_break (0);
 565     return GNUNET_SYSERR;
 566   }
 567   return GNUNET_OK;
 568 }
 569
 570
 571 /**
 572  * Send a response to the @a pm that we have processed a
 573  * "send" request with status @a success. We
 574  * transmitted @a bytes_physical on the actual wire.
 575  * Sends a confirmation to the "core" client responsible
 576  * for the original request and free's @a pm.
 577  *
 578  * @param pm handle to the original pending message
 579  * @param success status code, #GNUNET_OK on success, #GNUNET_SYSERR
 580  *          for transmission failure
 581  * @param bytes_physical amount of bandwidth consumed
 582  */
 583 static void
 584 client_send_response (struct PendingMessage *pm,
 585                       int success,
 586                       uint32_t bytes_physical)
 587 {
 588   struct TransportClient *tc = pm->client;
 589   struct Neighbour *target = pm->target;
 590   struct GNUNET_MQ_Envelope *env;
 591   struct SendOkMessage *som;
 592
 593   if (NULL != tc)
 594   {
 595     env = GNUNET_MQ_msg (som,
 596                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
 597     som->success = htonl ((uint32_t) success);
 598     som->bytes_msg = htonl (pm->bytes_msg);
 599     som->bytes_physical = htonl (bytes_physical);
 600     som->peer = target->pid;
 601     GNUNET_MQ_send (tc->mq,
 602                     env);
 603     GNUNET_CONTAINER_MDLL_remove (client,
 604                                   tc->details.core.pending_msg_head,
 605                                   tc->details.core.pending_msg_tail,
 606                                   pm);
 607   }
 608   GNUNET_CONTAINER_MDLL_remove (neighbour,
 609                                 target->pending_msg_head,
 610                                 target->pending_msg_tail,
 611                                 pm);
 612   GNUNET_free (pm);
 613 }
 614
 615
 616 /**
 617  * Client asked for transmission to a peer.  Process the request.
 618  *
 619  * @param cls the client
 620  * @param obm the send message that was sent
 621  */
 622 static void
 623 handle_client_send (void *cls,
 624                     const struct OutboundMessage *obm)
 625 {
 626   struct TransportClient *tc = cls;
 627   struct PendingMessage *pm;
 628   const struct GNUNET_MessageHeader *obmm;
 629   struct Neighbour *target;
 630   uint32_t bytes_msg;
 631
 632   GNUNET_assert (CT_CORE == tc->type);
 633   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
 634   bytes_msg = ntohs (obmm->size);
 635   target = lookup_neighbour (&obm->peer);
 636   if (NULL == target)
 637   {
 638     /* Failure: don't have this peer as a neighbour (anymore).
 639        Might have gone down asynchronously, so this is NOT
 640        a protocol violation by CORE. Still count the event,
 641        as this should be rare. */
 642     struct GNUNET_MQ_Envelope *env;
 643     struct SendOkMessage *som;
 644
 645     env = GNUNET_MQ_msg (som,
 646                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
 647     som->success = htonl (GNUNET_SYSERR);
 648     som->bytes_msg = htonl (bytes_msg);
 649     som->bytes_physical = htonl (0);
 650     som->peer = obm->peer;
 651     GNUNET_MQ_send (tc->mq,
 652                     env);
 653     GNUNET_SERVICE_client_continue (tc->client);
 654     GNUNET_STATISTICS_update (GST_stats,
 655                               "# messages dropped (neighbour unknown)",
 656                               1,
 657                               GNUNET_NO);
 658     return;
 659   }
 660   pm = GNUNET_new (struct PendingMessage);
 661   pm->client = tc;
 662   pm->target = target;
 663   pm->bytes_msg = bytes_msg;
 664   GNUNET_CONTAINER_MDLL_insert (neighbour,
 665                                 target->pending_msg_head,
 666                                 target->pending_msg_tail,
 667                                 pm);
 668   GNUNET_CONTAINER_MDLL_insert (client,
 669                                 tc->details.core.pending_msg_head,
 670                                 tc->details.core.pending_msg_tail,
 671                                 pm);
 672   // FIXME: do the work, continuation with:
 673   client_send_response (pm,
 674                         GNUNET_NO,
 675                         0);
 676 }
 677
 678
 679 /**
 680  * Communicator started.  Test message is well-formed.
 681  *
 682  * @param cls the client
 683  * @param cam the send message that was sent
 684  */
 685 static int
 686 check_communicator_available (void *cls,
 687                               const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
 688 {
 689   struct TransportClient *tc = cls;
 690   const char *addr;
 691   uint16_t size;
 692
 693   if (CT_NONE != tc->type)
 694   {
 695     GNUNET_break (0);
 696     return GNUNET_SYSERR;
 697   }
 698   tc->type = CT_COMMUNICATOR;
 699   size = ntohs (cam->header.size) - sizeof (*cam);
 700   if (0 == size)
 701     return GNUNET_OK; /* receive-only communicator */
 702   addr = (const char *) &cam[1];
 703   if ('\0' != addr[size-1])
 704   {
 705     GNUNET_break (0);
 706     return GNUNET_SYSERR;
 707   }
 708   return GNUNET_OK;
 709 }
 710
 711
 712 /**
 713  * Communicator started.  Process the request.
 714  *
 715  * @param cls the client
 716  * @param cam the send message that was sent
 717  */
 718 static void
 719 handle_communicator_available (void *cls,
 720                                const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
 721 {
 722   struct TransportClient *tc = cls;
 723   uint16_t size;
 724
 725   size = ntohs (cam->header.size) - sizeof (*cam);
 726   if (0 == size)
 727     return; /* receive-only communicator */
 728   tc->details.communicator.address_prefix = GNUNET_strdup ((const char *) &cam[1]);
 729   GNUNET_SERVICE_client_continue (tc->client);
 730 }
 731
 732
 733 /**
 734  * Address of our peer added.  Test message is well-formed.
 735  *
 736  * @param cls the client
 737  * @param aam the send message that was sent
 738  */
 739 static int
 740 check_add_address (void *cls,
 741                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
 742 {
 743   struct TransportClient *tc = cls;
 744   const char *addr;
 745   uint16_t size;
 746
 747   if (CT_COMMUNICATOR != tc->type)
 748   {
 749     GNUNET_break (0);
 750     return GNUNET_SYSERR;
 751   }
 752   size = ntohs (aam->header.size) - sizeof (*aam);
 753   if (0 == size)
 754   {
 755     GNUNET_break (0);
 756     return GNUNET_SYSERR;
 757   }
 758   addr = (const char *) &aam[1];
 759   if ('\0' != addr[size-1])
 760   {
 761     GNUNET_break (0);
 762     return GNUNET_SYSERR;
 763   }
 764   return GNUNET_OK;
 765 }
 766
 767
 768 /**
 769  * Address of our peer added.  Process the request.
 770  *
 771  * @param cls the client
 772  * @param aam the send message that was sent
 773  */
 774 static void
 775 handle_add_address (void *cls,
 776                     const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
 777 {
 778   struct TransportClient *tc = cls;
 779   struct AddressListEntry *ale;
 780   size_t slen;
 781
 782   slen = ntohs (aam->header.size) - sizeof (*aam);
 783   ale = GNUNET_malloc (sizeof (struct AddressListEntry) + slen);
 784   ale->tc = tc;
 785   ale->address = (const char *) &ale[1];
 786   ale->expiration = GNUNET_TIME_relative_ntoh (aam->expiration);
 787   ale->aid = aam->aid;
 788   ale->nt = (enum GNUNET_ATS_Network_Type) ntohl (aam->nt);
 789   memcpy (&ale[1],
 790           &aam[1],
 791           slen);
 792   GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
 793                                tc->details.communicator.addr_tail,
 794                                ale);
 795   // FIXME: notify somebody?!
 796   GNUNET_SERVICE_client_continue (tc->client);
 797 }
 798
 799
 800 /**
 801  * Address of our peer deleted.  Process the request.
 802  *
 803  * @param cls the client
 804  * @param dam the send message that was sent
 805  */
 806 static void
 807 handle_del_address (void *cls,
 808                     const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
 809 {
 810   struct TransportClient *tc = cls;
 811
 812   if (CT_COMMUNICATOR != tc->type)
 813   {
 814     GNUNET_break (0);
 815     GNUNET_SERVICE_client_drop (tc->client);
 816     return;
 817   }
 818   for (struct AddressListEntry *ale = tc->details.communicator.addr_head;
 819        NULL != ale;
 820        ale = ale->next)
 821   {
 822     if (dam->aid != ale->aid)
 823       continue;
 824     GNUNET_assert (ale->tc == tc);
 825     GNUNET_CONTAINER_DLL_remove (tc->details.communicator.addr_head,
 826                                  tc->details.communicator.addr_tail,
 827                                  ale);
 828     // FIXME: notify somebody?
 829     GNUNET_free (ale);
 830     GNUNET_SERVICE_client_continue (tc->client);
 831   }
 832   GNUNET_break (0);
 833   GNUNET_SERVICE_client_drop (tc->client);
 834 }
 835
 836
 837 /**
 838  * Client notified us about transmission from a peer.  Process the request.
 839  *
 840  * @param cls the client
 841  * @param obm the send message that was sent
 842  */
 843 static int
 844 check_incoming_msg (void *cls,
 845                     const struct GNUNET_TRANSPORT_IncomingMessage *im)
 846 {
 847   struct TransportClient *tc = cls;
 848   uint16_t size;
 849   const struct GNUNET_MessageHeader *obmm;
 850
 851   if (CT_COMMUNICATOR != tc->type)
 852   {
 853     GNUNET_break (0);
 854     return GNUNET_SYSERR;
 855   }
 856   size = ntohs (im->header.size) - sizeof (*im);
 857   if (size < sizeof (struct GNUNET_MessageHeader))
 858   {
 859     GNUNET_break (0);
 860     return GNUNET_SYSERR;
 861   }
 862   obmm = (const struct GNUNET_MessageHeader *) &im[1];
 863   if (size != ntohs (obmm->size))
 864   {
 865     GNUNET_break (0);
 866     return GNUNET_SYSERR;
 867   }
 868   return GNUNET_OK;
 869 }
 870
 871
 872 /**
 873  * Incoming meessage.  Process the request.
 874  *
 875  * @param cls the client
 876  * @param im the send message that was received
 877  */
 878 static void
 879 handle_incoming_msg (void *cls,
 880                      const struct GNUNET_TRANSPORT_IncomingMessage *im)
 881 {
 882   struct TransportClient *tc = cls;
 883
 884   GNUNET_SERVICE_client_continue (tc->client);
 885 }
 886
 887
 888 /**
 889  * New queue became available.  Check message.
 890  *
 891  * @param cls the client
 892  * @param aqm the send message that was sent
 893  */
 894 static int
 895 check_add_queue_message (void *cls,
 896                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 897 {
 898   struct TransportClient *tc = cls;
 899   const char *addr;
 900   uint16_t size;
 901
 902   if (CT_COMMUNICATOR != tc->type)
 903   {
 904     GNUNET_break (0);
 905     return GNUNET_SYSERR;
 906   }
 907   size = ntohs (aqm->header.size) - sizeof (*aqm);
 908   if (0 == size)
 909   {
 910     GNUNET_break (0);
 911     return GNUNET_SYSERR;
 912   }
 913   addr = (const char *) &aqm[1];
 914   if ('\0' != addr[size-1])
 915   {
 916     GNUNET_break (0);
 917     return GNUNET_SYSERR;
 918   }
 919   return GNUNET_OK;
 920 }
 921
 922
 923 /**
 924  * New queue became available.  Process the request.
 925  *
 926  * @param cls the client
 927  * @param aqm the send message that was sent
 928  */
 929 static void
 930 handle_add_queue_message (void *cls,
 931                           const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 932 {
 933   struct TransportClient *tc = cls;
 934   struct Queue *queue;
 935   struct Neighbour *neighbour;
 936   const char *addr;
 937   uint16_t addr_len;
 938
 939   neighbour = lookup_neighbour (&aqm->receiver);
 940   if (NULL == neighbour)
 941   {
 942     neighbour = GNUNET_new (struct Neighbour);
 943     neighbour->pid = aqm->receiver;
 944     GNUNET_assert (GNUNET_OK ==
 945                    GNUNET_CONTAINER_multipeermap_put (neighbours,
 946                                                       &neighbour->pid,
 947                                                       neighbour,
 948                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 949     // FIXME: notify ATS/COREs/monitors!
 950   }
 951   addr_len = ntohs (aqm->header.size) - sizeof (*aqm);
 952   addr = (const char *) &aqm[1];
 953
 954   queue = GNUNET_malloc (sizeof (struct Queue) + addr_len);
 955   queue->qid = aqm->qid;
 956   queue->nt = (enum GNUNET_ATS_Network_Type) ntohl (aqm->nt);
 957   queue->tc = tc;
 958   queue->neighbour = neighbour;
 959   queue->address = (const char *) &queue[1];
 960   memcpy (&queue[1],
 961           addr,
 962           addr_len);
 963   GNUNET_CONTAINER_MDLL_insert (neighbour,
 964                                 neighbour->queue_head,
 965                                 neighbour->queue_tail,
 966                                 queue);
 967   GNUNET_CONTAINER_MDLL_insert (client,
 968                                 tc->details.communicator.queue_head,
 969                                 tc->details.communicator.queue_tail,
 970                                 queue);
 971   // FIXME: possibly transmit queued messages?
 972   GNUNET_SERVICE_client_continue (tc->client);
 973 }
 974
 975
 976 /**
 977  * Release memory used by @a neighbour.
 978  *
 979  * @param neighbour neighbour entry to free
 980  */
 981 static void
 982 free_neighbour (struct Neighbour *neighbour)
 983 {
 984   GNUNET_assert (NULL == neighbour->queue_head);
 985   GNUNET_assert (GNUNET_YES ==
 986                  GNUNET_CONTAINER_multipeermap_remove (neighbours,
 987                                                        &neighbour->pid,
 988                                                        neighbour));
 989   GNUNET_free (neighbour);
 990 }
 991
 992
 993 /**
 994  * Queue to a peer went down.  Process the request.
 995  *
 996  * @param cls the client
 997  * @param dqm the send message that was sent
 998  */
 999 static void
1000 handle_del_queue_message (void *cls,
1001                           const struct GNUNET_TRANSPORT_DelQueueMessage *dqm)
1002 {
1003   struct TransportClient *tc = cls;
1004
1005   if (CT_COMMUNICATOR != tc->type)
1006   {
1007     GNUNET_break (0);
1008     GNUNET_SERVICE_client_drop (tc->client);
1009     return;
1010   }
1011   for (struct Queue *queue = tc->details.communicator.queue_head;
1012        NULL != queue;
1013        queue = queue->next_client)
1014   {
1015     struct Neighbour *neighbour = queue->neighbour;
1016
1017     if ( (dqm->qid != queue->qid) ||
1018          (0 != memcmp (&dqm->receiver,
1019                        &neighbour->pid,
1020                        sizeof (struct GNUNET_PeerIdentity))) )
1021       continue;
1022     GNUNET_CONTAINER_MDLL_remove (neighbour,
1023                                   neighbour->queue_head,
1024                                   neighbour->queue_tail,
1025                                   queue);
1026     GNUNET_CONTAINER_MDLL_remove (client,
1027                                   tc->details.communicator.queue_head,
1028                                   tc->details.communicator.queue_tail,
1029                                   queue);
1030     GNUNET_free (queue);
1031     if (NULL == neighbour->queue_head)
1032     {
1033       // FIXME: notify cores/monitors!
1034       free_neighbour (neighbour);
1035     }
1036     GNUNET_SERVICE_client_continue (tc->client);
1037     return;
1038   }
1039   GNUNET_break (0);
1040   GNUNET_SERVICE_client_drop (tc->client);
1041 }
1042
1043
1044 /**
1045  * Message was transmitted.  Process the request.
1046  *
1047  * @param cls the client
1048  * @param sma the send message that was sent
1049  */
1050 static void
1051 handle_send_message_ack (void *cls,
1052                          const struct GNUNET_TRANSPORT_SendMessageToAck *sma)
1053 {
1054   struct TransportClient *tc = cls;
1055
1056   if (CT_COMMUNICATOR != tc->type)
1057   {
1058     GNUNET_break (0);
1059     GNUNET_SERVICE_client_drop (tc->client);
1060     return;
1061   }
1062   GNUNET_SERVICE_client_continue (tc->client);
1063 }
1064
1065
1066 /**
1067  * Initialize a monitor client.
1068  *
1069  * @param cls the client
1070  * @param start the start message that was sent
1071  */
1072 static void
1073 handle_monitor_start (void *cls,
1074                      const struct GNUNET_TRANSPORT_MonitorStart *start)
1075 {
1076   struct TransportClient *tc = cls;
1077
1078   if (CT_NONE != tc->type)
1079   {
1080     GNUNET_break (0);
1081     GNUNET_SERVICE_client_drop (tc->client);
1082     return;
1083   }
1084   tc->type = CT_MONITOR;
1085   tc->details.monitor.peer = start->peer;
1086   tc->details.monitor.one_shot = ntohl (start->one_shot);
1087   // FIXME: do work!
1088   GNUNET_SERVICE_client_continue (tc->client);
1089 }
1090
1091
1092 /**
1093  * Free neighbour entry.
1094  *
1095  * @param cls NULL
1096  * @param pid unused
1097  * @param value a `struct Neighbour`
1098  * @return #GNUNET_OK (always)
1099  */
1100 static int
1101 free_neighbour_cb (void *cls,
1102                    const struct GNUNET_PeerIdentity *pid,
1103                    void *value)
1104 {
1105   struct Neighbour *neighbour = value;
1106
1107   (void) cls;
1108   (void) pid;
1109   GNUNET_break (0); // should this ever happen?
1110   free_neighbour (neighbour);
1111
1112   return GNUNET_OK;
1113 }
1114
1115
1116 /**
1117  * Function called when the service shuts down.  Unloads our plugins
1118  * and cancels pending validations.
1119  *
1120  * @param cls closure, unused
1121  */
1122 static void
1123 do_shutdown (void *cls)
1124 {
1125   (void) cls;
1126
1127   if (NULL != GST_stats)
1128   {
1129     GNUNET_STATISTICS_destroy (GST_stats,
1130                                GNUNET_NO);
1131     GST_stats = NULL;
1132   }
1133   if (NULL != GST_my_private_key)
1134   {
1135     GNUNET_free (GST_my_private_key);
1136     GST_my_private_key = NULL;
1137   }
1138   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
1139                                          &free_neighbour_cb,
1140                                          NULL);
1141   GNUNET_CONTAINER_multipeermap_destroy (neighbours);
1142 }
1143
1144
1145 /**
1146  * Initiate transport service.
1147  *
1148  * @param cls closure
1149  * @param c configuration to use
1150  * @param service the initialized service
1151  */
1152 static void
1153 run (void *cls,
1154      const struct GNUNET_CONFIGURATION_Handle *c,
1155      struct GNUNET_SERVICE_Handle *service)
1156 {
1157   (void) cls;
1158   /* setup globals */
1159   GST_cfg = c;
1160   neighbours = GNUNET_CONTAINER_multipeermap_create (1024,
1161                                                      GNUNET_YES);
1162   GST_my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (GST_cfg);
1163   if (NULL == GST_my_private_key)
1164   {
1165     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1166                 _("Transport service is lacking key configuration settings. Exiting.\n"));
1167     GNUNET_SCHEDULER_shutdown ();
1168     return;
1169   }
1170   GNUNET_CRYPTO_eddsa_key_get_public (GST_my_private_key,
1171                                       &GST_my_identity.public_key);
1172   GNUNET_log(GNUNET_ERROR_TYPE_INFO,
1173              "My identity is `%s'\n",
1174              GNUNET_i2s_full (&GST_my_identity));
1175
1176   GST_stats = GNUNET_STATISTICS_create ("transport",
1177                                         GST_cfg);
1178   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
1179                                  NULL);
1180   /* start subsystems */
1181 }
1182
1183
1184 /**
1185  * Define "main" method using service macro.
1186  */
1187 GNUNET_SERVICE_MAIN
1188 ("transport",
1189  GNUNET_SERVICE_OPTION_NONE,
1190  &run,
1191  &client_connect_cb,
1192  &client_disconnect_cb,
1193  NULL,
1194  /* communication with core */
1195  GNUNET_MQ_hd_fixed_size (client_start,
1196                           GNUNET_MESSAGE_TYPE_TRANSPORT_START,
1197                           struct StartMessage,
1198                           NULL),
1199  GNUNET_MQ_hd_var_size (client_send,
1200                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND,
1201                         struct OutboundMessage,
1202                         NULL),
1203  /* communication with communicators */
1204  GNUNET_MQ_hd_var_size (communicator_available,
1205                         GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
1206                         struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
1207                         NULL),
1208  GNUNET_MQ_hd_var_size (add_address,
1209                         GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
1210                         struct GNUNET_TRANSPORT_AddAddressMessage,
1211                         NULL),
1212  GNUNET_MQ_hd_fixed_size (del_address,
1213                           GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
1214                           struct GNUNET_TRANSPORT_DelAddressMessage,
1215                           NULL),
1216  GNUNET_MQ_hd_var_size (incoming_msg,
1217                         GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
1218                         struct GNUNET_TRANSPORT_IncomingMessage,
1219                         NULL),
1220  GNUNET_MQ_hd_var_size (add_queue_message,
1221                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
1222                         struct GNUNET_TRANSPORT_AddQueueMessage,
1223                         NULL),
1224  GNUNET_MQ_hd_fixed_size (del_queue_message,
1225                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
1226                           struct GNUNET_TRANSPORT_DelQueueMessage,
1227                           NULL),
1228  GNUNET_MQ_hd_fixed_size (send_message_ack,
1229                           GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
1230                           struct GNUNET_TRANSPORT_SendMessageToAck,
1231                           NULL),
1232  /* communication with monitors */
1233  GNUNET_MQ_hd_fixed_size (monitor_start,
1234                           GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_START,
1235                           struct GNUNET_TRANSPORT_MonitorStart,
1236                           NULL),
1237  GNUNET_MQ_handler_end ());
1238
1239
1240 /* end of file gnunet-service-transport.c */