From 73295e7acc1e5ec3ee8814fd4ccc45f7cee63392 Mon Sep 17 00:00:00 2001
From: user <user@clank>
Date: Fri, 17 Jul 2026 01:04:16 +0200
Subject: kernel+sh: real anonymous pipes (concurrent, streaming, SIGPIPE)

Replace the shell's temp-file pipe hack with proper in-kernel FIFOs.

Kernel (src/pipe.asm, new):
- A small pool of bounded ring buffers (PIPE_MAX=2, 64 B each) with
  writers/readers refcounts. Pipe fds are $F0+idx*2 (read) / +1 (write);
  $FF stays "console".
- SYS_PIPE allocates one (writers=readers=1) and returns the read fd
  (write = read+1). getb/putb/close dispatch pipe fds here; sys_exit drops
  the refcounts held as PROC_STDIN/PROC_STDOUT.
- Blocking with SchedYield, which is the flow control: read blocks while
  empty with a writer (EOF once writers hit 0), write blocks while full
  with a reader, and if the last reader is gone the writer is killed
  (SIGPIPE -> exit 141). Cooperative-scheduler friendly.

Shell (c/sh.c):
- run_pipeline(): split on '|', make a pipe between adjacent stages, and
  fork ALL stages concurrently (no wait between), wiring stdin/stdout;
  then wait for all. Per-stage >/< still honored; orphaned pipe ends are
  closed on a lookup miss so EOF/EPIPE propagate.
- Drop the __pipe temp file and its 2 KB / serialized limits.

libc: pipe(). New c/ptest.c exercises the FIFO (write, read back, EOF).

Now works (old version couldn't): multi-stage a|b|c; streaming beyond 2 KB
(count 120 | wc = 3372 bytes through a 64 B buffer); early-exit SIGPIPE
(count 200 | true kills count instead of hanging/overflowing a file).
---
 c/sh.c | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++++++-----------
 1 file changed, 57 insertions(+), 11 deletions(-)

(limited to 'c/sh.c')

diff --git a/c/sh.c b/c/sh.c
index df2c5e9..a8dd09d 100644
--- a/c/sh.c
+++ b/c/sh.c
@@ -95,6 +95,61 @@ done:
     if (myout != NOFD) close(myout);
 }
 
+/* run a multi-stage pipeline (a | b | c): wire stages together with real
+   kernel pipes, launch them ALL concurrently, then wait for them. */
+static void run_pipeline(char **tok, unsigned char nt) {
+    unsigned char pids[8], npid = 0;
+    unsigned char cfd[8], ncfd = 0;      /* file-redirect fds; closed after all exit */
+    unsigned char start = 0, prev_r = NOFD;
+    for (;;) {
+        unsigned char end = start, last, i, cmdend, fin = NOFD, fout = NOFD;
+        unsigned char my_out = NOFD, next_r = NOFD, id, pid;
+        while (end < nt && !isop(tok[end], '|')) end++;
+        last = (end >= nt);
+        if (!last) {                     /* pipe from this stage to the next */
+            unsigned char pr = pipe();
+            if (pr != NOFD) { next_r = pr; my_out = pr + 1; }
+        }
+        cmdend = end;                    /* this stage's own >/< redirects */
+        for (i = start; i < end; i++) {
+            if (isop(tok[i], '>')) { if (i + 1 < end) fout = open(tok[i+1], O_WRITE); if (i < cmdend) cmdend = i; }
+            else if (isop(tok[i], '<')) { if (i + 1 < end) fin = open(tok[i+1], O_READ); if (i < cmdend) cmdend = i; }
+        }
+        id = lookup(tok[start]);
+        if (id == NOFD) {
+            puts(tok[start]); puts(": not found"); nl();
+            if (prev_r != NOFD) close(prev_r);   /* orphaned pipe ends -> EOF/EPIPE */
+            if (my_out != NOFD) close(my_out);
+            if (fin != NOFD) close(fin);
+            if (fout != NOFD) close(fout);
+        } else {
+            set_cmdline(tok, start, cmdend);
+            pid = fork();
+            if (pid == 0) {
+                if (fin != NOFD) setin(fin); else if (prev_r != NOFD) setin(prev_r);
+                if (fout != NOFD) setout(fout); else if (my_out != NOFD) setout(my_out);
+                exec(id);
+            }
+            if (npid < 8) pids[npid++] = pid;
+            if (fin != NOFD && ncfd < 8) cfd[ncfd++] = fin;
+            if (fout != NOFD && ncfd < 8) cfd[ncfd++] = fout;
+        }
+        prev_r = next_r;
+        if (last) break;
+        start = end + 1;
+    }
+    { unsigned char got = 0;             /* wait for all stages */
+      while (got < npid) {
+          unsigned char r = wait(), j, mine = 0;
+          if (r == NOFD) break;
+          for (j = 0; j < npid; j++) if (pids[j] == r) { mine = 1; break; }
+          if (mine) got++;
+          else { putc('['); putu(r); puts(" done]"); nl(); }
+      }
+    }
+    { unsigned char k; for (k = 0; k < ncfd; k++) close(cfd[k]); }
+}
+
 void main(void) {
     char line[80];
     char *tok[16];
@@ -133,16 +188,7 @@ void main(void) {
         pipepos = 0;
         for (i = 1; i < nt; i++)
             if (isop(tok[i], '|')) { pipepos = i; break; }
-        if (pipepos) {
-            unsigned char pw = open("__pipe", O_WRITE);
-            run(tok, 0, pipepos, NOFD, pw, 0);
-            close(pw);
-            { unsigned char pr = open("__pipe", O_READ);
-              run(tok, pipepos + 1, nt, pr, NOFD, 0);
-              close(pr); }
-            fremove("__pipe");
-        } else {
-            run(tok, 0, nt, NOFD, NOFD, bg);
-        }
+        if (pipepos) run_pipeline(tok, nt);
+        else run(tok, 0, nt, NOFD, NOFD, bg);
     }
 }
-- 
cgit v1.3.1-sl0p