feat: add createReuseportFd

Author: oyyd

README.md

@@ -1,15 +1,6 @@
 # unix-socket
 
-## TODO
 
-- support `socket.setSendBufferSize(size)`
-- current, server don't work with cluster module
+## TODO
 
-## Tmp
-- Connect, Accept
-  - how to make connect async?
-    - set nonblock and connect
-    - wait POLLOUT?
-  - Mode
-    - all callbacks, a lot
-    - event emitter instead
+- mark that server don't work with cluster module

__test__/seqpacket.spec.ts

@@ -502,4 +502,8 @@ if (!kIsDarwin) {
       });
     });
   });
+} else {
+  describe('seqpacket', () => {
+    it('(tests skipped)', () => {})
+  })
 }

__test__/socket.spec.ts

@@ -0,0 +1,105 @@
+import * as net from 'net'
+import { createReuseportFd as createFd, closeFd  } from '../js/index'
+import { hasIPv6 } from './util'
+
+describe('tcp', () => {
+  describe('createFd', () => {
+    it('should work', async () => {
+      const host = '0.0.0.0'
+      let port = 0;
+
+      async function createServer() {
+        const fd = createFd(port, host);
+
+        const server = await new Promise<net.Server>((resolve, reject) => {
+          const server = net.createServer()
+
+          server.listen({
+            fd,
+          }, () => {
+            resolve(server)
+          })
+        })
+
+        port = (server.address() as any).port
+
+        return server
+      }
+
+      const servers = [];
+      for (let i = 0; i < 5; i += 1) {
+        const server = await createServer()
+        servers.push(server);
+      }
+
+      const pList = servers.map(server => {
+        return new Promise((resolve, reject) => {
+          server.once('connection', (socket) => {
+            socket.on('data', buf => {
+              resolve(buf)
+            })
+          })
+        })
+      })
+
+      const buf = Buffer.from('hello');
+      const socket = net.connect(port, host, () => {
+        socket.write(buf);
+      });
+
+      const ret = await Promise.race(pList);
+      expect(ret.toString()).toBe(buf.toString())
+
+      socket.destroy();
+
+      servers.forEach(server => server.close());
+    })
+
+    if (hasIPv6())  {
+      it('should work with ipv6', async () => {
+        const host = '::1'
+        let port = 0;
+
+        const fd = createFd(port, host);
+
+        const server = await new Promise<net.Server>((resolve, reject) => {
+          const server = net.createServer()
+
+          server.listen({
+            fd,
+          }, () => {
+            resolve(server)
+          })
+        })
+        port = (server.address() as any).port
+        const p = new Promise((resolve, reject) => {
+          server.once('connection', (socket) => {
+            socket.on('data', buf => {
+              resolve(buf)
+            })
+          })
+        })
+
+        const buf = Buffer.from('hello');
+        const socket = net.connect(port, host, () => {
+          socket.write(buf);
+        });
+
+        const ret = await p;
+        expect(ret.toString()).toBe(buf.toString())
+
+        socket.destroy();
+
+        server.close();
+      });
+    }
+  })
+
+  describe('closeFd', () => {
+    it('should work', async () => {
+      const fd = createFd(0)
+
+      closeFd(fd)
+    })
+  });
+})

__test__/util.ts

@@ -31,3 +31,14 @@ export function createDefer<T>() {
     reject,
   }
 }
+
+const isWindows = os.platform() === 'win32';
+
+export function hasIPv6() {
+  const iFaces = os.networkInterfaces();
+  const re = isWindows ? /Loopback Pseudo-Interface/ : /lo/;
+  return Object.keys(iFaces).some((name) => {
+    return re.test(name) &&
+           iFaces[name].some(({ family }) => family === 'IPv6');
+  });
+}

js/addon.d.ts

@@ -24,3 +24,5 @@ export function dgramGetSendBufferSize(ee: object): number
 export function dgramSetSendBufferSize(ee: object, size: number): void
 export function dgramSendTo(ee: object, buf: Buffer, offset: number, length: number, path: string, cb?: (...args: any[]) => any | undefined | null): void
 export function startRecv(ee: object): void
+export function socketNewSoReuseportFd(domain: string, port: number, ip: string): number
+export function socketClose(fd: number): void

js/index.ts

@@ -1,2 +1,3 @@
 export { SendCb, DgramSocket } from './dgram'
 export { NotifyCb, SeqpacketSocket, SeqpacketServer } from './seqpacket'
+export { createReuseportFd, closeFd } from './socket'

js/socket.ts

@@ -0,0 +1,18 @@
+import { isIPv4, isIP } from 'net';
+import { socketNewSoReuseportFd, socketClose } from './addon'
+
+export function createReuseportFd(port: number = 0, host: string = '0.0.0.0'): number {
+  if (!isIP(host)) {
+    throw new Error('invalid host');
+  }
+
+  const domain = isIPv4(host) ? 'ipv4' : 'ipv6'
+
+  const fd = socketNewSoReuseportFd(domain, port, host);
+
+  return fd;
+}
+
+export function closeFd(fd: number) {
+  socketClose(fd)
+}

src/socket.rs

@@ -1,18 +1,20 @@
+use std::ffi::CString;
 use std::mem;
