syscalls/pipe: turn into a POSIX-style api with separate rw ends

Without separate read/write ends you can't tell when there are no more writers left if you have multiple readers. Consider this piece of code:
	int fd = pipe();
	fork(); // execution continues in 2 processes
	while (read(fd, &some_buf, sizeof somebuf) >= 0) {
		...
	}

Once both processes call `read()`, it's obvious that no writes are possible - all the processes that hold a reference to the pipe are currently stuck on a `read()` call, so the kernel could just make it return an error in both. But, what then? It's still possible to write to the pipe, and you can't know if the other process will do that. Thus, if you don't want to miss any output, you have to keep reading the pipe. Forever. Both processes end up stuck.

Having separate read/write ends prevents that.

9 files changed, 125 insertions, 38 deletions

diff --git a/src/init/syscalls.c b/src/init/syscalls.c
index 14c0d52..25fa134 100644
--- a/src/init/syscalls.c
+++ b/src/init/syscalls.c
@@ -50,8 +50,8 @@ void __user *_syscall_memflag(void __user *addr, size_t len, int flags) {
 	return (void __user *)_syscall(_SYSCALL_MEMFLAG, (int)addr, (int)len, flags, 0);
 }
 
-handle_t _syscall_pipe(int flags) {
-	return (handle_t)_syscall(_SYSCALL_PIPE, flags, 0, 0, 0);
+int _syscall_pipe(handle_t __user user_ends[2], int flags) {
+	return _syscall(_SYSCALL_PIPE, (int)user_ends, flags, 0, 0);
 }
 
 void _syscall_debug_klog(const void __user *buf, size_t len) {
diff --git a/src/init/tests/pipe.c b/src/init/tests/pipe.c
index bbfe79a..697b9b1 100644
--- a/src/init/tests/pipe.c
+++ b/src/init/tests/pipe.c
@@ -6,36 +6,75 @@
 
 static const char *pipe_msgs[2] = {"hello", "world"};
 
-static void test_pipe_child(handle_t pipe) {
-	int ret = _syscall_write(pipe, pipe_msgs[0], 5, -1);
+static void test_pipe_child(handle_t ends[2]) {
+	int ret = _syscall_write(ends[1], pipe_msgs[0], 5, -1);
 	assert(ret == 5);
 
-	ret = _syscall_write(pipe, pipe_msgs[1], 5, -1);
+	ret = _syscall_write(ends[1], pipe_msgs[1], 5, -1);
 	assert(ret == 5);
 }
 
-static void test_pipe_parent(handle_t pipe) {
+static void test_pipe_parent(handle_t ends[2]) {
 	char buf[16];
-	int ret = _syscall_read(pipe, buf, 16, 0);
+	int ret = _syscall_read(ends[0], buf, 16, 0);
 	assert(ret == 5);
 	assert(!memcmp(buf, pipe_msgs[0], 5));
 
-	_syscall_read(pipe, buf, 16, 0);
+	_syscall_read(ends[0], buf, 16, 0);
 	assert(ret == 5);
-	assert(!memcmp(buf, pipe_msgs[1], 5)); // wrong compare for test
+	assert(!memcmp(buf, pipe_msgs[1], 5));
+
+	// TODO these calls fail for multiple reasons at once - split
+	assert(_syscall_read(ends[1], buf, 16, 0) < 0);
+	assert(_syscall_write(ends[0], buf, 16, 0) < 0);
 }
 
 void test_pipe(void) {
-	handle_t pipe = _syscall_pipe(0);
-	assert(pipe > 0);
+	handle_t ends[2];
+	char buf[16];
+	assert(_syscall_pipe(ends, 0) >= 0);
+
+	if (!_syscall_fork(0, NULL)) {
+		test_pipe_child(ends);
+		_syscall_exit(0);
+	} else {
+		test_pipe_parent(ends);
+		_syscall_await();
+	}
+
+	_syscall_close(ends[0]);
+	_syscall_close(ends[1]);
+
+	assert(_syscall_pipe(ends, 0) >= 0);
+	_syscall_close(ends[0]);
+	assert(_syscall_write(ends[1], buf, 16, 0) < 0);
+	_syscall_close(ends[1]);
+
+	assert(_syscall_pipe(ends, 0) >= 0);
+	_syscall_close(ends[1]);
+	assert(_syscall_read(ends[0], buf, 16, 0) < 0);
+	_syscall_close(ends[0]);
+
+
+	assert(_syscall_pipe(ends, 0) >= 0);
+	if (!_syscall_fork(0, NULL)) {
+		_syscall_exit(0);
+	} else {
+		_syscall_close(ends[1]);
+		assert(_syscall_read(ends[0], buf, 16, 0) < 0);
+		_syscall_await();
+	}
 
+	assert(_syscall_pipe(ends, 0) >= 0);
 	if (!_syscall_fork(0, NULL)) {
-		test_pipe_child(pipe);
 		_syscall_exit(0);
 	} else {
-		test_pipe_parent(pipe);
+		_syscall_close(ends[1]);
+		assert(_syscall_write(ends[1], buf, 16, 0) < 0);
 		_syscall_await();
 	}
 
+	// TODO detect when all processes that can read are stuck on writing to the pipe and vice verse
 	// TODO kill process that's waiting on a pipe
+	// TODO queue
 }
diff --git a/src/kernel/handle.c b/src/kernel/handle.c
index 70830ff..ed7896d 100644
--- a/src/kernel/handle.c
+++ b/src/kernel/handle.c
@@ -1,6 +1,7 @@
 #include <kernel/handle.h>
 #include <kernel/mem/alloc.h>
 #include <kernel/panic.h>
+#include <kernel/pipe.h>
 #include <kernel/proc.h>
 #include <kernel/vfs/request.h>
 #include <shared/mem.h>
@@ -25,6 +26,12 @@ void handle_close(struct handle *h) {
 				.caller = NULL,
 				.backend = h->backend,
 			});
+	} else if (h->type == HANDLE_PIPE) {
+		assert(!h->pipe.queued);
+		if (h->pipe.sister) {
+			pipe_invalidate_end(h->pipe.sister);
+			h->pipe.sister->pipe.sister = NULL;
+		}
 	}
 
 	if (h->backend)
diff --git a/src/kernel/handle.h b/src/kernel/handle.h
index 4a5cab2..be8250b 100644
--- a/src/kernel/handle.h
+++ b/src/kernel/handle.h
@@ -18,7 +18,9 @@ struct handle {
 	struct vfs_backend *backend; // HANDLE_FILE | HANDLE_FS_FRONT
 	void __user *file_id; // only applicable to HANDLE_FILE
 	struct {
-		struct process *reader, *writer;
+		struct process *queued;
+		bool write_end;
+		struct handle *sister; // the other end, not included in refcount
 	} pipe;
 
 	size_t refcount;
diff --git a/src/kernel/pipe.c b/src/kernel/pipe.c
index 8afd359..34cb263 100644
--- a/src/kernel/pipe.c
+++ b/src/kernel/pipe.c
@@ -3,29 +3,41 @@
 #include <kernel/pipe.h>
 #include <kernel/util.h>
 
-void pipe_joinqueue(struct handle *h, bool wants_write,
+bool pipe_joinqueue(struct handle *h, bool wants_write,
 		struct process *proc, void __user *pbuf, size_t pbuflen)
 {
-	struct process **slot = wants_write ? &h->pipe.reader : &h->pipe.writer;
-	if (*slot) {
-		assert((*slot)->state == PS_WAITS4PIPE);
+	assert(h && h->type == HANDLE_PIPE);
+	if (wants_write == h->pipe.write_end) return false;
+	if (!h->pipe.sister) return false;
+	if (h->pipe.queued) {
+		assert(h->pipe.queued->state == PS_WAITS4PIPE);
 		panic_unimplemented();
 	}
 
 	process_transition(proc, PS_WAITS4PIPE);
-	*slot = proc;
+	h->pipe.queued = proc;
 	proc->waits4pipe.pipe = h;
 	proc->waits4pipe.buf = pbuf;
 	proc->waits4pipe.len = pbuflen;
+	return true;
 }
 
 void pipe_trytransfer(struct handle *h) {
-	struct process *rdr = h->pipe.reader, *wtr = h->pipe.writer;
+	struct process *rdr, *wtr;
 	int len;
+	assert(h);
+	if (!h->pipe.sister) {
+		assert(!h->pipe.queued);
+		return;
+	}
+
+	rdr = h->pipe.write_end ? h->pipe.sister->pipe.queued : h->pipe.queued;
+	wtr = h->pipe.write_end ? h->pipe.queued : h->pipe.sister->pipe.queued;
+
 	if (!(rdr && wtr)) return;
 	assert(rdr->state == PS_WAITS4PIPE);
 	assert(wtr->state == PS_WAITS4PIPE);
-	
+
 	len = min(rdr->waits4pipe.len, wtr->waits4pipe.len);
 
 	if (!virt_cpy(
@@ -36,8 +48,17 @@ void pipe_trytransfer(struct handle *h) {
 	}
 	process_transition(rdr, PS_RUNNING);
 	process_transition(wtr, PS_RUNNING);
-	h->pipe.reader = NULL;
-	h->pipe.writer = NULL;
+	h->pipe.queued = NULL;
+	h->pipe.sister->pipe.queued = NULL;
 	regs_savereturn(&rdr->regs, len);
 	regs_savereturn(&wtr->regs, len);
 }
+
+void pipe_invalidate_end(struct handle *h) {
+	struct process *p = h->pipe.queued;
+	if (p) {
+		process_transition(p, PS_RUNNING);
+		regs_savereturn(&p->regs, -1);
+	}
+	h->pipe.queued = NULL;
+}
diff --git a/src/kernel/pipe.h b/src/kernel/pipe.h
index facaaaa..9040ab1 100644
--- a/src/kernel/pipe.h
+++ b/src/kernel/pipe.h
@@ -1,7 +1,11 @@
 #pragma once
 #include <kernel/proc.h>
+#include <stdbool.h>
 
-void pipe_joinqueue(struct handle *h, bool wants_write,
+/* returns false on failure */
+bool pipe_joinqueue(struct handle *h, bool wants_write,
 		struct process *proc, void __user *pbuf, size_t pbuflen);
 
 void pipe_trytransfer(struct handle *h);
+
+void pipe_invalidate_end(struct handle *h);
diff --git a/src/kernel/syscalls.c b/src/kernel/syscalls.c
index c916e81..421b5ce 100644
--- a/src/kernel/syscalls.c
+++ b/src/kernel/syscalls.c
@@ -194,8 +194,10 @@ int _syscall_read(handle_t handle_num, void __user *buf, size_t len, int offset)
 				});
 			break;
 		case HANDLE_PIPE:
-			pipe_joinqueue(h, true, process_current, buf, len);
-			pipe_trytransfer(h);
+			if (pipe_joinqueue(h, true, process_current, buf, len))
+				pipe_trytransfer(h);
+			else
+				SYSCALL_RETURN(-1);
 			break;
 		default:
 			SYSCALL_RETURN(-1);
@@ -221,8 +223,10 @@ int _syscall_write(handle_t handle_num, const void __user *buf, size_t len, int
 				});
 			break;
 		case HANDLE_PIPE:
-			pipe_joinqueue(h, false, process_current, (void __user *)buf, len);
-			pipe_trytransfer(h);
+			if (pipe_joinqueue(h, false, process_current, (void __user *)buf, len))
+				pipe_trytransfer(h);
+			else
+				SYSCALL_RETURN(-1);
 			break;
 		default:
 			SYSCALL_RETURN(-1);
@@ -310,15 +314,24 @@ void __user *_syscall_memflag(void __user *addr, size_t len, int flags) {
 	SYSCALL_RETURN((uintptr_t)addr);
 }
 
-handle_t _syscall_pipe(int flags) {
-	if (flags) return -1;
+int _syscall_pipe(handle_t __user user_ends[2], int flags) {
+	if (flags) SYSCALL_RETURN(-1);
+	handle_t ends[2];
+	struct handle *rend, *wend;
 
-	handle_t h = process_find_free_handle(process_current, 0);
-	if (h < 0) return -1;
-	process_current->handles[h] = handle_init(HANDLE_PIPE);
-	assert(process_current->handles[h]->pipe.reader == NULL);
-	assert(process_current->handles[h]->pipe.writer == NULL);
-	SYSCALL_RETURN(h);
+	ends[0] = process_find_free_handle(process_current, 0);
+	if (ends[0] < 0) SYSCALL_RETURN(-1);
+	ends[1] = process_find_free_handle(process_current, ends[0]+1);
+	if (ends[1] < 0) SYSCALL_RETURN(-1);
+
+	rend = process_current->handles[ends[0]] = handle_init(HANDLE_PIPE);
+	wend = process_current->handles[ends[1]] = handle_init(HANDLE_PIPE);
+	wend->pipe.write_end = true;
+	wend->pipe.sister = rend;
+	rend->pipe.sister = wend;
+
+	virt_cpy_to(process_current->pages, user_ends, ends, sizeof ends);
+	SYSCALL_RETURN(0);
 }
 
 void _syscall_debug_klog(const void __user *buf, size_t len) {
@@ -366,7 +379,7 @@ int _syscall(int num, int a, int b, int c, int d) {
 			_syscall_memflag((userptr_t)a, b, c);
 			break;
 		case _SYSCALL_PIPE:
-			_syscall_pipe(a);
+			_syscall_pipe((userptr_t)a, b);
 			break;
 		case _SYSCALL_DEBUG_KLOG:
 			_syscall_debug_klog((userptr_t)a, b);
diff --git a/src/shared/syscalls.h b/src/shared/syscalls.h
index c6d3c33..f19f05a 100644
--- a/src/shared/syscalls.h
+++ b/src/shared/syscalls.h
@@ -82,6 +82,6 @@ int _syscall_fs_respond(void __user *buf, int ret, int flags);
  * @return address of the first affected page (usually == addr)
  */
 void __user *_syscall_memflag(void __user *addr, size_t len, int flags);
-handle_t _syscall_pipe(int flags);
+int _syscall_pipe(handle_t __user user_ends[2], int flags);
 
 void _syscall_debug_klog(const void __user *buf, size_t len);
diff --git a/tools/syscall_wrappers.awk b/tools/syscall_wrappers.awk
index 12cc233..4a295f2 100644
--- a/tools/syscall_wrappers.awk
+++ b/tools/syscall_wrappers.awk
@@ -19,6 +19,7 @@ BEGIN {
 	sub(/ *$/, "", rets)
 
 	params = substr($0, match($0, /\(.+\)/) + 1, RLENGTH - 2);
+	gsub(/\[[^\]]\]/, "", params);
 	if (params == "void") params = ""
 
 	split(params, p, /,/);