Context Navigation

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
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	char * buf;
51
52	struct 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	};
62	static void my_sendfile(int out, int in, size_t size, struct stats *);
63	static void my_splice (int out, int in, size_t size, struct stats *);
64	static void my_iouring (int out, int in, size_t size, struct stats *);
65	static void my_ringlink(int out, int in, size_t size, struct stats *);
66	static void my_readwrit(int out, int in, size_t size, struct stats *);
67	typedef void (sender_t)(int out, int in, size_t size, struct stats );
68
69	static void run(sender_t sender, struct addrinfo * addr, int infd, size_t size);
70
71	int 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
225	static 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
277	static 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
297	static 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
332	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) {
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
361	static 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
396	static 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
476	static 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.

Download in other formats:

Original Format