Context Navigation

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

ADTast-experimentalenumpthread-emulationqualifiedEnum

Last change on this file since 3263e2a4 was 3263e2a4, checked in by Thierry Delisle <tdelisle@…>, 3 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
29	enum {
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
42	enum { buffer_len = 10240 };
43	char buffer[buffer_len];
44
45	enum { TIMEGRAN = 1000000000LL, TIMES = 100000 };
46
47	int pipefd[2];
48	struct io_uring ring;
49
50	struct 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	};
60	static void my_sendfile(int out, int in, size_t size, struct stats *);
61	static void my_splice (int out, int in, size_t size, struct stats *);
62	static void my_iouring (int out, int in, size_t size, struct stats *);
63	static void my_ringlink(int out, int in, size_t size, struct stats *);
64	typedef void (sender_t)(int out, int in, size_t size, struct stats );
65
66	static void run(sender_t sender, struct addrinfo * addr, int infd, size_t size);
67
68	int 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
233	static 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
285	static 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
305	static 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
340	static 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
369	static 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
404	static 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.

Download in other formats: