feat: close seqpacket socket automatically when both read and write ends are closed

Author: oyyd

README.md

@@ -1,33 +1,152 @@
 # nix-socket
 
 `nix-socket` allows you to use some nonblocking sockets that are not supported by Node.js native modules, including:
+- Using `SO_REUSEPORT` enabled TCP [net.Server](https://nodejs.org/dist/latest-v16.x/docs/api/net.html#class-netserver)
 - unix seqpacket(`SOCK_SEQPACKET`) sockets
 - unix datagram(`SOCK_DGRAM`) sockets
-- Using `SO_REUSEPORT` for TCP [net.Server](https://nodejs.org/dist/latest-v16.x/docs/api/net.html#class-netserver)
 
-`nix-socket` is a [napi-rs](https://napi.rs/) based [Node.js addons](https://nodejs.org/docs/latest-v16.x/api/addons.html). This lib uses [libuv](https://libuv.org/) inside Node.js so that it won't introduce other asynchronous runtimes.
+`nix-socket` is a [napi-rs](https://napi.rs/) based [Node.js addons](https://nodejs.org/docs/latest-v16.x/api/addons.html). This lib uses [libuv](https://libuv.org/) inside Node.js so that it won't introduce any other asynchronous runtimes.
 
-## Examples
+## API Documents
+
+[API Documents](./docs/modules.md)
+
+## `SO_REUSEPORT` enabled TCP net.Server
+
+The [cluster](https://nodejs.org/dist/latest-v18.x/docs/api/cluster.html) module share server ports by accepting new connections in the primary process and distributing them to worker processes.
+
+With `SO_REUSEPORT`, sockets will be distributed by kernel instead, and which should be more performant especially for scenario of having a lot of short-lived connections.
+
+For example, the arrow in the image below shows cpu usage of a PM2 primary process which we found in our environment.
+
+![cpu_usage](./resource/cpu_usage.png)
+
+Note that `SO_REUSEPORT` might behave much differently across operating systems. See this [post](https://stackoverflow.com/questions/14388706/how-do-so-reuseaddr-and-so-reuseport-differ) for more information.
+
+### Example
+
+```js
+const { createReuseportFd } = require('nix-socket')
+const { Server, Socket } = require('net')
+
+const port = 8080
+const host = '0.0.0.0'
+
+// create multple servers listening to a same host, port.
+for (let i = 0; i < 2; i += 1) {
+  const fd = createReuseportFd(port, host)
+  const server = new Server((socket) => {
+    socket.on('data', (buf) => {
+      console.log(`server ${i} received:`, buf)
+      // echo
+      socket.write(buf)
+    })
+  })
+
+  server.listen({
+    fd,
+  }, () => {
+    console.log(`server ${i} is listening on ${port}`)
+  })
+}
+
+setInterval(() => {
+  const client = new Socket()
+  client.on('data', (buf) => {
+    console.log('client received:', buf)
+    client.destroy()
+  })
+  client.connect(port, host, () => {
+    client.write(Buffer.from('hello'))
+  })
+}, 1000)
+```
+
+## Seqpacket Sockets
+
+`SOCK_SEQPACKET` sockets are like `SOCK_DGRAM` sockets and they will keep message boundaries.
 
-### Seqpacket Sockets
+Note that `SOCK_SEQPACKET` sockets don't work on MacOS.
+
+### Example
 
 ```js
 const { SeqpacketServer, SeqpacketSocket } = require('nix-socket')
 const os = require('os')
 const path = require('path')
+const fs = require('fs')
 
-const bindPath = path.resolve(os.tmp(), './my.sock')
+const bindPath = path.resolve(os.tmpdir(), './my_seqpacket.sock')
 
-const server = new SeqpacketServer()
+try { fs.unlinkSync(bindPath) } catch (e) {}
 
+const server = new SeqpacketServer()
 server.listen(bindPath)
+server.on('connection', socket => {
+  socket.on('data', buf => {
+    console.log('received', buf.toString())
+  })
+});
 
 const client = new SeqpacketSocket()
+client.connect(bindPath, () => {
+  const data = [
+    'hello, ',
+    'w',
+    'o',
+    'r',
+    'l',
+    'd'
+  ]
 
-// TODO
-client.write()
+  for (const str of data) {
+    client.write(Buffer.from(str))
+  }
+  client.end()
+})
 ```
 
-## API Documents
+## Dgram Sockets
 
-[API Documents](./docs/modules.md)
+### Example
+
+```js
+const { DgramSocket } = require('nix-socket')
+const os = require('os')
+const path = require('path')
+const fs = require('fs')
+
+const path1 = path.resolve(os.tmpdir(), './my_dgram_1.sock')
+const path2 = path.resolve(os.tmpdir(), './my_dgram_2.sock')
+
+try {
+  fs.unlinkSync(path1);
+  fs.unlinkSync(path2);
+} catch (err) {}
+
+const socket1 = new DgramSocket()
+const socket2 = new DgramSocket()
+
+socket1.bind(path1)
+socket2.bind(path2)
+
+socket2.on('data', (data, remoteAddr) => {
+  console.log(`socket2 received: ${data.toString()}`)
+  // echo
+  socket2.sendTo(data, 0, data.length, remoteAddr);
+})
+
+socket1.on('data', (data) => {
+  console.log(`socket1 received: ${data.toString()}`)
+})
+
+setInterval(() => {
+  const buf = Buffer.from('hello')
+  socket1.sendTo(buf, 0, buf.length, path2)
+}, 1000)
+```
+
+## LICENSE
+
+## TODO
+- seqpacket sockets should be closed automatically

__test__/seqpacket.spec.ts

@@ -11,8 +11,10 @@ async function createTestPair(
     client: SeqpacketSocket;
     server: SeqpacketServer;
     socket: SeqpacketSocket;
-  }) => Promise<any>
+  }) => Promise<any>,
+  options: { autoClose: boolean } = { autoClose: true }
 ) {
+  const { autoClose } = options
   const server = new SeqpacketServer();
   const client = new SeqpacketSocket();
   server.listen(kServerpath);
@@ -33,6 +35,9 @@ async function createTestPair(
     socket,
   });
 
+  if (!autoClose) {
+    return
+  }
   socket.destroy();
   client.destroy();
   server.close();
@@ -552,6 +557,47 @@ if (!kIsDarwin) {
         client.unref();
       });
     });
