]> git.meshlink.io Git - meshlink-tiny/blob - test/channels-aio-fd.c
Rename the library and public header to meshlink-tiny.
[meshlink-tiny] / test / channels-aio-fd.c
1 #ifdef NDEBUG
2 #undef NDEBUG
3 #endif
4
5 #define _POSIX_C_SOURCE 200809L
6
7 #include <assert.h>
8 #include <stdio.h>
9 #include <unistd.h>
10 #include <stdlib.h>
11 #include <string.h>
12 #include <time.h>
13 #include <limits.h>
14
15 #include "meshlink-tiny.h"
16 #include "utils.h"
17
18 static const size_t size = 1024 * 1024; // size of data to transfer
19 static const size_t nchannels = 4; // number of simultaneous channels
20
21 struct aio_info {
22         int callbacks;
23         size_t size;
24         struct timespec ts;
25         struct sync_flag flag;
26 };
27
28 struct channel_info {
29         FILE *file;
30         struct aio_info aio_infos[2];
31 };
32
33 static void aio_fd_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, int fd, size_t len, void *priv) {
34         (void)mesh;
35         (void)channel;
36         (void)fd;
37         (void)len;
38
39         struct aio_info *info = priv;
40         clock_gettime(CLOCK_MONOTONIC, &info->ts);
41         info->callbacks++;
42         info->size += len;
43         set_sync_flag(&info->flag, true);
44 }
45
46 static bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
47         assert(port && port <= nchannels);
48         assert(!data);
49         assert(!len);
50
51         struct channel_info *infos = mesh->priv;
52         struct channel_info *info = &infos[port - 1];
53
54         assert(meshlink_channel_aio_fd_receive(mesh, channel, fileno(info->file), size / 4, aio_fd_cb, &info->aio_infos[0]));
55         assert(meshlink_channel_aio_fd_receive(mesh, channel, fileno(info->file), size - size / 4, aio_fd_cb, &info->aio_infos[1]));
56
57         return true;
58 }
59
60 int main(void) {
61         meshlink_set_log_cb(NULL, MESHLINK_WARNING, log_cb);
62
63         // Prepare file
64
65         char *outdata = malloc(size);
66         assert(outdata);
67
68         for(size_t i = 0; i < size; i++) {
69                 // Human readable output
70                 outdata[i] = i % 96 ? i % 96 + 32 : '\n';
71         }
72
73         FILE *file = fopen("channels_aio_fd.in", "w");
74         assert(file);
75         assert(fwrite(outdata, size, 1, file) == 1);
76         assert(fclose(file) == 0);
77
78         struct channel_info in_infos[nchannels];
79         struct channel_info out_infos[nchannels];
80
81         memset(in_infos, 0, sizeof(in_infos));
82         memset(out_infos, 0, sizeof(out_infos));
83
84         for(size_t i = 0; i < nchannels; i++) {
85                 init_sync_flag(&in_infos[i].aio_infos[0].flag);
86                 init_sync_flag(&in_infos[i].aio_infos[1].flag);
87                 init_sync_flag(&out_infos[i].aio_infos[0].flag);
88                 init_sync_flag(&out_infos[i].aio_infos[1].flag);
89
90                 char filename[PATH_MAX];
91                 snprintf(filename, sizeof(filename), "channels_aio_fd.out%d", (int)i);
92                 in_infos[i].file = fopen(filename, "w");
93                 assert(in_infos[i].file);
94                 out_infos[i].file = fopen("channels_aio_fd.in", "r");
95                 assert(out_infos[i].file);
96         }
97
98         // Open two new meshlink instance.
99
100         meshlink_handle_t *mesh_a, *mesh_b;
101         open_meshlink_pair(&mesh_a, &mesh_b, "channels_aio_fd");
102
103         mesh_b->priv = in_infos;
104
105         meshlink_enable_discovery(mesh_a, false);
106         meshlink_enable_discovery(mesh_b, false);
107
108         // Set the callbacks.
109
110         meshlink_set_channel_accept_cb(mesh_b, accept_cb);
111
112         // Start both instances
113
114         start_meshlink_pair(mesh_a, mesh_b);
115
116         // Open channels from a to b.
117
118         meshlink_node_t *b = meshlink_get_node(mesh_a, "b");
119         assert(b);
120
121         meshlink_channel_t *channels[nchannels];
122
123         for(size_t i = 0; i < nchannels; i++) {
124                 channels[i] = meshlink_channel_open(mesh_a, b, i + 1, NULL, NULL, 0);
125                 assert(channels[i]);
126         }
127
128         // Send a large buffer of data on each channel.
129
130         for(size_t i = 0; i < nchannels; i++) {
131                 assert(meshlink_channel_aio_fd_send(mesh_a, channels[i], fileno(out_infos[i].file), size / 3, aio_fd_cb, &out_infos[i].aio_infos[0]));
132                 assert(meshlink_channel_aio_fd_send(mesh_a, channels[i], fileno(out_infos[i].file), size - size / 3, aio_fd_cb, &out_infos[i].aio_infos[1]));
133         }
134
135         // Wait for everyone to finish.
136
137         for(size_t i = 0; i < nchannels; i++) {
138                 assert(wait_sync_flag(&out_infos[i].aio_infos[0].flag, 10));
139                 assert(wait_sync_flag(&out_infos[i].aio_infos[1].flag, 10));
140                 assert(wait_sync_flag(&in_infos[i].aio_infos[0].flag, 10));
141                 assert(wait_sync_flag(&in_infos[i].aio_infos[1].flag, 10));
142         }
143
144         // Check that everything is correct.
145
146         for(size_t i = 0; i < nchannels; i++) {
147                 assert(fclose(in_infos[i].file) == 0);
148                 assert(fclose(out_infos[i].file) == 0);
149
150                 // One callback for each AIO buffer.
151                 assert(out_infos[i].aio_infos[0].callbacks == 1);
152                 assert(out_infos[i].aio_infos[1].callbacks == 1);
153                 assert(in_infos[i].aio_infos[0].callbacks == 1);
154                 assert(in_infos[i].aio_infos[1].callbacks == 1);
155
156                 // Correct size sent and received.
157                 assert(out_infos[i].aio_infos[0].size == size / 3);
158                 assert(out_infos[i].aio_infos[1].size == size - size / 3);
159                 assert(in_infos[i].aio_infos[0].size == size / 4);
160                 assert(in_infos[i].aio_infos[1].size == size - size / 4);
161
162                 // First batch of data should all be sent and received before the second batch
163                 for(size_t j = 0; j < nchannels; j++) {
164                         assert(timespec_lt(&out_infos[i].aio_infos[0].ts, &out_infos[j].aio_infos[1].ts));
165                         assert(timespec_lt(&in_infos[i].aio_infos[0].ts, &in_infos[j].aio_infos[1].ts));
166                 }
167
168                 // Files should be identical
169                 char command[PATH_MAX];
170                 snprintf(command, sizeof(command), "cmp channels_aio_fd.in channels_aio_fd.out%d", (int)i);
171                 assert(system(command) == 0);
172
173         }
174
175         // Clean up.
176
177         close_meshlink_pair(mesh_a, mesh_b);
178         free(outdata);
179 }