source: benchmark/io/sendfile/producer.c@ fc2c57a9

ADT ast-experimental pthread-emulation qualifiedEnum
Last change on this file since fc2c57a9 was 3a40df6, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

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