src/transport/test_communicator_basic.c

   1 /*
   2     This file is part of GNUnet.
   3     Copyright (C) 2019 GNUnet e.V.
   4
   5     GNUnet is free software: you can redistribute it and/or modify it
   6     under the terms of the GNU Affero General Public License as published
   7     by the Free Software Foundation, either version 3 of the License,
   8     or (at your option) any later version.
   9
  10     GNUnet is distributed in the hope that it will be useful, but
  11     WITHOUT ANY WARRANTY; without even the implied warranty of
  12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13     Affero General Public License for more details.
  14
  15     You should have received a copy of the GNU Affero General Public License
  16     along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18     SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20
  21 /**
  22 * @file transport/test_communicator_basic.c
  23 * @brief test the communicators
  24 * @author Julius Bünger
  25 * @author Martin Schanzenbach
  26 */
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "transport-testing2.h"
  30 #include "gnunet_ats_transport_service.h"
  31 #include "gnunet_signatures.h"
  32 #include "gnunet_testing_lib.h"
  33 #include "transport.h"
  34
  35 #include <inttypes.h>
  36
  37
  38 #define LOG(kind, ...) GNUNET_log_from (kind, \
  39                                         "test_transport_communicator", \
  40                                         __VA_ARGS__)
  41
  42 #define NUM_PEERS 2
  43
  44 static struct GNUNET_SCHEDULER_Task *to_task;
  45
  46 static struct GNUNET_SCHEDULER_Task *active_task;
  47
  48 static int queue_est = GNUNET_NO;
  49
  50 static struct GNUNET_PeerIdentity peer_id[NUM_PEERS];
  51
  52 static char *communicator_binary;
  53
  54 static struct
  55 GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_hs[NUM_PEERS];
  56
  57 static struct GNUNET_CONFIGURATION_Handle *cfg_peers[NUM_PEERS];
  58
  59 static char *cfg_peers_name[NUM_PEERS];
  60
  61 static int ret;
  62
  63 static struct GNUNET_TIME_Absolute start_short;
  64
  65 static struct GNUNET_TIME_Absolute start_long;
  66
  67 static struct GNUNET_TIME_Absolute timeout;
  68
  69 static struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *my_tc;
  70
  71 #define SHORT_MESSAGE_SIZE 128
  72
  73 #define LONG_MESSAGE_SIZE 32000
  74
  75 #define BURST_PACKETS 5000
  76
  77 #define TOTAL_ITERATIONS 5
  78
  79 static unsigned int iterations_left = TOTAL_ITERATIONS;
  80
  81 #define SHORT_BURST_WINDOW \
  82   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,2)
  83
  84 #define LONG_BURST_WINDOW \
  85   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,2)
  86
  87 enum TestPhase
  88 {
  89   TP_BURST_SHORT,
  90   TP_BURST_LONG,
  91   TP_SIZE_CHECK
  92 };
  93
  94
  95 static size_t num_sent = 0;
  96
  97 static uint32_t ack = 0;
  98
  99 static enum TestPhase phase;
 100
 101 static size_t num_received = 0;
 102
 103 static uint64_t avg_latency = 0;
 104
 105 static struct GNUNET_TIME_Relative duration;
 106
 107
 108 static void
 109 communicator_available_cb (void *cls,
 110                            struct
 111                            GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle
 112                            *tc_h,
 113                            enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc,
 114                            char *address_prefix)
 115 {
 116   LOG (GNUNET_ERROR_TYPE_INFO,
 117        "Communicator available. (cc: %u, prefix: %s)\n",
 118        cc,
 119        address_prefix);
 120 }
 121
 122
 123 static void
 124 add_address_cb (void *cls,
 125                 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *
 126                 tc_h,
 127                 const char *address,
 128                 struct GNUNET_TIME_Relative expiration,
 129                 uint32_t aid,
 130                 enum GNUNET_NetworkType nt)
 131 {
 132   LOG (GNUNET_ERROR_TYPE_DEBUG,
 133        "New address. (addr: %s, expir: %" PRIu32 ", ID: %" PRIu32 ", nt: %u\n",
 134        address,
 135        expiration.rel_value_us,
 136        aid,
 137        nt);
 138   // addresses[1] = GNUNET_strdup (address);
 139   if ((0 == strcmp ((char*) cls, cfg_peers_name[NUM_PEERS - 1])) &&
 140       (GNUNET_NO == queue_est))
 141   {
 142     queue_est = GNUNET_YES;
 143     GNUNET_TRANSPORT_TESTING_transport_communicator_open_queue (tc_hs[0],
 144                                                                 &peer_id[
 145                                                                   NUM_PEERS
 146                                                                   - 1],
 147                                                                 address);
 148   }
 149 }
 150
 151
 152 /**
 153  * @brief Callback that informs whether the requested queue will be
 154  * established
 155  *
 156  * Implements #GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback.
 157  *
 158  * @param cls Closure - unused
 159  * @param tc_h Communicator handle - unused
 160  * @param will_try #GNUNET_YES if queue will be established
 161  *                #GNUNET_NO if queue will not be established (bogous address)
 162  */
 163 static void
 164 queue_create_reply_cb (void *cls,
 165                        struct
 166                        GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *
 167                        tc_h,
 168                        int will_try)
 169 {
 170   if (GNUNET_YES == will_try)
 171     LOG (GNUNET_ERROR_TYPE_DEBUG,
 172          "Queue will be established!\n");
 173   else
 174     LOG (GNUNET_ERROR_TYPE_WARNING,
 175          "Queue won't be established (bougus address?)!\n");
 176 }
 177
 178
 179 static char*
 180 make_payload (size_t payload_size)
 181 {
 182   struct GNUNET_TIME_Absolute ts;
 183   struct GNUNET_TIME_AbsoluteNBO ts_n;
 184   char *payload = GNUNET_malloc (payload_size);
 185
 186   GNUNET_assert (payload_size >= 8); // So that out timestamp fits
 187   ts = GNUNET_TIME_absolute_get ();
 188   ts_n = GNUNET_TIME_absolute_hton (ts);
 189   memset (payload, 0, payload_size);
 190   memcpy (payload, &ts_n, sizeof (struct GNUNET_TIME_AbsoluteNBO));
 191   return payload;
 192 }
 193
 194
 195 static void
 196 latency_timeout (void *cls)
 197 {
 198   to_task = NULL;
 199   if (GNUNET_TIME_absolute_get_remaining (timeout).rel_value_us > 0)
 200   {
 201     to_task = GNUNET_SCHEDULER_add_at (timeout,
 202                                        &latency_timeout,
 203                                        NULL);
 204     return;
 205   }
 206
 207   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 208               "Latency too high. Test failed. (Phase: %d. Sent: %lu, Received: %lu)\n",
 209               phase, num_sent, num_received);
 210   ret = 2;
 211   GNUNET_SCHEDULER_shutdown ();
 212 }
 213
 214
 215 static void
 216 size_test (void *cls)
 217 {
 218   char *payload;
 219
 220   active_task = NULL;
 221   GNUNET_assert (TP_SIZE_CHECK == phase);
 222   if (ack >= 64000)
 223     return; /* Leave some room for our protocol, so not 2^16 exactly */
 224   payload = make_payload (ack);
 225   GNUNET_TRANSPORT_TESTING_transport_communicator_send (my_tc,
 226                                                         payload,
 227                                                         ack);
 228   GNUNET_free (payload);
 229   ack += 5;
 230   num_sent++;
 231   timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_SECONDS);
 232   if (ack < 64000)
 233     active_task = GNUNET_SCHEDULER_add_now (&size_test,
 234                                             NULL);
 235 }
 236
 237
 238 static void
 239 long_test (void *cls)
 240 {
 241   char *payload;
 242
 243   active_task = NULL;
 244   payload = make_payload (LONG_MESSAGE_SIZE);
 245   GNUNET_TRANSPORT_TESTING_transport_communicator_send (my_tc,
 246                                                         payload,
 247                                                         LONG_MESSAGE_SIZE);
 248   num_sent++;
 249   GNUNET_free (payload);
 250   timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_SECONDS);
 251   if (num_sent == BURST_PACKETS)
 252     return;
 253   active_task = GNUNET_SCHEDULER_add_now (&long_test,
 254                                           NULL);
 255 }
 256
 257
 258 static void
 259 short_test (void *cls)
 260 {
 261   char *payload;
 262
 263   active_task = NULL;
 264   payload = make_payload (SHORT_MESSAGE_SIZE);
 265   GNUNET_TRANSPORT_TESTING_transport_communicator_send (my_tc,
 266                                                         payload,
 267                                                         SHORT_MESSAGE_SIZE);
 268   num_sent++;
 269   GNUNET_free (payload);
 270   timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_SECONDS);
 271   if (num_sent >= BURST_PACKETS)
 272     return;
 273   active_task = GNUNET_SCHEDULER_add_now (&short_test,
 274                                           NULL);
 275 }
 276
 277
 278 /**
 279  * @brief Handle opening of queue
 280  *
 281  * Issues sending of test data
 282  *
 283  * Implements #GNUNET_TRANSPORT_TESTING_AddQueueCallback
 284  *
 285  * @param cls Closure
 286  * @param tc_h Communicator handle
 287  * @param tc_queue Handle to newly opened queue
 288  */
 289 static void
 290 add_queue_cb (void *cls,
 291               struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 292               struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *
 293               tc_queue)
 294 {
 295   if (0 != strcmp ((char*) cls, cfg_peers_name[0]))
 296     return; // TODO?
 297   LOG (GNUNET_ERROR_TYPE_DEBUG,
 298        "Queue established, starting test...\n");
 299   start_short = GNUNET_TIME_absolute_get ();
 300   my_tc = tc_queue;
 301   phase = TP_BURST_SHORT;
 302   timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_SECONDS);
 303   GNUNET_assert (NULL == to_task);
 304   to_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 305                                           &latency_timeout,
 306                                           NULL);
 307   GNUNET_assert (NULL == active_task);
 308   active_task = GNUNET_SCHEDULER_add_now (&short_test,
 309                                           NULL);
 310 }
 311
 312
 313 static void
 314 update_avg_latency (const char*payload)
 315 {
 316   struct GNUNET_TIME_AbsoluteNBO *ts_n;
 317   struct GNUNET_TIME_Absolute ts;
 318   struct GNUNET_TIME_Relative latency;
 319
 320   ts_n = (struct GNUNET_TIME_AbsoluteNBO *) payload;
 321   ts = GNUNET_TIME_absolute_ntoh (*ts_n);
 322   latency = GNUNET_TIME_absolute_get_duration (ts);
 323   if (1 >= num_received)
 324     avg_latency = latency.rel_value_us;
 325   else
 326     avg_latency = ((avg_latency * (num_received - 1)) + latency.rel_value_us)
 327                   / num_received;
 328
 329 }
 330
 331
 332 /**
 333  * @brief Handle an incoming message
 334  *
 335  * Implements #GNUNET_TRANSPORT_TESTING_IncomingMessageCallback
 336
 337  * @param cls Closure
 338  * @param tc_h Handle to the receiving communicator
 339  * @param msg Received message
 340  */
 341 static void
 342 incoming_message_cb (void *cls,
 343                      struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle
 344                      *tc_h,
 345                      const char*payload,
 346                      size_t payload_len)
 347 {
 348   if (0 != strcmp ((char*) cls, cfg_peers_name[NUM_PEERS - 1]))
 349   {
 350     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 351                 "unexpected receiver...\n");
 352     return;
 353   }
 354   /* Reset timeout */
 355   timeout = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_SECONDS);
 356   switch (phase)
 357   {
 358   case TP_BURST_SHORT:
 359     {
 360       GNUNET_assert (SHORT_MESSAGE_SIZE == payload_len);
 361       num_received++;
 362       duration = GNUNET_TIME_absolute_get_duration (start_short);
 363       update_avg_latency (payload);
 364       if (num_received == BURST_PACKETS)
 365       {
 366         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 367                     "Short size packet test done.\n");
 368         char *goodput = GNUNET_STRINGS_byte_size_fancy ((SHORT_MESSAGE_SIZE
 369                                                          * num_received)
 370                                                         / (GNUNET_MAX (1,
 371                                                                        duration.
 372                                                                        rel_value_us
 373                                                                        / (1000
 374                                                                           * 1000))));
 375         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 376                     "%lu/%lu packets in %llu us (%s/s) -- avg latency: %llu us\n",
 377                     (unsigned long) num_received,
 378                     (unsigned long) num_sent,
 379                     (unsigned long long) duration.rel_value_us,
 380                     goodput,
 381                     (unsigned long long) avg_latency);
 382         GNUNET_free (goodput);
 383         start_long = GNUNET_TIME_absolute_get ();
 384         phase = TP_BURST_LONG;
 385         num_sent = 0;
 386         avg_latency = 0;
 387         num_received = 0;
 388         active_task = GNUNET_SCHEDULER_add_now (&long_test,
 389                                                 NULL);
 390       }
 391       break;
 392     }
 393   case TP_BURST_LONG:
 394     {
 395       if (LONG_MESSAGE_SIZE != payload_len)
 396       {
 397         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 398                     "Ignoring packet with wrong length\n");
 399         return; // Ignore
 400       }
 401       num_received++;
 402       duration = GNUNET_TIME_absolute_get_duration (start_long);
 403       update_avg_latency (payload);
 404       if (num_received == BURST_PACKETS)
 405       {
 406         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 407                     "Long size packet test done.\n");
 408         char *goodput = GNUNET_STRINGS_byte_size_fancy ((LONG_MESSAGE_SIZE
 409                                                          * num_received)
 410                                                         / (GNUNET_MAX (1,
 411                                                                        duration.
 412                                                                        rel_value_us
 413                                                                        / (1000
 414                                                                           * 1000))));
 415
 416         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 417                     "%lu/%lu packets in %llu us (%s/s) -- avg latency: %llu us\n",
 418                     (unsigned long) num_received,
 419                     (unsigned long) num_sent,
 420                     (unsigned long long) duration.rel_value_us,
 421                     goodput,
 422                     (unsigned long long) avg_latency);
 423         GNUNET_free (goodput);
 424         ack = 10;
 425         phase = TP_SIZE_CHECK;
 426         num_received = 0;
 427         num_sent = 0;
 428         avg_latency = 0;
 429         active_task = GNUNET_SCHEDULER_add_now (&size_test,
 430                                                 NULL);
 431       }
 432       break;
 433     }
 434   case TP_SIZE_CHECK:
 435     {
 436       num_received++;
 437       update_avg_latency (payload);
 438       if (num_received >= (64000 - 10) / 5)
 439       {
 440         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 441                     "Size packet test done.\n");
 442         GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE,
 443                     "%lu/%lu packets -- avg latency: %llu us\n",
 444                     (unsigned long) num_received,
 445                     (unsigned long) num_sent,
 446                     (unsigned long long) avg_latency);
 447         num_received = 0;
 448         num_sent = 0;
 449         avg_latency = 0;
 450         iterations_left--;
 451         if (0 != iterations_left)
 452         {
 453           start_short = GNUNET_TIME_absolute_get ();
 454           phase = TP_BURST_SHORT;
 455           active_task = GNUNET_SCHEDULER_add_now (&short_test,
 456                                                   NULL);
 457           break;
 458         }
 459         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 460                     "Finished\n");
 461         GNUNET_SCHEDULER_shutdown ();
 462       }
 463       break;
 464     }
 465   }
 466 }
 467
 468
 469 static void
 470 do_shutdown (void *cls)
 471 {
 472   if (NULL != to_task)
 473   {
 474     GNUNET_SCHEDULER_cancel (to_task);
 475     to_task = NULL;
 476   }
 477   if (NULL != active_task)
 478   {
 479     GNUNET_SCHEDULER_cancel (active_task);
 480     active_task = NULL;
 481   }
 482 }
 483
 484
 485 /**
 486  * @brief Main function called by the scheduler
 487  *
 488  * @param cls Closure - Handle to configuration
 489  */
 490 static void
 491 run (void *cls)
 492 {
 493   ret = 0;
 494   num_received = 0;
 495   num_sent = 0;
 496   for (unsigned int i = 0; i < NUM_PEERS; i++)
 497   {
 498     tc_hs[i] = GNUNET_TRANSPORT_TESTING_transport_communicator_service_start (
 499       "transport",
 500       communicator_binary,
 501       cfg_peers_name[i],
 502       &communicator_available_cb,
 503       &add_address_cb,
 504       &queue_create_reply_cb,
 505       &add_queue_cb,
 506       &incoming_message_cb,
 507       cfg_peers_name[i]);   /* cls */
 508   }
 509   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
 510                                  NULL);
 511 }
 512
 513
 514 int
 515 main (int argc,
 516       char *const *argv)
 517 {
 518   struct GNUNET_CRYPTO_EddsaPrivateKey *private_key;
 519   char *communicator_name;
 520   char *cfg_peer;
 521
 522   ret = 1;
 523   communicator_name = GNUNET_TESTING_get_testname_from_underscore (argv[0]);
 524   GNUNET_asprintf (&communicator_binary,
 525                    "gnunet-communicator-%s",
 526                    communicator_name);
 527   if (GNUNET_OK !=
 528       GNUNET_log_setup ("test_communicator_basic",
 529                         "DEBUG",
 530                         NULL))
 531   {
 532     fprintf (stderr, "Unable to setup log\n");
 533     GNUNET_break (0);
 534     return 2;
 535   }
 536   for (unsigned int i = 0; i < NUM_PEERS; i++)
 537   {
 538     GNUNET_asprintf ((&cfg_peer),
 539                      "test_communicator_%s_peer%u.conf",
 540                      communicator_name, i + 1);
 541     cfg_peers_name[i] = cfg_peer;
 542     cfg_peers[i] = GNUNET_CONFIGURATION_create ();
 543     if (GNUNET_YES ==
 544         GNUNET_DISK_file_test (cfg_peers_name[i]))
 545     {
 546       if (GNUNET_SYSERR ==
 547           GNUNET_CONFIGURATION_load (cfg_peers[i],
 548                                      cfg_peers_name[i]))
 549       {
 550         fprintf (stderr,
 551                  "Malformed configuration file `%s', exiting ...\n",
 552                  cfg_peers_name[i]);
 553         return 1;
 554       }
 555     }
 556     else
 557     {
 558       if (GNUNET_SYSERR ==
 559           GNUNET_CONFIGURATION_load (cfg_peers[i],
 560                                      NULL))
 561       {
 562         fprintf (stderr,
 563                  "Configuration file %s does not exist, exiting ...\n",
 564                  cfg_peers_name[i]);
 565         return 1;
 566       }
 567     }
 568     private_key =
 569       GNUNET_CRYPTO_eddsa_key_create_from_configuration (cfg_peers[i]);
 570     if (NULL == private_key)
 571     {
 572       LOG (GNUNET_ERROR_TYPE_ERROR,
 573            "Unable to get peer ID\n");
 574       return 1;
 575     }
 576     GNUNET_CRYPTO_eddsa_key_get_public (private_key,
 577                                         &peer_id[i].public_key);
 578     GNUNET_free (private_key);
 579     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 580                 "Identity of peer %u is %s\n",
 581                 i,
 582                 GNUNET_i2s_full (&peer_id[i]));
 583   }
 584   GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE, "Starting test...\n");
 585   GNUNET_SCHEDULER_run (&run,
 586                         NULL);
 587   return ret;
 588 }