From fff36661c92f30685e2d83825a11b67ad8921a0e Mon Sep 17 00:00:00 2001
From: dzwdz
Date: Sat, 9 Jul 2022 20:47:13 +0200
Subject: kernel/pipes: process queueing

---
 src/init/tests/pipe.c | 66 +++++++++++++++++++++++++++++++++++++++++----------
 src/kernel/pipe.c     | 19 +++++++++------
 src/kernel/proc.c     |  3 +++
 src/kernel/proc.h     |  1 +
 4 files changed, 69 insertions(+), 20 deletions(-)

(limited to 'src')

diff --git a/src/init/tests/pipe.c b/src/init/tests/pipe.c
index e03c485..a980cb8 100644
--- a/src/init/tests/pipe.c
+++ b/src/init/tests/pipe.c
@@ -51,28 +51,68 @@ void test_pipe(void) {
 
 	/* writing to pipes with one end closed */
 	assert(_syscall_pipe(ends, 0) >= 0);
-	if (!_syscall_fork(0, NULL)) {
-		_syscall_close(ends[1]);
-		assert(_syscall_read(ends[0], buf, 16, 0) < 0);
-		_syscall_exit(0);
-	} else {
-		_syscall_close(ends[1]);
+	for (int i = 0; i < 16; i++) {
+		if (!_syscall_fork(0, NULL)) {
+			_syscall_close(ends[1]);
+			assert(_syscall_read(ends[0], buf, 16, 0) < 0);
+			_syscall_exit(0);
+		}
+	}
+	_syscall_close(ends[1]);
+	for (int i = 0; i < 16; i++)
 		_syscall_await();
+	_syscall_close(ends[0]);
+
+	assert(_syscall_pipe(ends, 0) >= 0);
+	for (int i = 0; i < 16; i++) {
+		if (!_syscall_fork(0, NULL)) {
+			_syscall_close(ends[0]);
+			assert(_syscall_write(ends[1], buf, 16, 0) < 0);
+			_syscall_exit(0);
+		}
 	}
+	_syscall_close(ends[0]);
+	for (int i = 0; i < 16; i++)
+		_syscall_await();
+	_syscall_close(ends[1]);
+
 
+	/* queueing */
 	assert(_syscall_pipe(ends, 0) >= 0);
-	if (!_syscall_fork(0, NULL)) {
-		_syscall_close(ends[0]);
-		assert(_syscall_write(ends[1], buf, 16, 0) < 0);
-		_syscall_exit(0);
-	} else {
-		_syscall_close(ends[0]);
+	for (int i = 0; i < 16; i++) {
+		if (!_syscall_fork(0, NULL)) {
+			assert(_syscall_write(ends[1], pipe_msgs[0], 5, -1) == 5);
+			_syscall_exit(0);
+		}
+	}
+	_syscall_close(ends[1]);
+	for (int i = 0; i < 16; i++) {
+		assert(_syscall_read(ends[0], buf, sizeof buf, 0) == 5);
 		_syscall_await();
 	}
+	assert(_syscall_read(ends[0], buf, sizeof buf, 0) < 0);
+	_syscall_close(ends[0]);
+
+	assert(_syscall_pipe(ends, 0) >= 0);
+	for (int i = 0; i < 16; i++) {
+		if (!_syscall_fork(0, NULL)) {
+			memset(buf, 0, sizeof buf);
+			assert(_syscall_read(ends[0], buf, 5, -1) == 5);
+			assert(!memcmp(buf, pipe_msgs[1], 5));
+			_syscall_exit(0);
+		}
+	}
+	_syscall_close(ends[0]);
+	for (int i = 0; i < 16; i++) {
+		assert(_syscall_write(ends[1], pipe_msgs[1], 5, -1) == 5);
+		_syscall_await();
+	}
+	assert(_syscall_write(ends[1], pipe_msgs[1], 5, -1) < 0);
+	_syscall_close(ends[1]);
+
 
 	// not a to.do detect when all processes that can read are stuck on writing to the pipe and vice versa
 	// it seems like linux just lets the process hang endlessly.
 
 	// TODO kill process that's waiting on a pipe
-	// TODO queue
 }
diff --git a/src/kernel/pipe.c b/src/kernel/pipe.c
index 34cb263..edd1388 100644
--- a/src/kernel/pipe.c
+++ b/src/kernel/pipe.c
@@ -9,16 +9,19 @@ bool pipe_joinqueue(struct handle *h, bool wants_write,
 	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();
+
+	struct process **slot = &h->pipe.queued;
+	while (*slot) {
+		assert((*slot)->state == PS_WAITS4PIPE);
+		slot = &((*slot)->waits4pipe.next);
 	}
 
 	process_transition(proc, PS_WAITS4PIPE);
-	h->pipe.queued = proc;
+	*slot = proc;
 	proc->waits4pipe.pipe = h;
 	proc->waits4pipe.buf = pbuf;
 	proc->waits4pipe.len = pbuflen;
+	proc->waits4pipe.next = NULL;
 	return true;
 }
 
@@ -46,19 +49,21 @@ void pipe_trytransfer(struct handle *h) {
 	{
 		panic_unimplemented();
 	}
+	h->pipe.queued = h->pipe.queued->waits4pipe.next;
+	h->pipe.sister->pipe.queued = h->pipe.sister->pipe.queued->waits4pipe.next;
 	process_transition(rdr, PS_RUNNING);
 	process_transition(wtr, PS_RUNNING);
-	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) {
+	while (p) {
+		assert(p->state == PS_WAITS4PIPE);
 		process_transition(p, PS_RUNNING);
 		regs_savereturn(&p->regs, -1);
+		p = p->waits4pipe.next;
 	}
 	h->pipe.queued = NULL;
 }
diff --git a/src/kernel/proc.c b/src/kernel/proc.c
index c69cb14..86170aa 100644
--- a/src/kernel/proc.c
+++ b/src/kernel/proc.c
@@ -116,6 +116,9 @@ void process_kill(struct process *p, int ret) {
 		if (p->state == PS_WAITS4FS)
 			p->waits4fs.req->caller = NULL;
 
+		if (p->state == PS_WAITS4PIPE)
+			panic_unimplemented();
+
 		for (handle_t h = 0; h < HANDLE_MAX; h++)
 			handle_close(p->handles[h]);
 
diff --git a/src/kernel/proc.h b/src/kernel/proc.h
index 20585bc..c400f7d 100644
--- a/src/kernel/proc.h
+++ b/src/kernel/proc.h
@@ -47,6 +47,7 @@ struct process {
 			struct handle *pipe;
 			char __user *buf;
 			size_t len;
+			struct process *next;
 		} waits4pipe;
 	};
 	struct vfs_request *handled_req;
-- 
cgit v1.2.3