]> git.meshlink.io Git - meshlink/blob - src/utcp-test.c
Move timespec functions to utils.h.
[meshlink] / src / utcp-test.c
1 /*
2     utcp-test.c -- Test application for UTCP
3     Copyright (C) 2014-2020 Guus Sliepen <guus@tinc-vpn.org>
4
5     This program is free software; you can redistribute it and/or modify
6     it under the terms of the GNU General Public License as published by
7     the Free Software Foundation; either version 2 of the License, or
8     (at your option) any later version.
9
10     This program is distributed in the hope that it will be useful,
11     but WITHOUT ANY WARRANTY; without even the implied warranty of
12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13     GNU General Public License for more details.
14
15     You should have received a copy of the GNU General Public License along
16     with this program; if not, write to the Free Software Foundation, Inc.,
17     51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20 #include "system.h"
21 #include <poll.h>
22 #include <time.h>
23
24 #include "utcp.h"
25
26 #define DIR_READ 1
27 #define DIR_WRITE 2
28
29 static struct utcp_connection *c;
30 static int dir = DIR_READ | DIR_WRITE;
31 static long inpktno;
32 static long outpktno;
33 static long dropfrom;
34 static long dropto;
35 static double reorder;
36 static long reorder_dist = 10;
37 static double dropin;
38 static double dropout;
39 static long total_out;
40 static long total_in;
41 static FILE *reference;
42 static long mtu;
43 static long bufsize;
44
45 static char *reorder_data;
46 static size_t reorder_len;
47 static int reorder_countdown;
48
49 #if UTCP_DEBUG
50 static void debug(const char *format, ...) {
51         struct timespec tv;
52         char buf[1024];
53         int len;
54
55         clock_gettime(CLOCK_REALTIME, &tv);
56         len = snprintf(buf, sizeof(buf), "%ld.%06lu ", (long)tv.tv_sec, tv.tv_nsec / 1000);
57         va_list ap;
58         va_start(ap, format);
59         len += vsnprintf(buf + len, sizeof(buf) - len, format, ap);
60         va_end(ap);
61
62         if(len > 0 && (size_t)len < sizeof(buf)) {
63                 fwrite(buf, len, 1, stderr);
64         }
65 }
66 #else
67 #define debug(...) do {} while(0)
68 #endif
69
70 static ssize_t do_recv(struct utcp_connection *rc, const void *data, size_t len) {
71         (void)rc;
72
73         if(!data || !len) {
74                 if(errno) {
75                         debug("Error: %s\n", strerror(errno));
76                         dir = 0;
77                 } else {
78                         dir &= ~DIR_WRITE;
79                         debug("Connection closed by peer\n");
80                 }
81
82                 return -1;
83         }
84
85         if(reference) {
86                 char buf[len];
87
88                 if(fread(buf, len, 1, reference) != 1) {
89                         debug("Error reading reference\n");
90                         abort();
91                 }
92
93                 if(memcmp(buf, data, len)) {
94                         debug("Received data differs from reference\n");
95                         abort();
96                 }
97         }
98
99         return write(1, data, len);
100 }
101
102 static void do_accept(struct utcp_connection *nc, uint16_t port) {
103         (void)port;
104         utcp_accept(nc, do_recv, NULL);
105         c = nc;
106
107         if(bufsize) {
108                 utcp_set_sndbuf(c, NULL, bufsize);
109                 utcp_set_rcvbuf(c, NULL, bufsize);
110         }
111
112         utcp_set_accept_cb(c->utcp, NULL, NULL);
113 }
114
115 static ssize_t do_send(struct utcp *utcp, const void *data, size_t len) {
116         int s = *(int *)utcp->priv;
117         outpktno++;
118
119         if(outpktno >= dropfrom && outpktno < dropto) {
120                 if(drand48() < dropout) {
121                         debug("Dropped outgoing packet\n");
122                         return len;
123                 }
124
125                 if(!reorder_data && drand48() < reorder) {
126                         reorder_data = malloc(len);
127
128                         if(!reorder_data) {
129                                 debug("Out of memory\n");
130                                 return len;
131                         }
132
133                         reorder_len = len;
134                         memcpy(reorder_data, data, len);
135                         reorder_countdown = 1 + drand48() * reorder_dist;
136                         return len;
137                 }
138         }
139
140         if(reorder_data) {
141                 if(--reorder_countdown < 0) {
142                         total_out += reorder_len;
143                         send(s, reorder_data, reorder_len, MSG_DONTWAIT);
144                         free(reorder_data);
145                         reorder_data = NULL;
146                 }
147         }
148
149         total_out += len;
150         ssize_t result = send(s, data, len, MSG_DONTWAIT);
151
152         if(result <= 0) {
153                 debug("Error sending UDP packet: %s\n", strerror(errno));
154         }
155
156         return result;
157 }
158
159 static void set_mtu(struct utcp *u, int s) {
160 #ifdef IP_MTU
161
162         if(!mtu) {
163                 socklen_t optlen = sizeof(mtu);
164                 getsockopt(s, IPPROTO_IP, IP_MTU, &mtu, &optlen);
165         }
166
167 #else
168         (void)s;
169 #endif
170
171         if(!mtu || mtu == 65535) {
172                 mtu = 1500;
173         }
174
175         debug("Using MTU %lu\n", mtu);
176
177         utcp_set_mtu(u, mtu ? mtu - 28 : 1300);
178 }
179
180 int main(int argc, char *argv[]) {
181         srand(time(NULL));
182         srand48(time(NULL));
183
184         if(argc < 2 || argc > 3) {
185                 return 1;
186         }
187
188         bool server = argc == 2;
189         bool connected = false;
190         uint32_t flags = UTCP_TCP;
191         size_t read_size = 102400;
192
193         if(getenv("DROPIN")) {
194                 dropin = atof(getenv("DROPIN"));
195         }
196
197         if(getenv("DROPOUT")) {
198                 dropout = atof(getenv("DROPOUT"));
199         }
200
201         if(getenv("DROPFROM")) {
202                 dropfrom = atoi(getenv("DROPFROM"));
203         }
204
205         if(getenv("DROPTO")) {
206                 dropto = atoi(getenv("DROPTO"));
207         }
208
209         if(getenv("REORDER")) {
210                 reorder = atof(getenv("REORDER"));
211         }
212
213         if(getenv("REORDER_DIST")) {
214                 reorder_dist = atoi(getenv("REORDER_DIST"));
215         }
216
217         if(getenv("FLAGS")) {
218                 flags = atoi(getenv("FLAGS"));
219         }
220
221         if(getenv("READ_SIZE")) {
222                 read_size = atoi(getenv("READ_SIZE"));
223         }
224
225         if(getenv("MTU")) {
226                 mtu = atoi(getenv("MTU"));
227         }
228
229         if(getenv("BUFSIZE")) {
230                 bufsize = atoi(getenv("BUFSIZE"));
231         }
232
233         char *reference_filename = getenv("REFERENCE");
234
235         if(reference_filename) {
236                 reference = fopen(reference_filename, "r");
237         }
238
239         if(dropto < dropfrom) {
240                 dropto = 1 << 30;
241         }
242
243         struct addrinfo *ai;
244
245         struct addrinfo hint = {
246                 .ai_flags = server ? AI_PASSIVE : 0,
247                 .ai_socktype = SOCK_DGRAM,
248         };
249
250         getaddrinfo(server ? NULL : argv[1], server ? argv[1] : argv[2], &hint, &ai);
251
252         if(!ai) {
253                 debug("Could not lookup address: %s\n", strerror(errno));
254                 return 1;
255         }
256
257         int s = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
258
259         if(s == -1) {
260                 debug("Could not create socket: %s\n", strerror(errno));
261                 return 1;
262         }
263
264         static const int one = 1;
265         setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
266
267         if(server) {
268                 if(bind(s, ai->ai_addr, ai->ai_addrlen)) {
269                         debug("Could not bind: %s\n", strerror(errno));
270                         return 1;
271                 }
272         } else {
273 #ifdef SO_NOSIGPIPE
274                 int nosigpipe = 1;
275                 setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe, sizeof(nosigpipe));
276 #endif
277
278                 if(connect(s, ai->ai_addr, ai->ai_addrlen)) {
279                         debug("Could not connect: %s\n", strerror(errno));
280                         return 1;
281                 }
282
283                 connected = true;
284         }
285
286         freeaddrinfo(ai);
287
288         struct utcp *u = utcp_init(server ? do_accept : NULL, NULL, do_send, &s);
289
290         if(!u) {
291                 debug("Could not initialize UTCP\n");
292                 return 1;
293         }
294
295         utcp_set_user_timeout(u, 10);
296
297         if(!server) {
298                 set_mtu(u, s);
299                 c = utcp_connect_ex(u, 1, do_recv, NULL, flags);
300
301                 if(bufsize) {
302                         utcp_set_sndbuf(c, NULL, bufsize);
303                         utcp_set_rcvbuf(c, NULL, bufsize);
304                 }
305         }
306
307         struct pollfd fds[2] = {
308                 {.fd = 0, .events = POLLIN | POLLERR | POLLHUP},
309                 {.fd = s, .events = POLLIN | POLLERR | POLLHUP},
310         };
311
312         char buf[102400];
313
314         struct timespec timeout = utcp_timeout(u);
315
316         while(!connected || utcp_is_active(u)) {
317                 size_t max = c ? utcp_get_sndbuf_free(c) : 0;
318
319                 if(max > sizeof(buf)) {
320                         max = sizeof(buf);
321                 }
322
323                 if(max > read_size) {
324                         max = read_size;
325                 }
326
327                 int timeout_ms = timeout.tv_sec * 1000 + timeout.tv_nsec / 1000000 + 1;
328
329                 debug("polling, dir = %d, timeout = %d\n", dir, timeout_ms);
330
331                 if((dir & DIR_READ) && max) {
332                         poll(fds, 2, timeout_ms);
333                 } else {
334                         poll(fds + 1, 1, timeout_ms);
335                 }
336
337                 if(fds[0].revents) {
338                         fds[0].revents = 0;
339                         ssize_t len = read(0, buf, max);
340                         debug("stdin %zd\n", len);
341
342                         if(len <= 0) {
343                                 fds[0].fd = -1;
344                                 dir &= ~DIR_READ;
345
346                                 if(c) {
347                                         utcp_shutdown(c, SHUT_WR);
348                                 }
349
350                                 if(len == -1) {
351                                         break;
352                                 } else {
353                                         continue;
354                                 }
355                         }
356
357                         if(c) {
358                                 ssize_t sent = utcp_send(c, buf, len);
359
360                                 if(sent != len) {
361                                         debug("Short send: %zd != %zd\n", sent, len);
362                                 }
363                         }
364                 }
365
366                 if(fds[1].revents) {
367                         fds[1].revents = 0;
368                         struct sockaddr_storage ss;
369                         socklen_t sl = sizeof(ss);
370                         int len = recvfrom(s, buf, sizeof(buf), MSG_DONTWAIT, (struct sockaddr *)&ss, &sl);
371                         debug("netin %zu\n", len);
372
373                         if(len <= 0) {
374                                 debug("Error receiving UDP packet: %s\n", strerror(errno));
375                                 break;
376                         }
377
378                         if(!connected) {
379                                 if(!connect(s, (struct sockaddr *)&ss, sl)) {
380                                         connected = true;
381                                         set_mtu(u, s);
382                                 }
383                         }
384
385                         inpktno++;
386
387                         if(inpktno >= dropto || inpktno < dropfrom || drand48() >= dropin) {
388                                 total_in += len;
389
390                                 if(utcp_recv(u, buf, len) == -1) {
391                                         debug("Error receiving UTCP packet: %s\n", strerror(errno));
392                                 }
393                         } else {
394                                 debug("Dropped incoming packet\n");
395                         }
396                 }
397
398                 timeout = utcp_timeout(u);
399         };
400
401         utcp_close(c);
402
403         utcp_exit(u);
404
405         free(reorder_data);
406
407         debug("Total bytes in: %ld, out: %ld\n", total_in, total_out);
408
409         return 0;
410 }