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,19 @@ addToLibrary({
         // able to read from the read end after write end is closed.
         refcnt : 2,
         timestamp: new Date(),
+        readableHandlers: [],
+        registerReadableHanlders: (notifyCallback) => {
+          if (notifyCallback == null) return;
+          notifyCallback.registerCleanupFunc(() => {
+            const i = pipe.readableHandlers.indexOf(notifyCallback);
+            if (i !== -1) pipe.readableHandlers.splice(i, 1);
+          });
+          pipe.readableHandlers.push(notifyCallback);
+        },
+        notifyReadableHanders: () => {
+          pipe.readableHandlers.forEach(cb => cb({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}}));
+          pipe.readableHandlers = [];
+        }
       };
 
       pipe.buckets.push({
@@ -80,7 +93,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 +105,7 @@ addToLibrary({
           }
         }
 
+        pipe.registerReadableHanlders(notifyCallback);
         return 0;
       },
       dup(stream) {
@@ -204,6 +218,7 @@ addToLibrary({
         if (freeBytesInCurrBuffer >= dataLen) {
           currBucket.buffer.set(data, currBucket.offset);
           currBucket.offset += dataLen;
+          pipe.notifyReadableHanders();
           return dataLen;
         } else if (freeBytesInCurrBuffer > 0) {
           currBucket.buffer.set(data.subarray(0, freeBytesInCurrBuffer), currBucket.offset);
@@ -235,6 +250,7 @@ addToLibrary({
           newBucket.buffer.set(data);
         }
 
+        pipe.notifyReadableHanders();
         return dataLen;
       },
       close(stream) {