Context Navigation

producer.cfa @ aa144c5a

ADTast-experimentalpthread-emulation

Last change on this file since aa144c5a 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

Rev	Line
[430ce61]	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