src/lib/libpipefs.js

/**
 * @license
 * Copyright 2017 The Emscripten Authors
 * SPDX-License-Identifier: MIT
 */

addToLibrary({
  $PIPEFS__postset: () => addAtInit('PIPEFS.root = FS.mount(PIPEFS, {}, null);'),
  $PIPEFS__deps: ['$FS'],
  $PIPEFS: {
    BUCKET_BUFFER_SIZE: 1024 * 8, // 8KiB Buffer
    mount(mount) {
      // Do not pollute the real root directory or its child nodes with pipes
      // Looks like it is OK to create another pseudo-root node not linked to the FS.root hierarchy this way
      return FS.createNode(null, '/', {{{ cDefs.S_IFDIR }}} | 0o777, 0);
    },
    createPipe() {
      var pipe = {
        buckets: [],
        // Open write ends. When it drops to 0 the reader sees EOF and poll must
        // report POLLHUP (Linux semantics). Buckets are freed once both counts
        // reach 0.
        writerCount: 1,
        writeClosed: false,
        // Open read ends. When it drops to 0 the writer sees POLLERR (a further
        // write would get EPIPE).
        readerCount: 1,
        readClosed: false,
        timestamp: new Date(),
      };

      pipe.buckets.push({
        buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE),
        offset: 0,
        roffset: 0
      });

      var rName = PIPEFS.nextname();
      var wName = PIPEFS.nextname();
      var rNode = FS.createNode(PIPEFS.root, rName, {{{ cDefs.S_IFIFO }}}, 0);
      var wNode = FS.createNode(PIPEFS.root, wName, {{{ cDefs.S_IFIFO }}}, 0);

      rNode.pipe = pipe;
      wNode.pipe = pipe;
      // The read end's node carries the reader poll wait-queue (writes wake it);
      // the write end's node carries the writer wait-queue (read-end close wakes it).
      pipe.readNode = rNode;
      pipe.writeNode = wNode;

      var readableStream = FS.createStream({
        path: rName,
        node: rNode,
        flags: {{{ cDefs.O_RDONLY }}},
        seekable: false,
        stream_ops: PIPEFS.stream_ops
      });
      rNode.stream = readableStream;

      var writableStream = FS.createStream({
        path: wName,
        node: wNode,
        flags: {{{ cDefs.O_WRONLY }}},
        seekable: false,
        stream_ops: PIPEFS.stream_ops
      });
      wNode.stream = writableStream;

      return {
        readable_fd: readableStream.fd,
        writable_fd: writableStream.fd
      };
    },
    stream_ops: {
      getattr(stream) {
        var node = stream.node;
        var timestamp = node.pipe.timestamp;
        return {
          dev: 14,
          ino: node.id,
          mode: 0o10600,
          nlink: 1,
          uid: 0,
          gid: 0,
          rdev: 0,
          size: 0,
          atime: timestamp,
          mtime: timestamp,
          ctime: timestamp,
          blksize: 4096,
          blocks: 0,
        };
      },
      poll(stream) {
        var pipe = stream.node.pipe;

        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
          // Linux keeps the write end writable (the write itself fails with
          // EPIPE) while also signalling POLLERR once every read end is closed.
          var mask = {{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}};
          if (pipe.readClosed) {
            mask |= {{{ cDefs.POLLERR }}};
          }
          return mask;
        }
        var mask = 0;
        for (var bucket of pipe.buckets) {
          if (bucket.offset - bucket.roffset > 0) {
            mask = {{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}};
            break;
          }
        }
        // With every write end closed the read end is at EOF: readable (read
        // returns 0) and hung up.
        if (pipe.writeClosed) {
          mask |= {{{ cDefs.POLLHUP }}} | {{{ cDefs.POLLIN }}};
        }
        return mask;
      },
      dup(stream) {
        var pipe = stream.node.pipe;
        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
          pipe.writerCount++;
        } else {
          pipe.readerCount++;
        }
      },
      ioctl(stream, request, argp) {
        if (request == {{{ cDefs.FIONREAD }}}) {
          var pipe = stream.node.pipe;
          var currentLength = 0;
          for (var bucket of pipe.buckets) {
            currentLength += bucket.offset - bucket.roffset;
          }
          {{{ makeSetValue('argp', 0, 'currentLength', 'i32') }}};
          return 0;
        }
        return {{{ cDefs.EINVAL }}};
      },
      fsync(stream) {
        return {{{ cDefs.EINVAL }}};
      },
      read(stream, buffer, offset, length, position /* ignored */) {
        var pipe = stream.node.pipe;
        var currentLength = 0;

        for (var bucket of pipe.buckets) {
          currentLength += bucket.offset - bucket.roffset;
        }

#if ASSERTIONS && !(MEMORY64 && MAXIMUM_MEMORY > FOUR_GB)
#if PTHREADS
        assert(buffer instanceof ArrayBuffer || buffer instanceof SharedArrayBuffer || ArrayBuffer.isView(buffer));
#else
        assert(buffer instanceof ArrayBuffer || ArrayBuffer.isView(buffer));
#endif
#endif
        var data = buffer.subarray(offset, offset + length);

        if (length <= 0) {
          return 0;
        }
        if (currentLength == 0) {
          // Behave as if the read end is always non-blocking
          throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});
        }
        var toRead = Math.min(currentLength, length);

        var totalRead = toRead;
        var toRemove = 0;

        for (var bucket of pipe.buckets) {
          var bucketSize = bucket.offset - bucket.roffset;

          if (toRead <= bucketSize) {
            var tmpSlice = bucket.buffer.subarray(bucket.roffset, bucket.offset);
            if (toRead < bucketSize) {
              tmpSlice = tmpSlice.subarray(0, toRead);
              bucket.roffset += toRead;
            } else {
              toRemove++;
            }
            data.set(tmpSlice);
            break;
          } else {
            var tmpSlice = bucket.buffer.subarray(bucket.roffset, bucket.offset);
            data.set(tmpSlice);
            data = data.subarray(tmpSlice.byteLength);
            toRead -= tmpSlice.byteLength;
            toRemove++;
          }
        }

        if (toRemove && toRemove == pipe.buckets.length) {
          // Do not generate excessive garbage in use cases such as
          // write several bytes, read everything, write several bytes, read everything...
          toRemove--;
          pipe.buckets[toRemove].offset = 0;
          pipe.buckets[toRemove].roffset = 0;
        }

        pipe.buckets.splice(0, toRemove);

        return totalRead;
      },
      write(stream, buffer, offset, length, position /* ignored */) {
        var pipe = stream.node.pipe;

#if ASSERTIONS && !(MEMORY64 && MAXIMUM_MEMORY > FOUR_GB)
#if PTHREADS
        assert(buffer instanceof ArrayBuffer || buffer instanceof SharedArrayBuffer || ArrayBuffer.isView(buffer));
#else
        assert(buffer instanceof ArrayBuffer || ArrayBuffer.isView(buffer));
#endif
#endif
        var data = buffer.subarray(offset, offset + length);

        var dataLen = data.byteLength;
        if (dataLen <= 0) {
          return 0;
        }

        var currBucket = null;

        if (pipe.buckets.length == 0) {
          currBucket = {
            buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE),
            offset: 0,
            roffset: 0
          };
          pipe.buckets.push(currBucket);
        } else {
          currBucket = pipe.buckets[pipe.buckets.length - 1];
        }

