X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=test%2Fecho-fork.c;h=33966da16645043e712c99ff9a22d98966e55048;hb=e40d5bf3a0e030105334046319f377efbf3f06c4;hp=3d0aa2f7ba509e8395aa45530ae5a43165ccc13b;hpb=028bba470982ee4803bcf91bd4bfa1dbb6545c61;p=meshlink diff --git a/test/echo-fork.c b/test/echo-fork.c old mode 100755 new mode 100644 index 3d0aa2f7..33966da1 --- a/test/echo-fork.c +++ b/test/echo-fork.c @@ -1,248 +1,189 @@ #define _GNU_SOURCE + +#ifdef NDEBUG +#undef NDEBUG +#endif + #include #include #include #include +#include +#include +#include +#include + +#ifdef __linux__ +#include +#endif -#include "../src/meshlink.h" +#include "meshlink.h" +#include "utils.h" /* * To run this test case, direct a large file to strd */ -volatile bool bar_reachable = false; -volatile bool bar_responded = false; +static struct sync_flag a_started; +static struct sync_flag a_stopped; +static struct sync_flag b_responded; -void log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, const char *text) { - if(mesh) - fprintf(stderr, "(%s) ", mesh->name); - fprintf(stderr, "[%d] %s\n", level, text); -} +static void a_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { + (void)mesh; + (void)channel; + (void)data; + (void)len; -void status_cb(meshlink_handle_t *mesh, meshlink_node_t *node, bool reachable) { - if(!strcmp(node->name, "bar")) - bar_reachable = reachable; + // One way only. } -void foo_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { - //char tmp[len+1]; - //memset( tmp, 0, sizeof tmp ); - //snprintf( tmp, len+1, "%s", (char*)data ); - //fprintf(stderr, "Foo received from Bar:\n%s\n", tmp); - //fprintf(stderr, "==============================\n"); - //fprintf(stdout, "%s", tmp ); -} +static void b_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { + (void)mesh; + (void)channel; -void bar_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { - // Echo the data back. - char tmp[len+1]; - memset( tmp, 0, sizeof tmp ); - snprintf( tmp, len+1, "%s", (char*)data ); - //fprintf(stderr, "Bar received:\n%s\n", tmp); - //fprintf(stderr, "==============================\n"); - fprintf(stdout, "%s", tmp ); - //meshlink_channel_send(mesh, channel, data, len); -} + if(!len) { + set_sync_flag(&a_stopped, true); + meshlink_channel_close(mesh, channel); + return; + } -bool reject_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) { - return false; + assert(write(1, data, len) == (ssize_t)len); } -bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) { - if(port != 7) +static bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) { + if(port != 7) { return false; - meshlink_set_channel_receive_cb(mesh, channel, bar_receive_cb); - if(data) - bar_receive_cb(mesh, channel, data, len); - return true; -} + } -void poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) { - meshlink_set_channel_poll_cb(mesh, channel, NULL); - bar_responded=true; -} + set_sync_flag(&a_started, true); -int main1(int rfd, int wfd) { - meshlink_set_log_cb(NULL, MESHLINK_DEBUG, log_cb); + meshlink_set_channel_receive_cb(mesh, channel, b_receive_cb); - meshlink_handle_t *mesh1 = meshlink_open("channels_conf.1", "foo", "channels", DEV_CLASS_BACKBONE); - if(!mesh1) { - fprintf(stderr, "Could not initialize configuration for foo\n"); - return 1; + if(data) { + b_receive_cb(mesh, channel, data, len); } - meshlink_add_address(mesh1, "localhost"); + return true; +} - char *data = meshlink_export(mesh1); - if(!data) { - fprintf(stderr, "Foo could not export its configuration\n"); - return 1; - } +static void poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) { + (void)len; - size_t len = strlen(data); - write(wfd, &len, sizeof len); - write(wfd, data, len); - free(data); + meshlink_set_channel_poll_cb(mesh, channel, NULL); + set_sync_flag(&b_responded, true); +} - read(rfd, &len, sizeof len); - char indata[len + 1]; - read(rfd, indata, len); - indata[len] = 0; +static int main1(void) { + close(1); - fprintf(stderr, "Foo exchanged data\n"); + meshlink_handle_t *mesh = meshlink_open("echo-fork_conf.1", "a", "echo-fork", DEV_CLASS_BACKBONE); + assert(mesh); - meshlink_import(mesh1, indata); + assert(meshlink_start(mesh)); - meshlink_set_channel_accept_cb(mesh1, reject_cb); - meshlink_set_node_status_cb(mesh1, status_cb); + // Open a channel. - if(!meshlink_start(mesh1)) { - fprintf(stderr, "Foo could not start\n"); - return 1; - } + meshlink_node_t *b = meshlink_get_node(mesh, "b"); + assert(b); - for(int i = 0; i < 20; i++) { - sleep(1); - if(bar_reachable) - break; - } + meshlink_channel_t *channel = meshlink_channel_open(mesh, b, 7, a_receive_cb, NULL, 0); + assert(channel); - if(!bar_reachable) { - fprintf(stderr, "Bar not reachable for foo after 20 seconds\n"); - return 1; - } - - // Open a channel from foo to bar. - - meshlink_node_t *bar = meshlink_get_node(mesh1, "bar"); - if(!bar) { - fprintf(stderr, "Foo could not find bar\n"); - return 1; - } - - meshlink_channel_t *channel = meshlink_channel_open(mesh1, bar, 7, foo_receive_cb, NULL, 0); - meshlink_set_channel_poll_cb(mesh1, channel, poll_cb); + meshlink_set_channel_poll_cb(mesh, channel, poll_cb); // read and buffer stdin - int BUF_SIZE = 1024*1024; + int BUF_SIZE = 1024 * 1024; char buffer[BUF_SIZE]; - size_t contentSize = 1; - char *content = malloc( sizeof(char) * BUF_SIZE ); - if (!content) { - fprintf(stderr, "Could not allocate buffer\n"); - } - fprintf(stderr, "Foo reading from stdin...\n"); - content[0] = '\0'; - while(fgets(buffer,BUF_SIZE,stdin)) { - char *old = content; - contentSize += strlen(buffer); - content = realloc(content, contentSize); - strcat(content,buffer); - } + assert(wait_sync_flag(&b_responded, 20)); - for(int i = 0; i < 5; i++) { - sleep(1); - if(bar_responded) + do { + ssize_t len = read(0, buffer, BUF_SIZE); + + if(len <= 0) { break; - } + } - if(!bar_responded) { - fprintf(stderr, "Bar did not respond to foo's channel message\n"); - return 1; - } + char *p = buffer; - //fprintf(stderr, "Foo sending:\n%s", content); - //fprintf(stderr, "==============================\n"); - - size_t total = 0; - while ( total != contentSize ) - { - size_t to_send = contentSize - total > 2000 ? 2000 : contentSize - total; - ssize_t tmp = meshlink_channel_send(mesh1, channel, content + total, to_send); - if (tmp >= 0) - { - total += tmp; - if (tmp != to_send) - sleep(1); - } else { - fprintf(stderr, "Sending message failed\n"); - return 1; - } - } - - fprintf(stderr, "Foo finished sending\n"); + while(len > 0) { + ssize_t sent = meshlink_channel_send(mesh, channel, p, len); - sleep(30); + if(sent < 0) { + fprintf(stderr, "Sending message failed\n"); + return 1; + } - free(content); + if(!sent) { + usleep(100000); + } else { + len -= sent; + p += sent; + } + } + } while(true); - meshlink_channel_close(mesh1, channel); + meshlink_channel_close(mesh, channel); // Clean up. - meshlink_close(mesh1); + meshlink_close(mesh); return 0; } -int main2(int rfd, int wfd) { - sleep(1); - - meshlink_set_log_cb(NULL, MESHLINK_DEBUG, log_cb); +static int main2(void) { +#ifdef __linux__ + prctl(PR_SET_PDEATHSIG, SIGTERM); +#endif - meshlink_handle_t *mesh2 = meshlink_open("channels_conf.2", "bar", "channels", DEV_CLASS_BACKBONE); - if(!mesh2) { - fprintf(stderr, "Could not initialize configuration for bar\n"); - return 1; - } - - char *data = meshlink_export(mesh2); - if(!data) { - fprintf(stderr, "Bar could not export its configuration\n"); - return 1; - } + close(0); - size_t len = strlen(data); - if(write(wfd, &len, sizeof len) <= 0) abort(); - if(write(wfd, data, len) <= 0) abort(); - free(data); + // Start mesh and wait for incoming channels. - read(rfd, &len, sizeof len); - char indata[len + 1]; - read(rfd, indata, len); - indata[len] = 0; + meshlink_handle_t *mesh = meshlink_open("echo-fork_conf.2", "b", "echo-fork", DEV_CLASS_BACKBONE); + assert(mesh); - fprintf(stderr, "Bar exchanged data\n"); + meshlink_set_channel_accept_cb(mesh, accept_cb); - meshlink_import(mesh2, indata); + assert(meshlink_start(mesh)); - meshlink_set_channel_accept_cb(mesh2, accept_cb); + // Let it run until a disappears. - if(!meshlink_start(mesh2)) { - fprintf(stderr, "Bar could not start\n"); - return 1; - } - - sleep(30); + assert(wait_sync_flag(&a_started, 20)); + assert(wait_sync_flag(&a_stopped, 1000000)); // Clean up. - meshlink_close(mesh2); + meshlink_close(mesh); return 0; } -int main(int argc, char *argv[]) { - int fda[2], fdb[2], result; +int main(void) { + init_sync_flag(&a_started); + init_sync_flag(&a_stopped); + init_sync_flag(&b_responded); + + meshlink_set_log_cb(NULL, MESHLINK_WARNING, log_cb); + + // Initialize and exchange configuration. + + meshlink_handle_t *mesh_a, *mesh_b; + + open_meshlink_pair(&mesh_a, &mesh_b, "echo-fork"); + close_meshlink_pair(mesh_a, mesh_b); + + if(!fork()) { + return main2(); + } - pipe2(fda, 0); - pipe2(fdb, 0); + assert(main1() == 0); - if(fork()) - return main1(fda[0], fdb[1]); - else - return main2(fdb[0], fda[1]); + int wstatus = 0; + assert(wait(&wstatus) != -1); + assert(WIFEXITED(wstatus)); + assert(WEXITSTATUS(wstatus) == 0); }