pipefs: Enable pipes to notify readiness

The poll method of PIPEFS receives a notification callback from the
caller. PIPEFS notifies the caller when the fd becomes readable using that
callback.

Signed-off-by: Kohei Tokunaga <ktokunaga.mail@gmail.com>

Author: ktock

src/lib/libpipefs.js

@@ -21,6 +21,23 @@ addToLibrary({
         // able to read from the read end after write end is closed.
         refcnt : 2,
         timestamp: new Date(),
+#if PTHREADS
+        readableHandlers: [],
+        registerReadableHandler: (callback) => {
+          callback.registerCleanupFunc(() => {
+            const i = pipe.readableHandlers.indexOf(callback);
+            if (i !== -1) pipe.readableHandlers.splice(i, 1);
+          });
+          pipe.readableHandlers.push(callback);
+        },
+        notifyReadableHandlers: () => {
+          while (pipe.readableHandlers.length > 0) {
+            const cb = pipe.readableHandlers.shift();
+            if (cb) cb({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
+          }
+          pipe.readableHandlers = [];
+        }
+#endif
       };
 
       pipe.buckets.push({
@@ -80,7 +97,7 @@ addToLibrary({
           blocks: 0,
         };
       },
-      poll(stream) {
+      poll(stream, timeout, notifyCallback) {
         var pipe = stream.node.pipe;
 
         if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
@@ -92,6 +109,9 @@ addToLibrary({
           }
         }
 
+#if PTHREADS
+        if (notifyCallback) pipe.registerReadableHandler(notifyCallback);
+#endif
         return 0;
       },
       dup(stream) {
@@ -204,6 +224,9 @@ addToLibrary({
         if (freeBytesInCurrBuffer >= dataLen) {
           currBucket.buffer.set(data, currBucket.offset);
           currBucket.offset += dataLen;
+#if PTHREADS
+          pipe.notifyReadableHandlers();
+#endif
           return dataLen;
         } else if (freeBytesInCurrBuffer > 0) {
           currBucket.buffer.set(data.subarray(0, freeBytesInCurrBuffer), currBucket.offset);
@@ -235,6 +258,9 @@ addToLibrary({
           newBucket.buffer.set(data);
         }
 
+#if PTHREADS
+        pipe.notifyReadableHandlers();
+#endif
         return dataLen;
       },
       close(stream) {