PIPE(2): Port to RTEMS

6 files changed, 486 insertions, 2 deletions

diff --git a/freebsd/sys/kern/sys_pipe.c b/freebsd/sys/kern/sys_pipe.c
index 5565c007..45d3ed13 100644..100755
--- a/freebsd/sys/kern/sys_pipe.c
+++ b/freebsd/sys/kern/sys_pipe.c
@@ -131,6 +131,9 @@ __FBSDID("$FreeBSD$");
 #include <vm/uma.h>
 
 /* XXX */
+#ifdef __rtems__
+static
+#endif /* __rtems__ */
 int	do_pipe(struct thread *td, int fildes[2], int flags);
 
 /*
@@ -143,6 +146,7 @@ int	do_pipe(struct thread *td, int fildes[2], int flags);
 /*
  * interfaces to the outside world
  */
+#ifndef __rtems__
 static fo_rdwr_t	pipe_read;
 static fo_rdwr_t	pipe_write;
 static fo_truncate_t	pipe_truncate;
@@ -165,6 +169,39 @@ static struct fileops pipeops = {
 	.fo_chown = invfo_chown,
 	.fo_flags = DFLAG_PASSABLE
 };
+#else /* __rtems__ */
+#define PIPE_NODIRECT
+#define	PRIBIO			(0)
+
+static int rtems_bsd_pipe_open(rtems_libio_t *iop, const char *path, int oflag, mode_t mode);
+static int rtems_bsd_pipe_close(rtems_libio_t *iop);
+static ssize_t rtems_bsd_pipe_read(rtems_libio_t *iop, void *buffer, size_t count);
+static ssize_t rtems_bsd_pipe_write(rtems_libio_t *iop, const void *buffer, size_t count);
+static int rtems_bsd_pipe_ioctl(rtems_libio_t *iop, ioctl_command_t request, void *buffer);
+static int rtems_bsd_pipe_stat(const rtems_filesystem_location_info_t *loc,	struct stat *buf);
+static int rtems_bsd_pipe_fcntl(rtems_libio_t *iop, int cmd);
+static int rtems_bsd_pipe_poll(rtems_libio_t *iop, int events);
+int rtems_bsd_pipe_kqfilter(rtems_libio_t *iop, struct knote *kn);
+
+static const rtems_filesystem_file_handlers_r pipeops = {
+	.open_h = rtems_bsd_pipe_open,
+	.close_h = rtems_bsd_pipe_close,
+	.read_h = rtems_bsd_pipe_read,
+	.write_h = rtems_bsd_pipe_write,
+	.ioctl_h = rtems_bsd_pipe_ioctl,
+	.lseek_h = rtems_filesystem_default_lseek,
+	.fstat_h = rtems_bsd_pipe_stat,
+	.ftruncate_h = rtems_filesystem_default_ftruncate,
+	.fsync_h = rtems_filesystem_default_fsync_or_fdatasync,
+	.fdatasync_h = rtems_filesystem_default_fsync_or_fdatasync,
+	.fcntl_h = rtems_bsd_pipe_fcntl,
+	.poll_h = rtems_bsd_pipe_poll,
+	.kqfilter_h = rtems_bsd_pipe_kqfilter
+};
+
+long	maxpipekva;			/* Limit on pipe KVA */
+
+#endif /* __rtems__ */
 
 static void	filt_pipedetach(struct knote *kn);
 static int	filt_piperead(struct knote *kn, long hint);
@@ -266,7 +303,11 @@ pipe_zone_ctor(void *mem, int size, void *arg, int flags)
 	 */
 	rpipe = &pp->pp_rpipe;
 	bzero(rpipe, sizeof(*rpipe));
+#ifndef __rtems__
 	vfs_timestamp(&rpipe->pipe_ctime);
+#else /* __rtems__ */
+	rpipe->pipe_ctime.tv_sec = time(NULL);
+#endif /* __rtems__ */
 	rpipe->pipe_atime = rpipe->pipe_mtime = rpipe->pipe_ctime;
 
 	wpipe = &pp->pp_wpipe;
