git.meshlink.io Git - meshlink/blob - test/channels-mixed.c

   1 #ifdef NDEBUG
   2 #undef NDEBUG
   3 #endif
   4
   5 #include <assert.h>
   6 #include <stdio.h>
   7 #include <unistd.h>
   8 #include <stdlib.h>
   9 #include <string.h>
  10 #include <sys/time.h>
  11 #include <time.h>
  12
  13 #include "../src/meshlink.h"
  14 #include "utils.h"
  15
  16 #define SMALL_SIZE 512
  17 #define SMALL_COUNT 2500
  18 #define LARGE_SIZE 131072
  19 #define NCHANNELS 10
  20 #define PORT 123
  21
  22 static struct channel {
  23         meshlink_channel_t *channel;
  24         bool udp;
  25         struct sync_flag open_flag;
  26         struct sync_flag close_flag;
  27 } channels[NCHANNELS];
  28
  29 static size_t received[NCHANNELS];
  30
  31 static struct sync_flag pmtu_flag;
  32
  33 static void a_pmtu_cb(meshlink_handle_t *mesh, meshlink_node_t *node, uint16_t pmtu) {
  34         assert(mesh);
  35
  36         if(pmtu >= SMALL_SIZE && !strcmp(node->name, "b")) {
  37                 set_sync_flag(&pmtu_flag, true);
  38         }
  39 }
  40
  41 static void a_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  42         struct channel *info = channel->priv;
  43         assert(info);
  44
  45         if(!data && !len) {
  46                 set_sync_flag(&info->close_flag, true);
  47                 meshlink_channel_close(mesh, channel);
  48                 return;
  49         }
  50 }
  51
  52 static void aio_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len, void *priv) {
  53         (void)data;
  54         (void)len;
  55         (void)priv;
  56
  57         meshlink_channel_shutdown(mesh, channel, SHUT_WR);
  58 }
  59
  60 static void b_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  61         if(!data && !len) {
  62                 meshlink_channel_close(mesh, channel);
  63                 return;
  64         }
  65
  66         assert(data && len);
  67         size_t *rx = channel->priv;
  68         *rx += len;
  69 }
  70
  71 static bool b_accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
  72         (void)data;
  73         (void)len;
  74
  75         assert(!data);
  76         assert(!len);
  77         assert(port == PORT);
  78
  79         static int i;
  80         channel->priv = &received[i++];
  81
  82         meshlink_set_channel_receive_cb(mesh, channel, b_receive_cb);
  83         return true;
  84 }
  85
  86 static void a_poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) {
  87         assert(len);
  88
  89         meshlink_set_channel_poll_cb(mesh, channel, NULL);
  90
  91         struct channel *info = channel->priv;
  92         assert(info);
  93
  94         set_sync_flag(&info->open_flag, true);
  95 }
  96
  97 int main(void) {
  98         meshlink_set_log_cb(NULL, MESHLINK_WARNING, log_cb);
  99
 100         meshlink_handle_t *a, *b;
 101         open_meshlink_pair(&a, &b, "channels-mixed");
 102
 103         meshlink_set_channel_accept_cb(b, b_accept_cb);
 104         meshlink_set_node_pmtu_cb(a, a_pmtu_cb);
 105         start_meshlink_pair(a, b);
 106
 107         // Create a number of TCP and UDP channels
 108
 109         meshlink_node_t *nb = meshlink_get_node(a, "b");
 110
 111         for(int i = 0; i < NCHANNELS; i++) {
 112                 channels[i].udp = i % 2 == 1;
 113                 init_sync_flag(&channels[i].open_flag);
 114                 init_sync_flag(&channels[i].close_flag);
 115                 channels[i].channel = meshlink_channel_open_ex(a, nb, PORT, a_receive_cb, &channels[i], 0, channels[i].udp ? MESHLINK_CHANNEL_UDP : MESHLINK_CHANNEL_TCP);
 116                 meshlink_set_channel_poll_cb(a, channels[i].channel, a_poll_cb);
 117                 assert(channels[i].channel);
 118         }
 119
 120         // Wait for all channels to connect
 121
 122         for(int i = 0; i < NCHANNELS; i++) {
 123                 assert(wait_sync_flag(&channels[i].open_flag, 10));
 124         }
 125
 126         // Wait for PMTU to finish
 127
 128         assert(wait_sync_flag(&pmtu_flag, 10));
 129
 130         // Send data on all channels
 131
 132         size_t size = SMALL_SIZE * SMALL_COUNT;
 133         char *data = malloc(size);
 134         memset(data, 'U', size);
 135
 136         for(int i = 0; i < NCHANNELS; i++) {
 137                 assert(meshlink_channel_aio_send(a, channels[i].channel, data, size, aio_cb, &channels[i]));
 138         }
 139
 140         // Wait for the other end to close the channels
 141
 142         for(int i = 0; i < NCHANNELS; i++) {
 143                 assert(wait_sync_flag(&channels[i].close_flag, 10));
 144         }
 145
 146         // Check that most of the data has been transmitted
 147
 148         for(int i = 0; i < NCHANNELS; i++) {
 149                 fprintf(stderr, "%d received %zu\n", i, received[i]);
 150         }
 151
 152         int received_all = 0;
 153
 154         for(int i = 0; i < NCHANNELS; i++) {
 155                 assert(received[i] >= size / 2);
 156                 assert(received[i] <= size);
 157
 158                 if(received[i] == size) {
 159                         received_all++;
 160                 }
 161         }
 162
 163         assert(received_all >= NCHANNELS / 2);
 164
 165         // Done
 166
 167         close_meshlink_pair(a, b);
 168
 169         return 0;
 170 }