feat: extend pipe with pipeWrite and pipeHasData

src/pipe.c

@@ -159,3 +159,133 @@ int64 pipeRead(int handle, void* buffer, uint64 length)
     if (p->writers == 0 && p->everHadWriter)
         return 0;
 
+    return -1;
+}
+
+int64 pipeWrite(int handle, const void* buffer, uint64 length)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return -1;
+    /* Symmetric with pipeRead's EOF check above: only a reader that WAS
+       attached and left counts as a broken pipe. */
+    if (p->readers == 0 && p->everHadReader)
+        return -1;
+    if (length == 0)
+        return 0;
+
+    if (ringIsFull(&p->ring))
+        return -1;
+
+    usize put = ringWrite(&p->ring, (const uint8*)buffer, (usize)length);
+    /* Writing adds ring data: any reader blocked on "buffer empty" may now
+       be able to proceed, so give it a chance to recheck. */
+    schedWakeChannel((const void*)p);
+    schedWakeChannel(schedPollChannel());
+    return (int64)put;
+}
+
+bool pipeHasData(int handle)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return false;
+    return !ringIsEmpty(&p->ring) || (p->writers == 0 && p->everHadWriter);
+}
+
+bool pipeHasSpace(int handle)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return false;
+    return !ringIsFull(&p->ring);
+}
+
+int64 pipeReadBlocking(int handle, void* buffer, uint64 length)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return -1;
+
+    for (;;) {
+        int64 result = pipeRead(handle, buffer, length);
+        if (result != -1)
+            return result;
+
+        schedBlockCurrent((const void*)p);
+        halt();
+    }
+}
+
+int64 pipeWriteBlocking(int handle, const void* buffer, uint64 length)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return -1;
+
+    for (;;) {
+        int64 result = pipeWrite(handle, buffer, length);
+        if (result != -1)
+            return result;
+
+        /* A reader that attached and then left is a broken pipe: a terminal
+           condition no amount of waiting will ever resolve, unlike a merely
+           full buffer (or a pipeCreateDetached handle whose reader simply
+           has not attached yet). */
+        if (pipeWriteBroken(handle))
+            return -1;
+
+        schedBlockCurrent((const void*)p);
+        halt();
+    }
+}
+
+static void pipeReleaseIfUnused(int handle)
+{
+    Pipe* p = &pipes[handle];
+    if (p->readers == 0 && p->writers == 0)
+        p->used = false;
+}
+
+void pipeCloseRead(int handle)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return;
+    if (p->readers > 0)
+        p->readers--;
+    /* A writer parked on "buffer full" must be woken to discover the pipe is
+       now broken instead of waiting forever for space that will never free
+       up on its own. */
+    if (p->readers == 0) {
+        schedWakeChannel((const void*)p);
+        schedWakeChannel(schedPollChannel());
+    }
+    pipeReleaseIfUnused(handle);
+}
+
+void pipeCloseWrite(int handle)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p)
+        return;
+    if (p->writers > 0)
+        p->writers--;
+    /* A reader parked on "buffer empty" must be woken to discover true
+       end-of-file instead of waiting forever for data that will never
+       arrive. */
+    if (p->writers == 0) {
+        schedWakeChannel((const void*)p);
+        schedWakeChannel(schedPollChannel());
+    }
+    pipeReleaseIfUnused(handle);
+}
+
+bool pipeDestroyIfUnattached(int handle)
+{
+    Pipe* p = pipeAt(handle);
+    if (!p || p->readers != 0 || p->writers != 0)
+        return false;
+    p->used = false;
+    return true;
+}