@@ -336,7 +377,11 @@ kern_pipe(struct thread *td, int fildes[2])
 int
 do_pipe(struct thread *td, int fildes[2], int flags)
 {
+#ifndef __rtems__
 	struct filedesc *fdp = td->td_proc->p_fd;
+#else /* __rtems__ */
+	struct filedesc *fdp = NULL;
+#endif /* __rtems__ */
 	struct file *rf, *wf;
 	struct pipepair *pp;
 	struct pipe *rpipe, *wpipe;
@@ -422,6 +467,28 @@ sys_pipe(struct thread *td, struct pipe_args *uap)
 
 	return (0);
 }
+#ifdef __rtems__
+int
+pipe(int fildes[2])
+{
+	struct thread *td = rtems_bsd_get_curthread_or_null();
+	int error;
+
+	if (td != NULL) {
+		error = sys_pipe(td, NULL);
+	} else {
+		error = ENOMEM;
+	}
+
+	if (error == 0) {
+		fildes[0] = td->td_retval[0];
+		fildes[1] = td->td_retval[1];
+		return error;
+	} else {
+		rtems_set_errno_and_return_minus_one(error);
+	}
+}
+#endif /* __rtems__ */
 
 /*
  * Allocate kva for pipe circular buffer, the space is pageable
@@ -448,12 +515,17 @@ retry:
 		size = cnt;
 
 	size = round_page(size);
+#ifndef __rtems__
 	buffer = (caddr_t) vm_map_min(pipe_map);
 
 	error = vm_map_find(pipe_map, NULL, 0,
 		(vm_offset_t *) &buffer, size, 1,
 		VM_PROT_ALL, VM_PROT_ALL, 0);
 	if (error != KERN_SUCCESS) {
+#else /* __rtems__ */
+	buffer = malloc(size, M_TEMP, M_WAITOK | M_ZERO);
+	if (buffer == NULL) {
+#endif /* __rtems__ */
 		if ((cpipe->pipe_buffer.buffer == NULL) &&
 			(size > SMALL_PIPE_SIZE)) {
 			size = SMALL_PIPE_SIZE;
@@ -716,7 +788,11 @@ pipe_read(fp, uio, active_cred, flags, td)
 			 * Handle non-blocking mode operation or
 			 * wait for more data.
 			 */
+#ifndef __rtems__
 			if (fp->f_flag & FNONBLOCK) {
+#else /* __rtems__ */
+			if (rtems_bsd_libio_flags_to_fflag(fp->f_io.flags) & FNONBLOCK) {
+#endif /* __rtems__ */
 				error = EAGAIN;
 			} else {
 				rpipe->pipe_state |= PIPE_WANTR;
@@ -736,7 +812,11 @@ locked_error:
 
 	/* XXX: should probably do this before getting any locks. */
 	if (error == 0)
+#ifndef __rtems__
 		vfs_timestamp(&rpipe->pipe_atime);
+#else /* __rtems__ */
+		rpipe->pipe_atime.tv_sec = time(NULL);
+#endif /* __rtems__ */
 unlocked_error:
 	--rpipe->pipe_busy;
 
@@ -762,6 +842,40 @@ unlocked_error:
 	PIPE_UNLOCK(rpipe);
 	return (error);
 }
+#ifdef __rtems__
+static ssize_t
+rtems_bsd_pipe_read(rtems_libio_t *iop, void *buffer, size_t count)
+{
+	struct thread *td = rtems_bsd_get_curthread_or_null();
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+	struct iovec iov = {
+		.iov_base = buffer,
+		.iov_len = count
+	};
+	struct uio auio = {
+		.uio_iov = &iov,
+		.uio_iovcnt = 1,
+		.uio_offset = 0,
+		.uio_resid = count,
+		.uio_segflg = UIO_USERSPACE,
+		.uio_rw = UIO_READ,
+		.uio_td = td
+	};
+	int error;
+
+	if (td != NULL) {
+		error = pipe_read(fp, &auio, NULL, 0, NULL);
+	} else {
+		error = ENOMEM;
+	}
+
+	if (error == 0) {
+		return (count - auio.uio_resid);
+	} else {
+		rtems_set_errno_and_return_minus_one(error);
+	}
+}
+#endif /* __rtems__ */
 
 #ifndef PIPE_NODIRECT
 /*
@@ -1189,7 +1303,11 @@ pipe_write(fp, uio, active_cred, flags, td)
 			/*
 			 * don't block on non-blocking I/O
 			 */
+#ifndef __rtems__
 			if (fp->f_flag & FNONBLOCK) {
+#else /* __rtems__ */
+			if (rtems_bsd_libio_flags_to_fflag(fp->f_io.flags) & FNONBLOCK) {
+#endif /* __rtems__ */
 				error = EAGAIN;
 				pipeunlock(wpipe);
 				break;
@@ -1237,7 +1355,11 @@ pipe_write(fp, uio, active_cred, flags, td)
 	}
 
 	if (error == 0)
+#ifndef __rtems__
 		vfs_timestamp(&wpipe->pipe_mtime);
+#else /* __rtems__ */
+		wpipe->pipe_mtime.tv_sec = time(NULL);
+#endif /* __rtems__ */
 
 	/*
 	 * We have something to offer,
@@ -1250,8 +1372,43 @@ pipe_write(fp, uio, active_cred, flags, td)
 	PIPE_UNLOCK(rpipe);
 	return (error);
 }
+#ifdef __rtems__
+static ssize_t
+rtems_bsd_pipe_write(rtems_libio_t *iop, const void *buffer, size_t count)
+{
+	struct thread *td = rtems_bsd_get_curthread_or_null();
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+	struct iovec iov = {
+		.iov_base = __DECONST(void *, buffer),
+		.iov_len = count
+	};
+	struct uio auio = {
+		.uio_iov = &iov,
+		.uio_iovcnt = 1,
+		.uio_offset = 0,
+		.uio_resid = count,
+		.uio_segflg = UIO_USERSPACE,
+		.uio_rw = UIO_WRITE,
+		.uio_td = td
+	};
+	int error;
+
+	if (td != NULL) {
+		error = pipe_write(fp, &auio, NULL, 0, NULL);
+	} else {
+		error = ENOMEM;
+	}
+
+	if (error == 0) {
+		return (count - auio.uio_resid);
+	} else {
+		rtems_set_errno_and_return_minus_one(error);
+	}
+}
+#endif /* __rtems__ */
 
 /* ARGSUSED */
+#ifndef __rtems__
 static int
 pipe_truncate(fp, length, active_cred, td)
 	struct file *fp;
@@ -1262,6 +1419,7 @@ pipe_truncate(fp, length, active_cred, td)
 
 	return (EINVAL);
 }
+#endif /* __rtems__ */
 
 /*
  * we implement a very minimal set of ioctls for compatibility with sockets.
@@ -1336,6 +1494,23 @@ pipe_ioctl(fp, cmd, data, active_cred, td)
 out_unlocked:
 	return (error);
 }
+#ifdef __rtems__
+static int
+rtems_bsd_pipe_ioctl(rtems_libio_t *iop, ioctl_command_t request, void *buffer)
+{
+	struct thread *td = rtems_bsd_get_curthread_or_null();
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+	int error;
+
+	if (td != NULL) {
+		error = pipe_ioctl(fp, request, buffer, NULL, td);
+	} else {
+		error = ENOMEM;
+	}
+
+	return rtems_bsd_error_to_status_and_errno(error);
+}
+#endif /* __rtems__ */
 
 static int
 pipe_poll(fp, events, active_cred, td)
@@ -1400,11 +1575,29 @@ locked_error:
 
 	return (revents);
 }
+#ifdef __rtems__
+static int
+rtems_bsd_pipe_poll(rtems_libio_t *iop, int events)
+{
+	struct thread *td = rtems_bsd_get_curthread_or_null();
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+	int error;
+
+	if (td != NULL) {
+		error = pipe_poll(fp, events, NULL, td);
+	} else {
+		error = ENOMEM;
+	}
+
+	return error;
+}
+#endif /* __rtems__ */
 
 /*
  * We shouldn't need locks here as we're doing a read and this should
  * be a natural race.
  */
+#ifndef __rtems__
 static int
 pipe_stat(fp, ub, active_cred, td)
 	struct file *fp;
@@ -1413,12 +1606,19 @@ pipe_stat(fp, ub, active_cred, td)
 	struct thread *td;
 {
 	struct pipe *pipe;
+#else /* __rtems__ */
+static int
+pipe_stat(struct pipe *pipe, struct stat *ub)
+{
+#endif /* __rtems__ */
 	int new_unr;
 #ifdef MAC
 	int error;
 #endif
 
+#ifndef __rtems__
 	pipe = fp->f_data;
+#endif /* __rtems__ */
 	PIPE_LOCK(pipe);
 #ifdef MAC
 	error = mac_pipe_check_stat(active_cred, pipe->pipe_pair);
@@ -1446,7 +1646,9 @@ pipe_stat(fp, ub, active_cred, td)
 	}
 	PIPE_UNLOCK(pipe);
 
+#ifndef __rtems__
 	bzero(ub, sizeof(*ub));
+#endif /* __rtems__ */
 	ub->st_mode = S_IFIFO;
 	ub->st_blksize = PAGE_SIZE;
 	if (pipe->pipe_state & PIPE_DIRECTW)
@@ -1457,15 +1659,35 @@ pipe_stat(fp, ub, active_cred, td)
 	ub->st_atim = pipe->pipe_atime;
 	ub->st_mtim = pipe->pipe_mtime;
 	ub->st_ctim = pipe->pipe_ctime;
+#ifndef __rtems__
 	ub->st_uid = fp->f_cred->cr_uid;
 	ub->st_gid = fp->f_cred->cr_gid;
 	ub->st_dev = pipedev_ino;
 	ub->st_ino = pipe->pipe_ino;
+#else /* __rtems__ */
+	ub->st_uid = BSD_DEFAULT_UID;
+	ub->st_gid = BSD_DEFAULT_GID;
+	ub->st_dev = rtems_filesystem_make_dev_t(0xcc494cd6U, 0x1d970b4dU);
+	ub->st_ino = pipe->pipe_ino;
+#endif /* __rtems__ */
 	/*
 	 * Left as 0: st_nlink, st_rdev, st_flags, st_gen.
 	 */
 	return (0);
 }
+#ifdef __rtems__
+static int
+rtems_bsd_pipe_stat(
+	const rtems_filesystem_location_info_t *loc,
+	struct stat *buf
+)
+{
+	struct pipe *pipe = rtems_bsd_loc_to_f_data(loc);
+	int error = pipe_stat(pipe, buf);
+
+	return rtems_bsd_error_to_status_and_errno(error);
+}
+#endif /* __rtems__ */
 
 /* ARGSUSED */
 static int
@@ -1475,7 +1697,11 @@ pipe_close(fp, td)
 {
 	struct pipe *cpipe = fp->f_data;
 
+#ifndef __rtems__
 	fp->f_ops = &badfileops;
+#else /* __rtems__ */
+	fp->f_io.pathinfo.handlers = &rtems_filesystem_handlers_default;
+#endif /* __rtems__ */
 	fp->f_data = NULL;
 	funsetown(&cpipe->pipe_sigio);
 	pipeclose(cpipe);
@@ -1492,9 +1718,13 @@ pipe_free_kmem(cpipe)
 
 	if (cpipe->pipe_buffer.buffer != NULL) {
 		atomic_subtract_long(&amountpipekva, cpipe->pipe_buffer.size);
+#ifndef __rtems__
 		vm_map_remove(pipe_map,
 		    (vm_offset_t)cpipe->pipe_buffer.buffer,
 		    (vm_offset_t)cpipe->pipe_buffer.buffer + cpipe->pipe_buffer.size);
+#else /* __rtems__ */
+		free(cpipe->pipe_buffer.buffer, M_TEMP);
+#endif /* __rtems__ */
 		cpipe->pipe_buffer.buffer = NULL;
 	}
 #ifndef PIPE_NODIRECT
@@ -1626,6 +1856,15 @@ pipe_kqfilter(struct file *fp, struct knote *kn)
 	PIPE_UNLOCK(cpipe);
 	return (0);
 }
+#ifdef __rtems__
+int
+rtems_bsd_pipe_kqfilter(rtems_libio_t *iop, struct knote *kn)
+{
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+
+	return pipe_kqfilter(fp, kn);
+}
+#endif /* __rtems__ */
 
 static void
 filt_pipedetach(struct knote *kn)
@@ -1687,3 +1926,35 @@ filt_pipewrite(struct knote *kn, long hint)
 	PIPE_UNLOCK(rpipe);
 	return (kn->kn_data >= PIPE_BUF);
 }
+#ifdef __rtems__
+static int
+rtems_bsd_pipe_open(rtems_libio_t *iop, const char *path, int oflag,
+    mode_t mode)
+{
+	return rtems_bsd_error_to_status_and_errno(ENXIO);
+}
+
+static int
+rtems_bsd_pipe_close(rtems_libio_t *iop)
+{
+	struct file *fp = rtems_bsd_iop_to_fp(iop);
+	int error = pipe_close(fp, NULL);
+
+	return rtems_bsd_error_to_status_and_errno(error);
+}
+
+static int
+rtems_bsd_pipe_fcntl(rtems_libio_t *iop, int cmd)
+{
+	int error = 0;
+
+	if (cmd == F_SETFL) {
+		struct file *fp = rtems_bsd_iop_to_fp(iop);
+		int nbio = iop->flags & LIBIO_FLAGS_NO_DELAY;
+
+		error = pipe_ioctl(fp, FIONBIO, &nbio, NULL, NULL);
+	}
+
+	return rtems_bsd_error_to_status_and_errno(error);
+}
+#endif /* __rtems__ */
diff --git a/freebsd/sys/sys/pipe.h b/freebsd/sys/sys/pipe.h
index 444c8fbe..c59ecc75 100644..100755
--- a/freebsd/sys/sys/pipe.h
+++ b/freebsd/sys/sys/pipe.h
@@ -78,7 +78,9 @@ struct pipemapping {
 	vm_size_t	cnt;		/* number of chars in buffer */
 	vm_size_t	pos;		/* current position of transfer */
 	int		npages;		/* number of pages */
+#ifndef __rtems__
 	vm_page_t	ms[PIPENPAGES];	/* pages in source process */
+#endif /* __rtems__ */
 };
 
 /*
diff --git a/libbsd.py b/libbsd.py
index 36851919..d7a0a5f7 100755
--- a/libbsd.py
+++ b/libbsd.py
@@ -258,6 +258,7 @@ def base(mm):
             'sys/sys/_null.h',
             'sys/sys/osd.h',
             'sys/sys/pcpu.h',
+            'sys/sys/pipe.h',
             'sys/sys/priv.h',
             'sys/sys/proc.h',
             'sys/sys/protosw.h',
@@ -348,6 +349,7 @@ def base(mm):
             'sys/kern/subr_uio.c',
             'sys/kern/subr_unit.c',
             'sys/kern/sys_generic.c',
+            'sys/kern/sys_pipe.c',
             'sys/kern/uipc_accf.c',
             'sys/kern/uipc_domain.c',
             'sys/kern/uipc_mbuf2.c',
diff --git a/libbsd_waf.py b/libbsd_waf.py
index 7ea6a369..c440015f 100644
--- a/libbsd_waf.py
+++ b/libbsd_waf.py
@@ -836,6 +836,7 @@ def build(bld):
               'freebsd/sys/kern/subr_uio.c',
               'freebsd/sys/kern/subr_unit.c',
               'freebsd/sys/kern/sys_generic.c',
+              'freebsd/sys/kern/sys_pipe.c',
               'freebsd/sys/kern/sys_socket.c',
               'freebsd/sys/kern/uipc_accf.c',
               'freebsd/sys/kern/uipc_domain.c',
diff --git a/rtemsbsd/sys/fs/devfs/devfs_devs.c b/rtemsbsd/sys/fs/devfs/devfs_devs.c
index a38bb4d2..75c9e270 100644..100755
--- a/rtemsbsd/sys/fs/devfs/devfs_devs.c
+++ b/rtemsbsd/sys/fs/devfs/devfs_devs.c
@@ -33,6 +33,7 @@
 
 #include <sys/types.h>
 #include <sys/conf.h>
+#include <sys/kernel.h>
 #include <sys/file.h>
 #include <sys/malloc.h>
 #include <stdlib.h>
@@ -45,8 +46,12 @@
 
 #include <rtems/imfs.h>
 
+#define DEVFS_ROOTINO 2
+
 const char rtems_cdev_directory[] = RTEMS_CDEV_DIRECTORY;
 
+struct unrhdr *devfs_inos;
+
 static struct cdev *
 devfs_imfs_get_context_by_iop(rtems_libio_t *iop)
 {
@@ -325,3 +330,27 @@ devfs_dev_exists(const char *name)
 	else
 		return 0;
 }
+
+ino_t
+devfs_alloc_cdp_inode(void)
+{
+
+	return (alloc_unr(devfs_inos));
+}
+
+void
+devfs_free_cdp_inode(ino_t ino)
+{
+
+	if (ino > 0)
+		free_unr(devfs_inos, ino);
+}
+
+static void
+	devfs_devs_init(void *junk __unused)
+{
+
+	devfs_inos = new_unrhdr(DEVFS_ROOTINO + 1, INT_MAX, &devmtx);
+}
+
+SYSINIT(devfs_devs, SI_SUB_DEVFS, SI_ORDER_FIRST, devfs_devs_init, NULL);
diff --git a/testsuite/selectpollkqueue01/test_main.c b/testsuite/selectpollkqueue01/test_main.c
index 8f76049b..c45fff78 100644..100755
--- a/testsuite/selectpollkqueue01/test_main.c
+++ b/testsuite/selectpollkqueue01/test_main.c
@@ -55,7 +55,7 @@
 #include <rtems/libcsupport.h>
 #include <rtems.h>
 
-#define TEST_NAME "LIBBSD SELECT AND POLL AND KQUEUE 1"
+#define TEST_NAME "LIBBSD SELECT AND POLL AND KQUEUE AND PIPE 1"
 
 #define PRIO_MASTER 1
 
@@ -71,6 +71,8 @@
 
 #define EVENT_SHUTDOWN RTEMS_EVENT_4
 
+#define EVENT_CLOSE_PIPE RTEMS_EVENT_5
+
 #define BUF_SIZE 4096
 
 #define PORT 1234
@@ -88,6 +90,7 @@ typedef struct {
 	int afd;
 	int rfd;
 	int wfd;
+	int pfd[2];
 	struct sockaddr_in caddr;
 	rtems_id worker_task;
 } test_context;
@@ -173,6 +176,8 @@ worker_task(rtems_task_argument arg)
 		ssize_t n;
 		int rv;
 		int cfd = ctx->cfd;
+		int rfd = ctx->pfd[0];
+		int wfd = ctx->pfd[1];
 
 		sc = rtems_event_receive(
 			RTEMS_ALL_EVENTS,
@@ -236,6 +241,19 @@ worker_task(rtems_task_argument arg)
 			assert(rv == 0);
 		}
 
+		if ((events & EVENT_CLOSE_PIPE) != 0) {
+			puts("worker: close pipe");
+
+			ctx->pfd[0] = -1;
+			ctx->pfd[1] = -1;
+
+			rv = close(wfd);
+			assert(rv == 0);
+
+			rv = close(rfd);
+			assert(rv == 0);
+		}
+
 		if ((events & EVENT_SHUTDOWN) != 0) {
 			puts("worker: shutdown");
 
@@ -281,8 +299,13 @@ static void
 set_non_blocking(int fd, int enable)
 {
 	int rv;
+	int flags = fcntl(fd, F_GETFL, 0);
 
-	rv = ioctl(fd, FIONBIO, &enable);
+	if (enable) {
+		rv = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+	} else {
+		rv = fcntl(fd, F_SETFL, flags & ~O_NONBLOCK);
+	}
 	assert(rv == 0);
 }
 
@@ -1004,6 +1027,157 @@ test_kqueue_user(test_context *ctx)
 }
 
 static void
+test_pipe_timeout(test_context *ctx)
+{
+	struct pipe_poll_events
+	{
+		short event;
+		int rv;
+	};
+	const struct pipe_poll_events events[] = {
+		{ POLLIN, 0 },
+		{ POLLPRI, 0 },
+		{ POLLOUT, 1 },
+		{ POLLRDNORM, 0 },
+		{ POLLWRNORM, 1 },
+		{ POLLRDBAND, 0 },
+		{ POLLWRBAND, 0 },
+		{ POLLINIGNEOF, 0 }
+	};
+
+	int timeout = 100;
+	struct pollfd pfd;
+	size_t i;
+	int rv;
+
+	puts("test pipe timeout");
+
+	rv = pipe(ctx->pfd);
+	assert(rv == 0);
+
+	pfd.fd = ctx->pfd[1];
+
+	for (i = 0; i < nitems(events); ++i) {
+		int rv;
+
+		pfd.events = events[i].event;
+		pfd.revents = 0;
+
+		rv = poll(&pfd, 1, timeout);
+		assert(rv == events[i].rv);
+	}
+}
+
+static void
+test_pipe_read(test_context *ctx)
+{
+	int rfd = ctx->pfd[0];
+	int wfd = ctx->pfd[1];
+	struct pollfd pfd = {
+		.fd = rfd,
+		.events = POLLIN
+	};
+	int timeout = -1;
+	int rv;
+	ssize_t n;
+
+	puts("test pipe read");
+
+	assert(rfd >= 0);
+	assert(wfd >= 0);
+
+	ctx->wfd = wfd;
+	ctx->wbuf = &msg[0];
+	ctx->wn = sizeof(msg);
+	send_events(ctx, EVENT_WRITE);
+
+	set_non_blocking(rfd, 1);
+
+	errno = 0;
+	n = read(rfd, &ctx->buf[0], sizeof(ctx->buf));
+	assert(n == -1);
+	assert(errno == EAGAIN);
+
+	rv = poll(&pfd, 1, timeout);
+	assert(rv == 1);
+	assert(pfd.revents == POLLIN);
+
+	n = read(rfd, &ctx->buf[0], sizeof(ctx->buf));
+	assert(n == (ssize_t) sizeof(msg));
+	assert(memcmp(&msg[0], &ctx->buf[0], sizeof(msg)) == 0);
+}
+
+static void
+test_pipe_write(test_context *ctx)
+{
+	int rfd = ctx->pfd[0];
+	int wfd = ctx->pfd[1];
+	struct pollfd pfd = {
+		.fd = wfd,
+		.events = POLLOUT
+	};
+	int timeout = -1;
+	int rv;
+	ssize_t n;
+
+	puts("test pipe write");
+
+	assert(rfd >= 0);
+	assert(wfd >= 0);
+
+	ctx->rfd = rfd;
+	ctx->rbuf = &ctx->buf[0];
+	ctx->rn = sizeof(ctx->buf);
+	send_events(ctx, EVENT_READ);
+
+	set_non_blocking(wfd, 1);
+
+	do {
+		errno = 0;
+		n = write(wfd, &ctx->buf[0], sizeof(ctx->buf));
+		if (n == -1) {
+			assert(errno == EAGAIN);
+		}
+	} while (n > 0);
+
+	rv = poll(&pfd, 1, timeout);
+	assert(rv == 1);
+	assert(pfd.revents == POLLOUT);
+}
+
+static void
+test_pipe_close(test_context *ctx)
+{
+	int rfd = ctx->pfd[0];
+	int wfd = ctx->pfd[1];
+	struct pollfd pfd = {
+		.fd = rfd,
+		.events = POLLIN
+	};
+	int timeout = -1;
+	int rv;
+
+	puts("test pipe close");
+
+	assert(ctx->pfd[0] >= 0);
+	assert(ctx->pfd[1] >= 0);
+
+	send_events(ctx, EVENT_CLOSE_PIPE);
+
+	set_non_blocking(rfd, 0);
+
+	assert(ctx->pfd[0] >= 0);
+	assert(ctx->pfd[1] >= 0);
+
+	rv = poll(&pfd, 1, timeout);
+	assert(rv == 1);
+	assert(pfd.revents == (POLLIN | POLLHUP));
+
+	assert(ctx->pfd[0] == -1);
+	assert(ctx->pfd[1] == -1);
+}
+
+static void
 test_main(void)
 {
 	test_context *ctx = &test_instance;
@@ -1034,6 +1208,11 @@ test_main(void)
 	test_kqueue_close(ctx);
 	test_kqueue_user(ctx);
 
+	test_pipe_timeout(ctx);
+	test_pipe_read(ctx);
+	test_pipe_write(ctx);
+	test_pipe_close(ctx);
+
 	exit(0);
 }