Add -sNODENET backend for real outgoing TCP via node:net

Adds a new NODENET setting that backs the POSIX sockets API directly with
Node.js's node:net module, giving real, non-blocking outgoing (client) TCP
sockets without WebSockets, an external proxy process, or pthreads.

Unlike PROXY_POSIX_SOCKETS this is single-threaded and event-driven: socket
readiness is delivered through the same emscripten_set_socket_*_callback hooks
the default WebSocket backend uses, so it drops into existing readiness reactors
unchanged.

This initial backend supports outgoing TCP only: connect, send, recv and close,
plus get/setsockopt (SO_ERROR, TCP_NODELAY, SO_KEEPALIVE and the TCP keep-alive
tunables). There is no bind/listen/accept (server) support and no UDP yet; those
land in follow-ups.

- new nodenet_sock_ops in libsockfs.js implementing the sock_ops contract over
  net.createConnection
- route get/setsockopt through the backend under -sNODENET, and compile out the
  weak __syscall_setsockopt stub via a libstubs variation so the JS symbol wins
- test/sockets/test_nodenet.c: outgoing connect/send/recv against a loopback
  echo server started by the test harness

Author: guybedford

src/lib/libsockfs.js

@@ -8,7 +8,11 @@ addToLibrary({
   $SOCKFS__postset: () => {
     addAtInit('SOCKFS.root = FS.mount(SOCKFS, {}, null);');
   },
-  $SOCKFS__deps: ['$FS'],
+  $SOCKFS__deps: ['$FS',
+#if NODENET
+    '$ERRNO_CODES',
+#endif
+  ],
   $SOCKFS: {
 #if expectToReceiveOnModule('websocket')
     websocketArgs: {},
@@ -69,6 +73,8 @@ addToLibrary({
         pending: [],
         recv_queue: [],
 #if SOCKET_WEBRTC
+#elif NODENET
+        sock_ops: SOCKFS.nodenet_sock_ops
 #else
         sock_ops: SOCKFS.websocket_sock_ops
 #endif
@@ -726,7 +732,243 @@ addToLibrary({
 
         return res;
       }
-    }
+    },
+#if NODENET
+    // Outgoing TCP over node:net, using the same sock_ops contract and
+    // SOCKFS.emit readiness callbacks as the WebSocket backend. Client TCP only
+    // for now, with no bind, listen, accept or UDP.
+    nodenet_sock_ops: {
+      errnoForNode(e) {
+        return (e && e.code && ERRNO_CODES[e.code]) || {{{ cDefs.ECONNREFUSED }}};
+      },
+      // Replay buffered opts once the socket is live.
+      applyOptions(sock) {
+        var conn = sock.connection;
+        var o = sock.opts;
+        if (!conn || !o) return;
+        if (o.noDelay !== undefined) {
+          try { conn.setNoDelay(!!o.noDelay); } catch (e) {}
+        }
+        SOCKFS.nodenet_sock_ops.applyKeepAlive(sock);
+      },
+      // The keepalive tunables arrive from C in seconds, but node wants
+      // milliseconds, so we scale by 1000. A non-positive value keeps node's
+      // default for that field.
+      applyKeepAlive(sock) {
+        var conn = sock.connection;
+        var o = sock.opts;
+        if (!conn || !o || o.keepAlive === undefined) return;
+        try {
+          conn.setKeepAlive(
+            !!o.keepAlive,
+            (o.keepAliveIdle || 0) * 1000,
+            (o.keepAliveIntvl || 0) * 1000,
+            o.keepAliveCnt || 0);
+        } catch (e) {}
+      },
+      // Forward a connected node socket's events onto sock.
+      wireConnection(sock, conn) {
+        sock.connection = conn;
+        conn.on('data', (buf) => {
+          var data = new Uint8Array(buf.length);
+          data.set(buf);
+          sock.recv_queue.push({ addr: sock.daddr, port: sock.dport, data });
+          sock.recv_bytes = (sock.recv_bytes || 0) + data.length;
+          // If the peer outruns the reader, pause node and resume in recvmsg.
+          if (sock.recv_bytes >= 262144 /* 256 KiB */) {
+            try { conn.pause(); } catch (e) {}
+            sock.paused = true;
+          }
+          SOCKFS.emit('message', sock.stream.fd);
+        });
+        // A peer FIN surfaces as EOF to the reader.
+        conn.on('end', () => {
+          sock.readClosed = true;
+          SOCKFS.emit('message', sock.stream.fd);
+        });
+        conn.on('close', () => {
+          sock.readClosed = true;
+          sock.state = 'closed';
+          SOCKFS.emit('close', sock.stream.fd);
+        });
+        // Backpressure relieved, so we are writable again.
+        conn.on('drain', () => {
+          sock.writeBlocked = false;
+          SOCKFS.emit('open', sock.stream.fd);
+        });
+        conn.on('error', (e) => {
+          sock.error = SOCKFS.nodenet_sock_ops.errnoForNode(e);
+          // Let a failed connect resolve so SO_ERROR can be read.
+          if (sock.state === 'connecting') sock.state = 'connected';
+          SOCKFS.emit('error', [sock.stream.fd, sock.error, (e && e.message) || 'socket error']);
+        });
+      },
+      poll(sock) {
+        var mask = 0;
+        if (sock.recv_queue.length || sock.readClosed || sock.error) {
+          mask |= ({{{ cDefs.POLLRDNORM }}} | {{{ cDefs.POLLIN }}});
+        }
+        if (sock.error) {
+          // Mark writable on error so SO_ERROR can be read.
+          mask |= {{{ cDefs.POLLOUT }}};
+        } else if (sock.connection && sock.state === 'connected' && !sock.writeBlocked) {
+          mask |= {{{ cDefs.POLLOUT }}};
+        }
+        if (sock.readClosed) mask |= {{{ cDefs.POLLHUP }}};
+        return mask;
+      },
+      ioctl(sock, request, arg) {
+        switch (request) {
+          case {{{ cDefs.FIONREAD }}}:
+            var bytes = sock.recv_queue.length ? sock.recv_queue[0].data.length : 0;
+            {{{ makeSetValue('arg', '0', 'bytes', 'i32') }}};
+            return 0;
+          case {{{ cDefs.FIONBIO }}}:
+            var on = {{{ makeGetValue('arg', '0', 'i32') }}};
+            if (on) sock.stream.flags |= {{{ cDefs.O_NONBLOCK }}};
+            else sock.stream.flags &= ~{{{ cDefs.O_NONBLOCK }}};
+            return 0;
+          default:
+            return {{{ cDefs.EINVAL }}};
+        }
+      },
+      close(sock) {
+        sock.state = 'closed';
+        if (sock.connection) { try { sock.connection.destroy(); } catch (e) {} sock.connection = null; }
+        return 0;
+      },
+      connect(sock, addr, port) {
+        if (sock.connection) {
+          throw new FS.ErrnoError(sock.state === 'connecting' ? {{{ cDefs.EALREADY }}} : {{{ cDefs.EISCONN }}});
+        }
+        sock.daddr = addr;
+        sock.dport = port;
+        sock.state = 'connecting';
+        var conn = require('net').createConnection({ host: addr, port, allowHalfOpen: true });
+        conn.once('connect', () => {
+          sock.state = 'connected';
+          sock.saddr = conn.localAddress;
+          sock.sport = conn.localPort;
+          sock.daddr = conn.remoteAddress || addr;
+          sock.dport = conn.remotePort || port;
+          SOCKFS.nodenet_sock_ops.applyOptions(sock);
+          SOCKFS.emit('open', sock.stream.fd);
+        });
+        SOCKFS.nodenet_sock_ops.wireConnection(sock, conn);
+      },
+      sendmsg(sock, buffer, offset, length, addr, port) {
+        if (!sock.connection || sock.state === 'closed') {
+          throw new FS.ErrnoError({{{ cDefs.ENOTCONN }}});
+        }
+        if (ArrayBuffer.isView(buffer)) { offset += buffer.byteOffset; buffer = buffer.buffer; }
+        var data = new Uint8Array(buffer.slice(offset, offset + length));
+        var ok;
+        try {
+          ok = sock.connection.write(data);
+        } catch (e) {
+          throw new FS.ErrnoError(SOCKFS.nodenet_sock_ops.errnoForNode(e));
+        }
+        if (!ok) sock.writeBlocked = true; // cleared on 'drain'
+        return length;
+      },
+      recvmsg(sock, length) {
+        var queued = sock.recv_queue.shift();
+        if (!queued) {
+          if (sock.readClosed) return null; // EOF
+          if (!sock.connection) {
+            throw new FS.ErrnoError({{{ cDefs.ENOTCONN }}});
+          }
+          throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});
+        }
+        var q = queued.data;
+        var bytesRead = Math.min(length, q.length);
+        var res = { buffer: q.subarray(0, bytesRead), addr: queued.addr, port: queued.port };
+        if (bytesRead < q.length) {
+          queued.data = q.subarray(bytesRead);
+          sock.recv_queue.unshift(queued);
+        }
+        sock.recv_bytes = Math.max(0, (sock.recv_bytes || 0) - bytesRead);
+        if (sock.paused && sock.recv_bytes < 262144 && sock.connection) {
+          sock.paused = false;
+          try { sock.connection.resume(); } catch (e) {}
+        }
+        return res;
+      },
+      setsockopt(sock, level, optname, optval, optlen) {
+        sock.opts ||= {};
+        var val = {{{ makeGetValue('optval', 0, 'i32') }}};
+        if (level === {{{ cDefs.SOL_SOCKET }}}) {
+          switch (optname) {
+            case 9: // SO_KEEPALIVE
+              sock.opts.keepAlive = !!val;
+              SOCKFS.nodenet_sock_ops.applyKeepAlive(sock);
+              return 0;
+            case 8: // SO_RCVBUF. Node TCP cannot set this, so we just store it.
+              sock.opts.recvBuf = val;
+              return 0;
+            case 7: // SO_SNDBUF. Node TCP cannot set this, so we just store it.
+              sock.opts.sendBuf = val;
+              return 0;
+            case 2: // SO_REUSEADDR. Node manages reuse, so we just store it.
+              sock.opts.reuseAddr = !!val;
+              return 0;
+          }
+        } else if (level === {{{ cDefs.IPPROTO_TCP }}}) {
+          switch (optname) {
+            case 1: // TCP_NODELAY
+              sock.opts.noDelay = !!val;
+              if (sock.connection) { try { sock.connection.setNoDelay(!!val); } catch (e) {} }
+              return 0;
+            case 4: // TCP_KEEPIDLE (seconds)
+              sock.opts.keepAliveIdle = val;
+              SOCKFS.nodenet_sock_ops.applyKeepAlive(sock);
+              return 0;
+            case 5: // TCP_KEEPINTVL (seconds)
+              sock.opts.keepAliveIntvl = val;
+              SOCKFS.nodenet_sock_ops.applyKeepAlive(sock);
+              return 0;
+            case 6: // TCP_KEEPCNT (probe count)
+              sock.opts.keepAliveCnt = val;
+              SOCKFS.nodenet_sock_ops.applyKeepAlive(sock);
+              return 0;
+          }
+        }
+        // Accept unknown options silently, like a permissive stack.
+        return 0;
+      },
+      getsockopt(sock, level, optname, optval, optlen) {
+        sock.opts ||= {};
+        var val;
+        if (level === {{{ cDefs.SOL_SOCKET }}}) {
+          switch (optname) {
+            case {{{ cDefs.SO_ERROR }}}:
+              {{{ makeSetValue('optval', 0, 'sock.error || 0', 'i32') }}};
+              {{{ makeSetValue('optlen', 0, 4, 'i32') }}};
+              sock.error = null; // SO_ERROR reads and clears
+              return 0;
+            case 9: val = sock.opts.keepAlive ? 1 : 0; break; // SO_KEEPALIVE
+            case 8: val = sock.opts.recvBuf || 65536; break;  // SO_RCVBUF
+            case 7: val = sock.opts.sendBuf || 65536; break;  // SO_SNDBUF
+            case 2: val = sock.opts.reuseAddr ? 1 : 0; break; // SO_REUSEADDR
+            default: return -{{{ cDefs.ENOPROTOOPT }}};
+          }
+        } else if (level === {{{ cDefs.IPPROTO_TCP }}}) {
+          switch (optname) {
+            case 1: val = sock.opts.noDelay ? 1 : 0; break;    // TCP_NODELAY
+            case 4: val = sock.opts.keepAliveIdle || 0; break; // TCP_KEEPIDLE
+            case 5: val = sock.opts.keepAliveIntvl || 0; break;// TCP_KEEPINTVL
+            case 6: val = sock.opts.keepAliveCnt || 0; break;  // TCP_KEEPCNT
+            default: return -{{{ cDefs.ENOPROTOOPT }}};
+          }
+        } else {
+          return -{{{ cDefs.ENOPROTOOPT }}};
+        }
+        {{{ makeSetValue('optval', 0, 'val', 'i32') }}};
+        {{{ makeSetValue('optlen', 0, 4, 'i32') }}};
+        return 0;
+      }
+    },
+#endif
   },
 
   /*

src/lib/libsyscall.js

@@ -445,6 +445,10 @@ var SyscallsLibrary = {
   __syscall_getsockopt__deps: ['$getSocketFromFD'],
   __syscall_getsockopt: (fd, level, optname, optval, optlen, d1) => {
     var sock = getSocketFromFD(fd);
+#if NODENET
+    // The node:net backend handles all socket options.
+    return sock.sock_ops.getsockopt(sock, level, optname, optval, optlen);
+#else
     // Minimal getsockopt aimed at resolving https://github.com/emscripten-core/emscripten/issues/2211
     // so only supports SOL_SOCKET with SO_ERROR.
     if (level === {{{ cDefs.SOL_SOCKET }}}) {
@@ -456,7 +460,17 @@ var SyscallsLibrary = {
       }
     }
     return -{{{ cDefs.ENOPROTOOPT }}}; // The option is unknown at the level indicated.
+#endif
   },
+#if NODENET
+  // Only the node:net backend implements setsockopt. By default it stays
+  // unimplemented.
+  __syscall_setsockopt__deps: ['$getSocketFromFD'],
+  __syscall_setsockopt: (fd, level, optname, optval, optlen, d1) => {
+    var sock = getSocketFromFD(fd);
+    return sock.sock_ops.setsockopt(sock, level, optname, optval, optlen);
+  },
+#endif
   __syscall_sendmsg__deps: ['$getSocketFromFD', '$getSocketAddress'],
   __syscall_sendmsg: (fd, message, flags, d1, d2, d3) => {
     var sock = getSocketFromFD(fd);

src/settings.js

@@ -419,6 +419,20 @@ var WEBSOCKET_URL = 'ws://';
 // [link]
 var PROXY_POSIX_SOCKETS = false;
 
+// If 1, the POSIX sockets API is backed by Node.js's ``node:net`` module,
+// giving real non-blocking outgoing TCP sockets with no WebSockets, proxy
+// process or pthreads. This only works under node and is ignored elsewhere.
+//
+// For now it supports outgoing TCP only. There is no bind, listen, accept or
+// UDP yet.
+//
+// It is single-threaded and event-driven. Socket readiness comes through the
+// same ``emscripten_set_socket_*_callback`` hooks the WebSocket backend uses,
+// so it works with existing readiness reactors. It cannot be combined with the
+// WebSocket emulation, PROXY_POSIX_SOCKETS or SOCKET_WEBRTC.
+// [link]
+var NODENET = false;
+
 // A string containing a comma separated list of WebSocket subprotocols
 // as would be present in the Sec-WebSocket-Protocol header.
 // You can set 'null', if you don't want to specify it.

system/lib/libc/emscripten_syscall_stubs.c

@@ -248,10 +248,14 @@ weak int __syscall_prlimit64(int pid, int resource, intptr_t new_limit, intptr_t
   return 0;
 }
 
+// Under NODENET the JS library provides setsockopt, so drop this weak stub to
+// leave the symbol undefined for it to fill.
+#ifndef __EMSCRIPTEN_NODENET__
 weak int __syscall_setsockopt(int sockfd, int level, int optname, intptr_t optval, size_t optlen, int dummy) {
   REPORT(setsockopt);
   return -ENOPROTOOPT; // The option is unknown at the level indicated.
 }
+#endif
 
 UNIMPLEMENTED(acct, (intptr_t filename))
 UNIMPLEMENTED(mincore, (intptr_t addr, size_t length, intptr_t vec))