-use std::{str::FromStr};
+use std::str::FromStr;
 
-use libc::{sockaddr_un};
-use napi::{Result, Env, Ref, JsFunction, JsUnknown, JsObject};
+use crate::util::{error, get_err, resolve_libc_err, resolve_uv_err};
+use libc::{c_void, sockaddr_storage, sockaddr_un};
+use napi::{Env, JsFunction, JsNumber, JsObject, JsString, JsUnknown, Ref, Result};
 use uv_sys::sys;
-use crate::util::{get_err, error};
 
-pub (crate) fn get_loop(env: &Env) -> Result<*mut sys::uv_loop_t> {
+pub(crate) fn get_loop(env: &Env) -> Result<*mut sys::uv_loop_t> {
   Ok(env.get_uv_event_loop()? as *mut _ as *mut sys::uv_loop_t)
 }
 
 pub(crate) fn close(fd: i32) -> Result<()> {
   let ret = unsafe { libc::close(fd) };
 
+  // TODO should we loop?
   if ret != 0 {
     if ret != libc::EINTR && ret != libc::EINPROGRESS {
       return Err(get_err());
@@ -53,7 +55,6 @@ pub(crate) fn sockaddr_from_string(bytes: &str) -> Result<(sockaddr_un, usize)>
   Ok((sockaddr, mem::size_of::<sockaddr_un>()))
 }
 
-
 pub(crate) struct Emitter {
   env: Env,
   emit_ref: Option<Ref<()>>,
@@ -124,7 +125,6 @@ impl Emitter {
   }
 }
 
-
 pub(crate) struct HandleData {
   env: Env,
   this_ref: Ref<()>,
@@ -156,3 +156,106 @@ impl HandleData {
     Ok(())
   }
 }
+
+fn bind_socket(env: Env, fd: i32, domain: i32, port: JsNumber, ip: JsString) -> Result<JsNumber> {
+  let mut on: i32 = 1;
+  resolve_libc_err(unsafe {
+    libc::setsockopt(
+      fd,
+      libc::SOL_SOCKET,
+      libc::SO_REUSEADDR,
+      &mut on as *mut _ as *mut c_void,
+      mem::size_of::<i32>() as u32,
+    )
+  })?;
+
+  let mut on: i32 = 1;
+  resolve_libc_err(unsafe {
+    libc::setsockopt(
+      fd,
+      libc::SOL_SOCKET,
+      libc::SO_REUSEPORT,
+      &mut on as *mut _ as *mut c_void,
+      mem::size_of::<i32>() as u32,
+    )
+  })?;
+
+  // parse ip port
+  let ip = ip.into_utf8()?;
+  let ip_str = CString::new(ip.as_str()?.to_string().into_bytes())?;
+  let mut addr = unsafe { mem::MaybeUninit::<sockaddr_storage>::zeroed().assume_init() };
+  let addr_len: u32;
+  let port = port.get_int32()?;
+  if domain == libc::AF_INET {
+    resolve_uv_err(unsafe {
+      sys::uv_ip4_addr(
+        ip_str.as_c_str().as_ptr(),
+        port,
+        &mut addr as *mut _ as *mut sys::sockaddr_in,
+      )
+    })?;
+    addr_len = mem::size_of::<sys::sockaddr_in>() as u32;
+  } else {
+    resolve_uv_err(unsafe {
+      sys::uv_ip6_addr(
+        ip_str.as_c_str().as_ptr(),
+        port,
+        &mut addr as *mut _ as *mut sys::sockaddr_in6,
+      )
+    })?;
+    addr_len = mem::size_of::<sys::sockaddr_in6>() as u32;
+  };
+
+  // bind socket
+  resolve_libc_err(unsafe {
+    libc::bind(
+      fd,
+      &mut addr as *mut _ as *mut libc::sockaddr,
+      addr_len,
+    )
+  })?;
+
+  Ok(env.create_int32(fd)?)
+}
+
+#[allow(dead_code)]
+#[napi]
+fn socket_new_so_reuseport_fd(
+  env: Env,
+  domain: JsString,
+  port: JsNumber,
+  ip: JsString,
+) -> Result<JsNumber> {
+  let domain = domain.into_utf8()?;
+  let s = domain.as_str()?;
+  let domain = match s {
+    "ipv4" => libc::AF_INET,
+    "ipv6" => libc::AF_INET6,
+    _ => {
+      return Err(error(
+        "unexpected domain paramter, expect 'ipv4' or 'ipv6'".to_string(),
+      ))
+    }
+  };
+
+  // create socket and set SO_REUSEPORT
+  let fd = resolve_libc_err(unsafe { libc::socket(domain, libc::SOCK_STREAM, 0) })?;
+
+  let fd = match bind_socket(env, fd, domain, port, ip) {
+    Ok(fd) => fd,
+    Err(e) => {
+      close(fd).unwrap();
+      return Err(e)
+    }
+  };
+
+  Ok(fd)
+}
+
+#[allow(dead_code)]
+#[napi]
+fn socket_close(fd: JsNumber) -> Result<()> {
+  let fd = fd.get_int32()?;
+
+  close(fd)
+}