How do you make blocking writes on a pipe?

[RichardHabeeb]

It seems like io_uring doesn't always respect O_NONBLOCK flag on pipe file descriptors, regardless of whether IOSQE_ASYNC is used. I.e. I have been unable to consistently get submissions to write more than a full pipe of bytes when the reader is sluggish. The worker thread seems to be writing till the pipe is full (however many pages are free), then completing. If you just used a synchronous write() it typically blocks until all bytes are written or the pipe is closed. I've cleared any O_NONBLOCK flags on the fd and tried setting IOSQE_ASYNC. I have been working extensively with io_uring on some research for awhile now, but mostly on files so I just noticed this.

I think this could theoretically still be semantically correct according to Posix (https://pubs.opengroup.org/onlinepubs/9799919799/functions/write.html), but it makes for many more trips to user space and many more SQEs for large pipe writes.

Here's a test program, to illustrate, and an example execution on my machine:

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <liburing.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>



int
main(int argc, const char *argv[])
{
	struct io_uring ring;
	struct io_uring_sqe *sqe;
	struct io_uring_cqe *cqe;
	int ret;

	if(argc != 3) {
		fprintf(stderr, "Usage: %s <size> <N>\n", argv[0]);
		exit(EXIT_FAILURE);
	}

	size_t sz = atoll(argv[1]);
	size_t n_buf = atoll(argv[2]);

	if(sz < 2) {
		fprintf(stderr, "size must be >= 2\n");
		exit(EXIT_FAILURE);
	}

	int stdout2 = dup(STDOUT_FILENO);

	int flags = fcntl(stdout2, F_GETFL);
	if (flags < 0) {
		perror("fcntl");
		exit(EXIT_FAILURE);
	}

	//flags |= O_NONBLOCK; //tried both ways
	flags &= ~O_NONBLOCK;
	if (fcntl(stdout2, F_SETFL, flags) < 0) {
		perror("fcntl F_SETFL");
		exit(EXIT_FAILURE);
	}


	unsigned char * bufs[n_buf];

	for(size_t i = 0; i < n_buf; i++) {
		bufs[i] = malloc(sz);
		if(bufs[i] == NULL) {
			perror("malloc");
			exit(EXIT_FAILURE);
		}
		for(size_t j = 0; j < sz; j++) {
			bufs[i][j] = 'a' + (i % 26);
		}
		bufs[i][sz - 2] = '\n';
		bufs[i][sz - 1] = '\0';
	}



	ret = io_uring_queue_init(2*n_buf, &ring, 0);
	if (ret < 0) {
		perror("io_uring_queue_init");
		return 1;
	}


	for(size_t i = 0; i < n_buf; i++) {
		sqe = io_uring_get_sqe(&ring);
		if (!sqe) {
			perror("io_uring_get_sqe");
			return 1;
		}

		io_uring_prep_write(sqe, stdout2, bufs[i], sz, -1);
		io_uring_sqe_set_flags(sqe, IOSQE_ASYNC);
	}

	ret = io_uring_submit(&ring);
	if (ret < 0) {
		perror("io_uring_submit");
		return 1;
	}

	for(size_t i = 0; i < n_buf; i++) {
		ret = io_uring_wait_cqe(&ring, &cqe);
		if (ret < 0) {
			perror("io_uring_wait_cqe");
			return 1;
		}

		if (cqe->res < 0) {
			fprintf(stderr, "Write %zi failed: %s\n", i, strerror(-cqe->res));
			break;
		}

		fprintf(stderr, "Wrote %d bytes of %zu (flags=%x)\n", cqe->res, sz, cqe->flags);

		io_uring_cqe_seen(&ring, cqe);
	}

	io_uring_queue_exit(&ring);
	close(stdout2);
	sync();
	return 0;
}

$ ./iouring_write2 $((1024*1024)) 65 | wc > /dev/null
Wrote 65536 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 32768 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 32768 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 1048576 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)
Wrote 16384 bytes of 1048576 (flags=0)

[RichardHabeeb]

Note that if you set O_NONBLOCK the worker thread will sometimes return without writing as expected. This makes me think that for io_uring O_NONBLOCK only makes the write "poll" for any space, but not to block until the entire write is complete.

I know that for a correct program you will need to each "logical" write to be a loop anyway, adjusting based on how much you have already written; however, it seems like without adjusting the pipe size, you will need issue many sqe's for large pipe writes.

E.g. if the reader is never concurrent with the writer, then for each 1MB write and 64KiB pipe size, you will need to issue at 16 sqes, and for each 1GiB you would need to issue around 64K SQEs (which for me seems to be degrading performance, but I am doing more testing).