git.meshlink.io Git - meshlink/blob - test/echo-fork.c

   1 #define _GNU_SOURCE
   2
   3 #ifdef NDEBUG
   4 #undef NDEBUG
   5 #endif
   6
   7 #include <stdio.h>
   8 #include <unistd.h>
   9 #include <stdlib.h>
  10 #include <string.h>
  11 #include <limits.h>
  12 #include <assert.h>
  13 #include <sys/types.h>
  14 #include <sys/wait.h>
  15
  16 #ifdef __linux__
  17 #include <sys/prctl.h>
  18 #endif
  19
  20 #include "meshlink.h"
  21 #include "utils.h"
  22
  23 /*
  24  * To run this test case, direct a large file to strd
  25  */
  26
  27 static struct sync_flag a_started;
  28 static struct sync_flag a_stopped;
  29 static struct sync_flag b_responded;
  30
  31 static void a_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  32         (void)mesh;
  33         (void)channel;
  34         (void)data;
  35         (void)len;
  36
  37         // One way only.
  38 }
  39
  40 static void b_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  41         (void)mesh;
  42         (void)channel;
  43
  44         if(!len) {
  45                 set_sync_flag(&a_stopped, true);
  46                 meshlink_channel_close(mesh, channel);
  47                 return;
  48         }
  49
  50         assert(write(1, data, len) == (ssize_t)len);
  51 }
  52
  53 static bool reject_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
  54         (void)mesh;
  55         (void)channel;
  56         (void)port;
  57         (void)data;
  58         (void)len;
  59
  60         return false;
  61 }
  62
  63 static bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
  64         if(port != 7) {
  65                 return false;
  66         }
  67
  68         set_sync_flag(&a_started, true);
  69
  70         meshlink_set_channel_receive_cb(mesh, channel, b_receive_cb);
  71
  72         if(data) {
  73                 b_receive_cb(mesh, channel, data, len);
  74         }
  75
  76         return true;
  77 }
  78
  79 static void poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) {
  80         (void)len;
  81
  82         meshlink_set_channel_poll_cb(mesh, channel, NULL);
  83         set_sync_flag(&b_responded, true);
  84 }
  85
  86 static int main1(void) {
  87         close(1);
  88
  89         meshlink_handle_t *mesh = meshlink_open("echo-fork_conf.1", "a", "echo-fork", DEV_CLASS_BACKBONE);
  90         assert(mesh);
  91
  92         meshlink_set_channel_accept_cb(mesh, reject_cb);
  93
  94         assert(meshlink_start(mesh));
  95
  96         // Open a channel.
  97
  98         meshlink_node_t *b = meshlink_get_node(mesh, "b");
  99         assert(b);
 100
 101         meshlink_channel_t *channel = meshlink_channel_open(mesh, b, 7, a_receive_cb, NULL, 0);
 102         assert(channel);
 103
 104         meshlink_set_channel_poll_cb(mesh, channel, poll_cb);
 105
 106         // read and buffer stdin
 107         int BUF_SIZE = 1024 * 1024;
 108         char buffer[BUF_SIZE];
 109
 110         assert(wait_sync_flag(&b_responded, 20));
 111
 112         do {
 113                 ssize_t len = read(0, buffer, BUF_SIZE);
 114
 115                 if(len <= 0) {
 116                         break;
 117                 }
 118
 119                 char *p = buffer;
 120
 121                 while(len > 0) {
 122                         ssize_t sent = meshlink_channel_send(mesh, channel, p, len);
 123
 124                         if(sent < 0) {
 125                                 fprintf(stderr, "Sending message failed\n");
 126                                 return 1;
 127                         }
 128
 129                         if(!sent) {
 130                                 usleep(100000);
 131                         } else {
 132                                 len -= sent;
 133                                 p += sent;
 134                         }
 135                 }
 136         } while(true);
 137
 138         meshlink_channel_close(mesh, channel);
 139
 140         // Clean up.
 141
 142         meshlink_close(mesh);
 143
 144         return 0;
 145 }
 146
 147
 148 static int main2(void) {
 149 #ifdef __linux__
 150         prctl(PR_SET_PDEATHSIG, SIGTERM);
 151 #endif
 152
 153         close(0);
 154
 155         // Start mesh and wait for incoming channels.
 156
 157         meshlink_handle_t *mesh = meshlink_open("echo-fork_conf.2", "b", "echo-fork", DEV_CLASS_BACKBONE);
 158         assert(mesh);
 159
 160         meshlink_set_channel_accept_cb(mesh, accept_cb);
 161
 162         assert(meshlink_start(mesh));
 163
 164         // Let it run until a disappears.
 165
 166         assert(wait_sync_flag(&a_started, 20));
 167         assert(wait_sync_flag(&a_stopped, 1000000));
 168
 169         // Clean up.
 170
 171         meshlink_close(mesh);
 172
 173         return 0;
 174 }
 175
 176
 177 int main(void) {
 178         init_sync_flag(&a_started);
 179         init_sync_flag(&a_stopped);
 180         init_sync_flag(&b_responded);
 181
 182         meshlink_set_log_cb(NULL, MESHLINK_WARNING, log_cb);
 183
 184         // Initialize and exchange configuration.
 185
 186         meshlink_handle_t *mesh_a, *mesh_b;
 187
 188         open_meshlink_pair(&mesh_a, &mesh_b, "echo-fork");
 189         close_meshlink_pair(mesh_a, mesh_b);
 190
 191         if(!fork()) {
 192                 return main2();
 193         }
 194
 195         assert(main1() == 0);
 196
 197         int wstatus = 0;
 198         assert(wait(&wstatus) != -1);
 199         assert(WIFEXITED(wstatus));
 200         assert(WEXITSTATUS(wstatus) == 0);
 201 }