+
+    it('should write whole buffer if "offset" and "length" are missed', async () => {
+      await createTestPair(async (args) => {
+        const { client, socket } = args;
+        const buf = Buffer.from('hello, world');
+        client.write(buf)
+
+        const {p, resolve } = createDefer();
+        socket.on('data', received => {
+          expect(received.toString('hex')).toBe(buf.toString('hex'))
+          resolve()
+        })
+        await p
+      });
+    });
+
+    it('should emit "close" in sockets automatically when both read and write side of sockets are end', async () => {
+      await createTestPair(async (args) => {
+        const { client, socket, server } = args;
+
+        const { p: p1, resolve: r1 } = createDefer();
+        const { p: p2, resolve: r2 } = createDefer();
+
+        client.on('close', () => {
+          r1()
+        });
+        socket.on('close', () => {
+          r2()
+        })
+
+        client.write(Buffer.alloc(1024 * 64))
+        client.end()
+        socket.write(Buffer.alloc(1024 * 64))
+        socket.end()
+        await Promise.all([p1, p2])
+
+        server.close();
+      }, {
+        autoClose: false,
+      })
+    });
   });
 } else {
   describe('seqpacket', () => {

docs/README.md

@@ -1,7 +1,86 @@
-unix-socket / [Exports](modules.md)
+nix-socket / [Exports](modules.md)
 
-# unix-socket
+# nix-socket
+
+`nix-socket` allows you to use some nonblocking sockets that are not supported by Node.js native modules, including:
+- Using `SO_REUSEPORT` enabled TCP [net.Server](https://nodejs.org/dist/latest-v16.x/docs/api/net.html#class-netserver)
+- unix seqpacket(`SOCK_SEQPACKET`) sockets
+- unix datagram(`SOCK_DGRAM`) sockets
+
+`nix-socket` is a [napi-rs](https://napi.rs/) based [Node.js addons](https://nodejs.org/docs/latest-v16.x/api/addons.html). This lib uses [libuv](https://libuv.org/) inside Node.js so that it won't introduce any other asynchronous runtimes.
 
 ## API Documents
 
 [API Documents](./docs/modules.md)
+
+## `SO_REUSEPORT` enabled TCP net.Server
+
+The [cluster](https://nodejs.org/dist/latest-v18.x/docs/api/cluster.html) module share server ports by accepting new connections in the primary process and distributing them to worker processes.
+With `SO_REUSEPORT`, sockets will be distributed by kernel instead, and which should be more performant especially for scenario of having a lot of short-lived connections.
+
+Note that `SO_REUSEPORT` might behave much differently across operating systems. See this informative [post](https://stackoverflow.com/questions/14388706/how-do-so-reuseaddr-and-so-reuseport-differ) for more information.
+
+### Example
+
+```js
+const { createReuseportFd } = require('nix-socket')
+const { Server, Socket } = require('net')
+
+const port = 8080
+const host = '0.0.0.0'
+
+// create multple servers listening to a same host, port.
+for (let i = 0; i < 2; i += 1) {
+  const fd = createReuseportFd(port, host)
+  const server = new Server((socket) => {
+    socket.on('data', (buf) => {
+      console.log(`server ${i} received:`, buf)
+      // echo
+      socket.write(buf)
+    })
+  })
+
+  server.listen({
+    fd,
+  }, () => {
+    console.log(`server ${i} is listening on ${port}`)
+  })
+}
+
+setInterval(() => {
+    const client = new Socket()
+    client.on('data', (buf) => {
+      console.log('client received:', buf)
+      client.destroy()
+    })
+    client.connect(port, host, () => {
+      client.write(Buffer.from('hello'))
+    })
+}, 1000)
+```
+
+## Seqpacket Sockets
+
+```js
+const { SeqpacketServer, SeqpacketSocket } = require('nix-socket')
+const os = require('os')
+const path = require('path')
+
+const bindPath = path.resolve(os.tmp(), './my.sock')
+
+const server = new SeqpacketServer()
+
+server.listen(bindPath)
+
+const client = new SeqpacketSocket()
+
+// TODO
+client.write()
+```
+
+## Dgram Sockets
+
+```js
+```
+
+## LICENSE

docs/classes/DgramSocket.md

@@ -1,4 +1,4 @@
-[unix-socket](../README.md) / [Exports](../modules.md) / DgramSocket
+[nix-socket](../README.md) / [Exports](../modules.md) / DgramSocket
 
 # Class: DgramSocket
 

docs/classes/SeqpacketServer.md

@@ -1,4 +1,4 @@
-[unix-socket](../README.md) / [Exports](../modules.md) / SeqpacketServer
+[nix-socket](../README.md) / [Exports](../modules.md) / SeqpacketServer
 
 # Class: SeqpacketServer
 

docs/classes/SeqpacketSocket.md

@@ -1,4 +1,4 @@
-[unix-socket](../README.md) / [Exports](../modules.md) / SeqpacketSocket
+[nix-socket](../README.md) / [Exports](../modules.md) / SeqpacketSocket
 
 # Class: SeqpacketSocket
 

docs/modules.md

@@ -1,6 +1,6 @@
-[unix-socket](README.md) / Exports
+[nix-socket](README.md) / Exports
 
-# unix-socket
+# nix-socket
 
 ## Table of contents