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

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