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: [],
        // refcnt 2 because pipe has a read end and a write end. We need to be
        // able to read from the read end after write end is closed.
        refcnt : 2,
        timestamp: new Date(),
#if PTHREADS || ASYNCIFY
        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({
        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;

      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, timeout, notifyCallback) {
        var pipe = stream.node.pipe;

        if ((stream.flags & {{{ cDefs.O_ACCMODE }}}) === {{{ cDefs.O_WRONLY }}}) {
          return ({{{ cDefs.POLLWRNORM }}} | {{{ cDefs.POLLOUT }}});
        }
        for (var bucket of pipe.buckets) {
          if (bucket.offset - bucket.roffset > 0) {
            return ({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
          }
        }

#if PTHREADS || ASYNCIFY
        if (notifyCallback) pipe.registerReadableHandler(notifyCallback);
#endif
        return 0;
      },
      dup(stream) {
        stream.node.pipe.refcnt++;
      },
      ioctl(stream, request, varargs) {
        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;
#if PTHREADS || ASYNCIFY
          pipe.notifyReadableHandlers();
#endif
          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);
        }

#if PTHREADS || ASYNCIFY
        pipe.notifyReadableHandlers();
#endif
        return dataLen;
      },
      close(stream) {
        var pipe = stream.node.pipe;
        pipe.refcnt--;
        if (pipe.refcnt === 0) {
          pipe.buckets = null;
        }
      }
    },
    nextname() {
      if (!PIPEFS.nextname.current) {
        PIPEFS.nextname.current = 0;
      }
      return 'pipe[' + (PIPEFS.nextname.current++) + ']';
    },
  },
});