source: benchmark/io/sendfile/producer.c@ 3263e2a4

ADT ast-experimental enum pthread-emulation qualifiedEnum stuck-waitfor-destruct
Last change on this file since 3263e2a4 was 3263e2a4, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Producer now has multiple io_uring implementations.
  • Property mode set to 100644
File size: 12.3 KB
Line 
1// programs that sends a file many times as fast as it can
2// compares sendfile to splice
3
4#define _GNU_SOURCE
5
6#include <stdint.h>
7#include <stdio.h>
8#include <stdlib.h>
9#include <string.h>
10
11#include <errno.h>
12#include <locale.h>
13#include <time.h>
14#include <unistd.h>
15
16#include <sys/ioctl.h>
17#include <sys/sendfile.h>
18#include <sys/socket.h>
19#include <sys/stat.h>
20#include <sys/types.h>
21#include <fcntl.h>
22
23#include <netinet/in.h>
24#include <arpa/inet.h>
25#include <netdb.h>
26
27#include <liburing.h>
28
29enum {
30 USAGE_ERROR = 1,
31 HOST_ERROR,
32 PIPE_ERROR,
33 FSTAT_ERROR,
34 SOCKET_ERROR,
35 CONNECT_ERROR,
36 SENDFILE_ERROR,
37 SPLICEIN_ERROR,
38 SPLICEOUT_ERROR,
39 URINGWAIT_ERROR
40};
41
42enum { buffer_len = 10240 };
43char buffer[buffer_len];
44
45enum { TIMEGRAN = 1000000000LL, TIMES = 100000 };
46
47int pipefd[2];
48struct io_uring ring;
49
50struct stats {
51 size_t calls;
52 size_t bytes;
53 struct {
54 struct {
55 size_t cnt;
56 size_t bytes;
57 } r, w;
58 } shorts;
59};
60static void my_sendfile(int out, int in, size_t size, struct stats *);
61static void my_splice (int out, int in, size_t size, struct stats *);
62static void my_iouring (int out, int in, size_t size, struct stats *);
63static void my_ringlink(int out, int in, size_t size, struct stats *);
64typedef void (*sender_t)(int out, int in, size_t size, struct stats *);
65
66static void run(sender_t sender, struct addrinfo * addr, int infd, size_t size);
67
68int main(int argc, char * argv[]) {
69 setlocale(LC_ALL, "");
70 const char * file_path;
71 struct addrinfo * addr;
72 int file_fd;
73 switch(argc) {
74 case 3:
75 {
76 // Open the file
77 const char * const path = argv[2];
78 int ret = open(path, 0, O_RDONLY);
79 if(ret < 0) {
80 fprintf( stderr, "cannot open file '%s': %s\n\n", path, strerror(errno) );
81 goto USAGE;
82 }
83
84 file_path = path;
85 file_fd = ret;
86
87
88 // connect to the address
89 char * state = 0;
90 char * str = argv[1];
91 const char * const host = strtok_r(str, ":", &state);
92 if(NULL == host) {
93 fprintf( stderr, "Invalid host:port specification, no host.\n\n" );
94 goto USAGE;
95 }
96
97 const char * const port = strtok_r(NULL, ":", &state);
98 if(NULL == port) {
99 fprintf( stderr, "Invalid host:port specification, no port.\n\n" );
100 goto USAGE;
101 }
102
103 printf("looking up '%s:%s'\n", host, port);
104
105 struct addrinfo hints = {};
106 struct addrinfo * pResultList = NULL;
107
108 hints.ai_family = AF_INET;
109 hints.ai_socktype = SOCK_STREAM;
110 hints.ai_flags = AI_NUMERICSERV;
111
112 ret = getaddrinfo(host, port, &hints, &pResultList);
113
114 switch(ret) {
115 case 0:
116 addr = pResultList;
117 goto DONE;
118
119 case EAI_ADDRFAMILY:
120 fprintf( stderr, "The specified network host does not have any network addresses in the requested address family.\n\n" );
121 break;
122
123 case EAI_AGAIN:
124 fprintf( stderr, "The name server returned a temporary failure indication. Try again later.\n\n" );
125 exit( HOST_ERROR );
126
127 case EAI_BADFLAGS:
128 fprintf( stderr, "hints.ai_flags contains invalid flags; or, hints.ai_flags included AI_CANONNAME and name was NULL.\n\n" );
129 exit( HOST_ERROR );
130
131 case EAI_FAIL:
132 fprintf( stderr, "The name server returned a permanent failure indication.\n\n" );
133 break;
134
135 case EAI_FAMILY:
136 fprintf( stderr, "The requested address family is not supported.\n\n" );
137 exit( HOST_ERROR );
138
139 case EAI_MEMORY:
140 fprintf( stderr, "Out of memory.\n\n" );
141 exit( HOST_ERROR );
142
143 case EAI_NODATA:
144 fprintf( stderr, "The specified network host exists, but does not have any network addresses defined.\n\n" );
145 break;
146
147 case EAI_NONAME:
148 fprintf( stderr, "The unkonwn host or invalid port.\n\n" );
149 break;
150
151 case EAI_SERVICE:
152 fprintf( stderr, "The requested service is not available for the requested socket type.\n\n" );
153 break;
154
155 case EAI_SOCKTYPE:
156 fprintf( stderr, "The requested socket type is not supported.\n\n" );
157 exit( HOST_ERROR );
158
159 case EAI_SYSTEM:
160 // Other system error, check errno for details.
161 default:
162 fprintf( stderr, "Unnown hostname error: (%d) %s\n\n", (int)errno, strerror(errno) );
163 exit( HOST_ERROR );
164 }
165 if(pResultList) freeaddrinfo(pResultList);
166 goto USAGE;
167 }
168 USAGE:
169 default:
170 fprintf( stderr, "USAGE: %s host:port file\n", argv[0] );
171 exit( USAGE_ERROR );
172 }
173
174 DONE:
175
176 io_uring_queue_init(16, &ring, 0);
177
178 {
179 char addr_str[INET_ADDRSTRLEN];
180 struct sockaddr_in * address = (struct sockaddr_in *) addr->ai_addr;
181 inet_ntop( AF_INET, &address->sin_addr, addr_str, INET_ADDRSTRLEN );
182 printf("sending '%s' to '%s:%i'\n", file_path, addr_str, ntohs(address->sin_port));
183 }
184
185 int ret = pipe(pipefd);
186 if( ret < 0 ) {
187 fprintf( stderr, "pipe error: (%d) %s\n\n", (int)errno, strerror(errno) );
188 exit( PIPE_ERROR );
189 }
190
191 {
192 ret = fcntl(pipefd[0], F_GETPIPE_SZ);
193 if( ret < 0 ) {
194 fprintf( stderr, "pipe fcntl error: (%d) %s\n\n", (int)errno, strerror(errno) );
195 exit( PIPE_ERROR );
196 }
197 printf("%d\n", ret);
198
199 ret = fcntl(pipefd[1], F_GETPIPE_SZ);
200 if( ret < 0 ) {
201 fprintf( stderr, "pipe fcntl 2 error: (%d) %s\n\n", (int)errno, strerror(errno) );
202 exit( PIPE_ERROR );
203 }
204 printf("%d\n", ret);
205 }
206
207 size_t file_size = 0;
208 {
209 struct stat buf;
210 ret = fstat(file_fd, &buf);
211 if(0 != ret) {
212 fprintf( stderr, "fstat error: (%d) %s\n\n", (int)errno, strerror(errno) );
213 exit( FSTAT_ERROR );
214 }
215 file_size = buf.st_size;
216 }
217
218 printf("--- splice ---\n");
219 run(my_splice , addr, file_fd, file_size);
220 printf("--- sendfile ---\n");
221 run(my_sendfile, addr, file_fd, file_size);
222 printf("--- io_uring ---\n");
223 run(my_iouring, addr, file_fd, file_size);
224 printf("--- io_uring + link ---\n");
225 run(my_ringlink, addr, file_fd, file_size);
226
227 close(pipefd[0]);
228 close(pipefd[1]);
229 close(file_fd);
230 return 0;
231}
232
233static void run(sender_t sender, struct addrinfo * addr, int infd, size_t size) {
234
235 int sock = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
236 if(sock < 0) {
237 fprintf( stderr, "socket error: (%d) %s\n\n", (int)errno, strerror(errno) );
238 exit( SOCKET_ERROR );
239 }
240
241 int ret = connect(sock, addr->ai_addr, addr->ai_addrlen);
242 if(ret < 0) {
243 fprintf( stderr, "connect error: (%d) %s\n\n", (int)errno, strerror(errno) );
244 exit( CONNECT_ERROR );
245 }
246
247 struct stats st;
248 st.calls = 0;
249 st.bytes = 0;
250 st.shorts.r.cnt = 0;
251 st.shorts.r.bytes = 0;
252 st.shorts.w.cnt = 0;
253 st.shorts.w.bytes = 0;
254
255 struct timespec after, before;
256
257 clock_gettime(CLOCK_MONOTONIC, &before);
258
259 for(long long int i = 0; i < TIMES; i++) {
260 sender( sock, infd, size, &st );
261 }
262
263 clock_gettime(CLOCK_MONOTONIC, &after);
264
265 close(sock);
266
267 uint64_t tb = ((int64_t)before.tv_sec * TIMEGRAN) + before.tv_nsec;
268 uint64_t ta = ((int64_t)after.tv_sec * TIMEGRAN) + after.tv_nsec;
269 double secs = ((double)ta - tb) / TIMEGRAN;
270
271 printf("Sent %'zu bytes in %'zu files, %f seconds\n", st.bytes, st.calls, secs);
272 printf(" - %'3.3f bytes per second\n", (((double)st.bytes) / secs));
273 printf(" - %'f seconds per file\n", secs / st.calls);
274 printf(" - %'3.3f bytes per calls\n", (((double)st.bytes) / st.calls));
275 if(st.shorts.r.cnt ){
276 printf(" - %'zu short reads\n", st.shorts.r.cnt);
277 printf(" - %'3.3f bytes per short read\n", (((double)st.shorts.r.bytes) / st.shorts.r.cnt));
278 } else printf("No short reads\n");
279 if(st.shorts.w.cnt ){
280 printf(" - %'zu short reads\n", st.shorts.w.cnt);
281 printf(" - %'3.3f bytes per short read\n", (((double)st.shorts.w.bytes) / st.shorts.w.cnt));
282 } else printf("No short writes\n");
283}
284
285static void my_sendfile(int out, int in, size_t size, struct stats * st) {
286 off_t off = 0;
287 for(;;) {
288
289 ssize_t ret = sendfile(out, in, &off, size);
290 if(ret < 0) {
291 fprintf( stderr, "connect error: (%d) %s\n\n", (int)errno, strerror(errno) );
292 exit( SENDFILE_ERROR );
293 }
294
295 st->calls++;
296 st->bytes += ret;
297 off += ret;
298 size -= ret;
299 if( size == 0 ) return;
300 st->shorts.r.cnt++;
301 st->shorts.r.bytes += ret;
302 }
303}
304
305static void my_splice (int out, int in, size_t size, struct stats * st) {
306 unsigned flags = 0; //SPLICE_F_MOVE; // | SPLICE_F_MORE;
307 off_t offset = 0;
308 size_t writes = 0;
309 for(;;) {
310 ssize_t reti = 0;
311 reti = splice(in, &offset, pipefd[1], NULL, size, flags);
312 if( reti < 0 ) {
313 fprintf( stderr, "splice in error: (%d) %s\n\n", (int)errno, strerror(errno) );
314 exit( SPLICEIN_ERROR );
315 }
316
317 size -= reti;
318 size_t in_pipe = reti;
319 for(;;) {
320 ssize_t reto = 0;
321 reto = splice(pipefd[0], NULL, out, NULL, in_pipe, flags);
322 if( reto < 0 ) {
323 fprintf( stderr, "splice out error: (%d) %s\n\n", (int)errno, strerror(errno) );
324 exit( SPLICEOUT_ERROR );
325 }
326 in_pipe -= reto;
327 writes += reto;
328 if(0 == in_pipe) break;
329 st->shorts.w.cnt++;
330 st->shorts.w.bytes += reto;
331 }
332 if(0 == size) break;
333 st->shorts.r.cnt++;
334 st->shorts.r.bytes += reti;
335 }
336 st->calls++;
337 st->bytes += writes;
338}
339
340static ssize_t naive_splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags) {
341 struct io_uring_sqe * sqe = io_uring_get_sqe(&ring);
342
343 io_uring_prep_splice(sqe, fd_in, NULL != off_in ? *off_in: -1, fd_out, NULL != off_out ? *off_out: -1, len, flags);
344
345 io_uring_submit(&ring);
346
347 struct io_uring_cqe * cqe = NULL;
348 /* wait for the sqe to complete */
349 int ret = io_uring_wait_cqe_nr(&ring, &cqe, 1);
350
351 /* read and process cqe event */
352 switch(ret) {
353 case 0:
354 {
355 ssize_t val = cqe->res;
356 if( cqe->res < 0 ) {
357 printf("Completion Error : %s\n", strerror( -cqe->res ));
358 return EXIT_FAILURE;
359 }
360 io_uring_cqe_seen(&ring, cqe);
361 return val;
362 }
363 default:
364 fprintf( stderr, "io_uring_wait error: (%d) %s\n\n", (int)-ret, strerror(-ret) );
365 exit( URINGWAIT_ERROR );
366 }
367}
368
369static void my_iouring (int out, int in, size_t size, struct stats * st) {
370 unsigned flags = 0; //SPLICE_F_MOVE; // | SPLICE_F_MORE;
371 off_t offset = 0;
372 size_t writes = 0;
373 for(;;) {
374 ssize_t reti = 0;
375 reti = naive_splice(in, &offset, pipefd[1], NULL, size, flags);
376 if( reti < 0 ) {
377 fprintf( stderr, "splice in error: (%d) %s\n\n", (int)errno, strerror(errno) );
378 exit( SPLICEIN_ERROR );
379 }
380
381 size -= reti;
382 size_t in_pipe = reti;
383 for(;;) {
384 ssize_t reto = 0;
385 reto = naive_splice(pipefd[0], NULL, out, NULL, in_pipe, flags);
386 if( reto < 0 ) {
387 fprintf( stderr, "splice out error: (%d) %s\n\n", (int)errno, strerror(errno) );
388 exit( SPLICEOUT_ERROR );
389 }
390 in_pipe -= reto;
391 writes += reto;
392 if(0 == in_pipe) break;
393 st->shorts.w.cnt++;
394 st->shorts.w.bytes += reto;
395 }
396 if(0 == size) break;
397 st->shorts.r.cnt++;
398 st->shorts.r.bytes += reti;
399 }
400 st->calls++;
401 st->bytes += writes;
402}
403
404static void my_ringlink(int out, int in, size_t size, struct stats * st) {
405 enum { SPLICE_IN, SPLICE_OUT };
406
407 size_t in_pipe = size;
408 off_t offset = 0;
409 bool has_in = false;
410 bool has_out = false;
411 while(true) {
412 if(!has_in && size > 0) {
413 struct io_uring_sqe * sqe = io_uring_get_sqe(&ring);
414 io_uring_prep_splice(sqe, in, offset, pipefd[1], -1, size, 0);
415 sqe->user_data = SPLICE_IN;
416 has_in = true;
417 }
418 if(!has_out) {
419 struct io_uring_sqe * sqe = io_uring_get_sqe(&ring);
420 io_uring_prep_splice(sqe, pipefd[0], -1, out, -1, in_pipe, 0);
421 sqe->user_data = SPLICE_OUT;
422 if(has_in) sqe->flags = IOSQE_IO_LINK;
423 has_out = true;
424 }
425
426 int ret = io_uring_submit_and_wait(&ring, 1);
427 if(ret < 0) {
428 fprintf( stderr, "io_uring_submit error: (%d) %s\n\n", (int)-ret, strerror(-ret) );
429 exit( URINGWAIT_ERROR );
430 }
431
432 /* poll the cq and count how much polling we did */
433 while(true) {
434 struct io_uring_cqe * cqe = NULL;
435 /* wait for the sqe to complete */
436 int ret = io_uring_wait_cqe_nr(&ring, &cqe, 0);
437
438 /* read and process cqe event */
439 switch(ret) {
440 case 0:
441 if( cqe->res < 0 ) {
442 printf("Completion Error : %s\n", strerror( -cqe->res ));
443 exit( URINGWAIT_ERROR );
444 }
445
446 ssize_t write = cqe->res;
447 int which = cqe->user_data;
448 io_uring_cqe_seen(&ring, cqe);
449 switch( which ) {
450 case SPLICE_IN:
451 has_in = false;
452 size -= write;
453 offset += write;
454 if(0 == size) break;
455 st->shorts.r.cnt++;
456 st->shorts.r.bytes += write;
457 break;
458 case SPLICE_OUT:
459 has_out = false;
460 in_pipe -= write;
461 st->bytes += write;
462 if(0 == in_pipe) break;
463 st->shorts.w.cnt++;
464 st->shorts.w.bytes += write;
465 break;
466 default:
467 printf("Completion Error : unknown user data\n");
468 exit( URINGWAIT_ERROR );
469 }
470 continue;
471 case -EAGAIN:
472 goto OUTER;
473 default:
474 fprintf( stderr, "io_uring_get_cqe error: (%d) %s\n\n", (int)-ret, strerror(-ret) );
475 exit( URINGWAIT_ERROR );
476 }
477 }
478 OUTER:
479 if(0 == in_pipe) break;
480 }
481 st->calls++;
482}
Note: See TracBrowser for help on using the repository browser.