multi thread socket

api/index.js

@@ -9,7 +9,7 @@ import { vd } from './interface/vd.js'
 import { hawk } from './interface/hawk.js'
 import * as vdb from './interface/vdb.js'
 import { socket as stateSocket } from './interface/state.js'
-import { ioRaw, connectionLimit } from './interface/io.js'
+import { ioRaw, setupConnectionLimit, setupConnectionMaster } from './interface/io.js'
 
 import snake from './snake.js'
 
@@ -19,10 +19,17 @@ import httpAPI from './http.js'
 const PARALLEL = 16
 const INTERVAL = 1000 * 60 * 5
 
+const wait = ms => new Promise(resolve => setTimeout(resolve, ms))
+
 if (cluster.isPrimary) {
   console.log('starting spider')
   spider({ INTERVAL })
   ant({ INTERVAL })
+  const server = http.createServer()
+  setupConnectionMaster(server)
+  await wait(2000)
+  server.listen(8001)
+  console.log('listening on 8001')
 } else {
   console.log('oh no, I am a worker')
   stateSocket.on('log', log => ioRaw.to('state').emit('stateLog', log))
@@ -33,6 +40,5 @@ if (cluster.isPrimary) {
   snake()
   hawk(ioRaw)
   ioRaw.on('connection', connect({ PARALLEL, INTERVAL }))
-  server.listen(8001)
-  connectionLimit(server, [ioRaw, vd])
+  setupConnectionLimit()
 }

api/interface/io.ts

@@ -2,6 +2,7 @@ import cluster from 'node:cluster'
 import type { Server as HTTPServer } from 'node:http'
 
 import { Server } from 'socket.io'
+import { setupMaster, setupWorker } from '@socket.io/sticky'
 
 import * as vdb from './vdb.js'
 
@@ -36,12 +37,10 @@ type Message = {
   emitInfoArray?: boolean
   sharedDB?: ShareDB
   updateVDB?: true
-  close?: true
   online?: number
 }
 
 type MessageToPrimary = {
-  overLimit?: true
   online?: number
 }
 
@@ -118,28 +117,46 @@ const rawEmit = (emit: Emit, to: To) => {
   }
 }
 
+const fork = () => {
+  if (Object.keys(cluster.workers).length < 6) {
+    cluster.fork()
+  }
+}
+
 if (cluster.isPrimary) {
+  cluster.setupPrimary({
+    serialization: 'advanced'
+  })
+
+  cluster.on('disconnect', worker => {
+    console.log('Worker disconnect', worker.id)
+    fork()
+  })
+
+  cluster.on('exit', worker => {
+    console.log('Worker exited', worker.id)
+    fork()
+  })
+
   let online = 0
-  cluster.fork()
 
-  setInterval(() => {
-    sendMessageToWorkers({ close: true })
-    cluster.fork()
-  }, 1000 * 60 * 60)
+  fork()
+  fork()
+  fork()
+  fork()
+  fork()
+  fork()
 
   setInterval(async () => {
     sendMessageToWorkers({ online })
   }, ONLINE_REPORT_INTERVAL * 0.7)
 
   cluster.on('message', (_, message: MessageToPrimary) => {
-    if (message.overLimit) {
-      cluster.fork()
-    }
     if (message.online !== undefined) {
       online += message.online
       setTimeout(() => {
         online -= message.online
-      }, ONLINE_REPORT_INTERVAL);
+      }, ONLINE_REPORT_INTERVAL)
     }
   })
 } else {
@@ -197,45 +214,10 @@ export const updateVDB = () => {
   }
 }
 
-const sumOnline = (ios: Server[]) => ios.reduce((sum, io) => sum + (io.engine as any).clientsCount as number, 0)
-
-export const connectionLimit = (server: HTTPServer, ios: Server[]) => {
-  const worker = cluster.worker
-  let closed = false
-
-  const close = () => {
-    console.log('Closing Worker ', worker.id)
-    closed = true
-    server.close()
-    setInterval(() => {
-      if (sumOnline(ios) === 0) {
-        console.log('Killing Worker ', worker.id)
-        worker.kill()
-      }
-    }, 1000 * 60)
-    setTimeout(() => {
-      console.log('Timeout Worker ', worker.id)
-      worker.kill()
-    }, 1000 * 60 * 60 * 24)
-  }
+export const setupConnectionMaster = (server: HTTPServer) => {
+  setupMaster(server)
+}
 
-  ios.forEach(io => {
-    io.on('connection', () => {
-      if (closed) {
-        return
-      }
-      if (sumOnline(ios) >= 128) {
-        sendMessageToPrimary({ overLimit: true })
-        close()
-      }
-    })
-  })
-  process.on('message', ({ close: c }: Message) => {
-    if (c) {
-      if (closed) {
-        return
-      }
-      close()
-    }
-  })
+export const setupConnectionLimit = () => {
+  setupWorker(ioRaw)
 }

api/socket.js

@@ -28,12 +28,12 @@ const wsRouter = ({ socket }) => ([key, ...rest], map = []) => {
     fullInfo: async () => {
       const vtbs = await vdb.getPure()
       return (await Promise.all(vtbs.map(({ mid }) => mid).map(async mid => {
-          const i = await info.get(mid)
-          if (i) {
-            return { ...i, mid }
-          }
-          return undefined
-        })))
+        const i = await info.get(mid)
+        if (i) {
+          return { ...i, mid }
+        }
+        return undefined
+      })))
         .filter(Boolean)
     },
     async guardMacroK([week = false]) {