source: benchmark/io/sendfile/producer.cfa @ 9c447e2

Last change on this file since 9c447e2 was 430ce61, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Added sendfile test in Cforall

  • 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.