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