streams.mjs

/**
 * This is a pared down version of
 * https://github.com/pyodide/pyodide/blob/main/src/js/streams.ts
 *
 * It replaces the standard streams devices that Emscripten provides with our
 * own better ones. It fixes the following deficiencies:
 *
 * 1. The emscripten std streams always have isatty set to true. These set
 *    isatty to match the value for the stdin/stdout/stderr that node sees.
 * 2. The emscripten std streams don't support the ttygetwinsize ioctl. If
 *    isatty() returns true, then these do, and it returns the actual window
 *    size as the OS reports it to Node.
 * 3. The emscripten std streams introduce an extra layer of buffering which has
 *    to be flushed with fsync().
 * 4. The emscripten std streams are slow and complex because they go through a
 *    character-based handler layer. This is particularly awkward because both
 *    sides of this character based layer deal with buffers and so we need
 *    complex adaptors, buffering, etc on both sides. Removing this
 *    character-based middle layer makes everything better.
 *    https://github.com/emscripten-core/emscripten/blob/1aa7fb531f11e11e7ae49b75a24e1a8fe6fa4a7d/src/lib/libtty.js?plain=1#L104-L114
 *
 * Ideally some version of this should go upstream to Emscripten since it is not
 * in any way specific to Python. But I (Hood) haven't gotten around to it yet.
 */

import * as tty from "node:tty";
import * as fs from "node:fs";

let FS;
const DEVOPS = {};
const DEVS = {};

function isErrnoError(e) {
  return e && typeof e === "object" && "errno" in e;
}

const waitBuffer = new Int32Array(
  new WebAssembly.Memory({ shared: true, initial: 1, maximum: 1 }).buffer,
);
function syncSleep(timeout) {
  try {
    Atomics.wait(waitBuffer, 0, 0, timeout);
    return true;
  } catch (_) {
    return false;
  }
}

/**
 * Calls the callback and handle node EAGAIN errors.
 */
function handleEAGAIN(cb) {
  while (true) {
    try {
      return cb();
    } catch (e) {
      if (e && e.code === "EAGAIN") {
        // Presumably this means we're in node and tried to read from/write to
        // an O_NONBLOCK file descriptor. Synchronously sleep for 10ms then try
        // again. In case for some reason we fail to sleep, propagate the error
        // (it will turn into an EOFError).
        if (syncSleep(10)) {
          continue;
        }
      }
      throw e;
    }
  }
}

function readWriteHelper(stream, cb, method) {
  let nbytes;
  try {
    nbytes = handleEAGAIN(cb);
  } catch (e) {
    if (e && e.code && Module.ERRNO_CODES[e.code]) {
      throw new FS.ErrnoError(Module.ERRNO_CODES[e.code]);
    }
    if (isErrnoError(e)) {
      // the handler set an errno, propagate it
      throw e;
    }
    console.error("Error thrown in read:");
    console.error(e);
    throw new FS.ErrnoError(Module.ERRNO_CODES.EIO);
  }
  if (nbytes === undefined) {
    // Prevent an infinite loop caused by incorrect code that doesn't return a
    // value.
    // Maybe we should set nbytes = buffer.length here instead?
    console.warn(
      `${method} returned undefined; a correct implementation must return a number`,
    );
    throw new FS.ErrnoError(Module.ERRNO_CODES.EIO);
  }
  if (nbytes !== 0) {
    stream.node.timestamp = Date.now();
  }
  return nbytes;
}

function asUint8Array(arg) {
  if (ArrayBuffer.isView(arg)) {
    return new Uint8Array(arg.buffer, arg.byteOffset, arg.byteLength);
  } else {
    return new Uint8Array(arg);
  }
}

const prepareBuffer = (buffer, offset, length) =>
  asUint8Array(buffer).subarray(offset, offset + length);

const TTY_OPS = {
  ioctl_tiocgwinsz(tty) {
    return tty.devops.ioctl_tiocgwinsz?.() ?? [24, 80];
  },
};