#if ASSERTIONS
        assert(currBucket.offset <= PIPEFS.BUCKET_BUFFER_SIZE);
#endif

        var freeBytesInCurrBuffer = PIPEFS.BUCKET_BUFFER_SIZE - currBucket.offset;
        if (freeBytesInCurrBuffer >= dataLen) {
          currBucket.buffer.set(data, currBucket.offset);
          currBucket.offset += dataLen;
          pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
          return dataLen;
        } else if (freeBytesInCurrBuffer > 0) {
          currBucket.buffer.set(data.subarray(0, freeBytesInCurrBuffer), currBucket.offset);
          currBucket.offset += freeBytesInCurrBuffer;
          data = data.subarray(freeBytesInCurrBuffer, data.byteLength);
        }

        var numBuckets = (data.byteLength / PIPEFS.BUCKET_BUFFER_SIZE) | 0;
        var remElements = data.byteLength % PIPEFS.BUCKET_BUFFER_SIZE;

        for (var i = 0; i < numBuckets; i++) {
          var newBucket = {
            buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE),
            offset: PIPEFS.BUCKET_BUFFER_SIZE,
            roffset: 0
          };
          pipe.buckets.push(newBucket);
          newBucket.buffer.set(data.subarray(0, PIPEFS.BUCKET_BUFFER_SIZE));
          data = data.subarray(PIPEFS.BUCKET_BUFFER_SIZE, data.byteLength);
        }

        if (remElements > 0) {
          var newBucket = {
            buffer: new Uint8Array(PIPEFS.BUCKET_BUFFER_SIZE),
            offset: data.byteLength,
            roffset: 0
          };
          pipe.buckets.push(newBucket);
          newBucket.buffer.set(data);
        }

        pipe.readNode.notifyListeners({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
        return dataLen;
      },
      close(stream) {
        var pipe = stream.node.pipe;
        // When the last write end closes, wake any poll/epoll waiter on the read
        // end with POLLHUP so a reader blocked on the writer dropping unblocks.
        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
          if (--pipe.writerCount === 0) {
            pipe.writeClosed = true;
            pipe.readNode.notifyListeners({{{ cDefs.POLLHUP }}} | {{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
          }
        } else if (--pipe.readerCount === 0) {
          // Mirror: when the last read end closes, wake any poll/epoll waiter on
          // the write end with POLLERR (a further write would get EPIPE).
          pipe.readClosed = true;
          pipe.writeNode.notifyListeners({{{ cDefs.POLLERR }}} | {{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}});
        }
        if (pipe.readerCount === 0 && pipe.writerCount === 0) {
          pipe.buckets = null;
        }
      }
    },
    nextname() {
      if (!PIPEFS.nextname.current) {
        PIPEFS.nextname.current = 0;
      }
      return 'pipe[' + (PIPEFS.nextname.current++) + ']';
    },
  },
});