const stream_ops = {
  open: function (stream) {
    const devops = DEVOPS[stream.node.rdev];
    if (!devops) {
      throw new FS.ErrnoError(Module.ERRNO_CODES.ENODEV);
    }
    stream.devops = devops;
    stream.tty = stream.devops.isatty ? { ops: TTY_OPS, devops } : undefined;
    stream.seekable = false;
  },
  close: function (stream) {
    // flush any pending line data
    stream.stream_ops.fsync(stream);
  },
  fsync: function (stream) {
    const ops = stream.devops;
    if (ops.fsync) {
      ops.fsync();
    }
  },
  read: function (stream, buffer, offset, length, pos /* ignored */) {
    buffer = prepareBuffer(buffer, offset, length);
    return readWriteHelper(stream, () => stream.devops.read(buffer), "read");
  },
  write: function (stream, buffer, offset, length, pos /* ignored */) {
    buffer = prepareBuffer(buffer, offset, length);
    return readWriteHelper(stream, () => stream.devops.write(buffer), "write");
  },
};

function nodeFsync(fd) {
  try {
    fs.fsyncSync(fd);
  } catch (e) {
    if (e?.code === "EINVAL") {
      return;
    }
    // On Mac, calling fsync when not isatty returns ENOTSUP
    // On Windows, stdin/stdout/stderr may be closed, returning EBADF or EPERM
    if (
      e?.code === "ENOTSUP" || e?.code === "EBADF" || e?.code === "EPERM"
    ) {
      return;
    }

    throw e;
  }
}

class NodeReader {
  constructor(nodeStream) {
    this.nodeStream = nodeStream;
    this.isatty = tty.isatty(nodeStream.fd);
  }

  read(buffer) {
    try {
      return fs.readSync(this.nodeStream.fd, buffer);
    } catch (e) {
      // Platform differences: on Windows, reading EOF throws an exception,
      // but on other OSes, reading EOF returns 0. Uniformize behavior by
      // catching the EOF exception and returning 0.
      if (e.toString().includes("EOF")) {
        return 0;
      }
      throw e;
    }
  }

  fsync() {
    nodeFsync(this.nodeStream.fd);
  }

  ioctl_tiocgwinsz() {
    return [this.nodeStream.rows ?? 24, this.nodeStream.columns ?? 80];
  }
}

class NodeWriter {
  constructor(nodeStream) {
    this.nodeStream = nodeStream;
    this.isatty = tty.isatty(nodeStream.fd);
  }

  write(buffer) {
    return fs.writeSync(this.nodeStream.fd, buffer);
  }

  fsync() {
    nodeFsync(this.nodeStream.fd);
  }

  ioctl_tiocgwinsz() {
    return [this.nodeStream.rows ?? 24, this.nodeStream.columns ?? 80];
  }
}

export function initializeStreams(fsarg) {
  FS = fsarg;
  const major = FS.createDevice.major++;
  DEVS.stdin = FS.makedev(major, 0);
  DEVS.stdout = FS.makedev(major, 1);
  DEVS.stderr = FS.makedev(major, 2);

  FS.registerDevice(DEVS.stdin, stream_ops);
  FS.registerDevice(DEVS.stdout, stream_ops);
  FS.registerDevice(DEVS.stderr, stream_ops);

  FS.unlink("/dev/stdin");
  FS.unlink("/dev/stdout");
  FS.unlink("/dev/stderr");

  FS.mkdev("/dev/stdin", DEVS.stdin);
  FS.mkdev("/dev/stdout", DEVS.stdout);
  FS.mkdev("/dev/stderr", DEVS.stderr);

  DEVOPS[DEVS.stdin] = new NodeReader(process.stdin);
  DEVOPS[DEVS.stdout] = new NodeWriter(process.stdout);
  DEVOPS[DEVS.stderr] = new NodeWriter(process.stderr);

  FS.closeStream(0 /* stdin */);
  FS.closeStream(1 /* stdout */);
  FS.closeStream(2 /* stderr */);
  FS.open("/dev/stdin", 0 /* O_RDONLY */);
  FS.open("/dev/stdout", 1 /* O_WRONLY */);
  FS.open("/dev/stderr", 1 /* O_WRONLY */);
}