diff --git a/build/start.js b/build/start.js index e94ba203f..7fe67bb2d 100644 --- a/build/start.js +++ b/build/start.js @@ -2,39 +2,30 @@ /* eslint-disable import/no-extraneous-dependencies */ const os = require('os'); const path = require('path'); -const cluster = require('cluster'); const fs = require('fs-extra'); const { filter } = require('lodash'); const argv = require('cac')().parse(); const hydro = require('hydrooj'); -if (!cluster.isMaster) { - // Forked by hydro - hydro.load().catch((e) => { - console.error(e); - process.exit(1); - }); -} else { - fs.ensureDirSync(path.resolve(os.homedir(), '.hydro')); - const addonPath = path.resolve(os.homedir(), '.hydro', 'addon.json'); - let addons = filter( - fs.readdirSync(path.resolve(process.cwd(), 'packages')), - (i) => i !== 'hydrooj', - ).map((i) => `@hydrooj/${i}`); - fs.writeFileSync(addonPath, JSON.stringify(addons, null, 2)); +fs.ensureDirSync(path.resolve(os.homedir(), '.hydro')); +const addonPath = path.resolve(os.homedir(), '.hydro', 'addon.json'); +let addons = filter( + fs.readdirSync(path.resolve(process.cwd(), 'packages')), + (i) => i !== 'hydrooj', +).map((i) => `@hydrooj/${i}`); +fs.writeFileSync(addonPath, JSON.stringify(addons, null, 2)); - try { - const ui = argv.options.ui || '@hydrooj/ui-default'; - require.resolve(ui); - addons = [ui, ...addons]; - } catch (e) { - console.error('Please also install @hydrooj/ui-default'); - } - - addons = Array.from(new Set(addons)); - for (const addon of addons) hydro.addon(addon); - (argv.args[0] === 'cli' ? hydro.loadCli : hydro.load)().catch((e) => { - console.error(e); - process.exit(1); - }); +try { + const ui = argv.options.ui || '@hydrooj/ui-default'; + require.resolve(ui); + addons = [ui, ...addons]; +} catch (e) { + console.error('Please also install @hydrooj/ui-default'); } + +addons = Array.from(new Set(addons)); +for (const addon of addons) hydro.addon(addon); +(argv.args[0] === 'cli' ? hydro.loadCli : hydro.load)().catch((e) => { + console.error(e); + process.exit(1); +}); diff --git a/packages/hydrojudge/package.json b/packages/hydrojudge/package.json index ac6d9af05..7e74bff70 100644 --- a/packages/hydrojudge/package.json +++ b/packages/hydrojudge/package.json @@ -1,7 +1,7 @@ { "name": "@hydrooj/hydrojudge", "bin": "bin/hydrojudge.js", - "version": "2.10.15", + "version": "2.11.0", "main": "package.json", "author": "undefined ", "repository": "https://github.com/hydro-dev/Hydro.git", diff --git a/packages/hydrojudge/src/sandbox.ts b/packages/hydrojudge/src/sandbox.ts index fa19ffd08..0d401b012 100644 --- a/packages/hydrojudge/src/sandbox.ts +++ b/packages/hydrojudge/src/sandbox.ts @@ -123,7 +123,6 @@ export async function del(fileId: string) { export async function run(execute, params?) { let result; - global.reqCount++; // eslint-disable-next-line no-return-await if (typeof execute === 'object') return await runMultiple(execute); try { diff --git a/packages/hydrojudge/src/service.ts b/packages/hydrojudge/src/service.ts index fba513071..38645d0f4 100644 --- a/packages/hydrojudge/src/service.ts +++ b/packages/hydrojudge/src/service.ts @@ -34,7 +34,7 @@ const logger = new Logger('judge'); async function postInit() { // Only start a single daemon - if (!global.Hydro.isFirstWorker) return; + if (process.env.NODE_APP_INSTANCE !== '0') return; const judge = require('./judge'); const { diff --git a/packages/hydrojudge/src/sysinfo.ts b/packages/hydrojudge/src/sysinfo.ts index 5daf3df64..17182b73f 100644 --- a/packages/hydrojudge/src/sysinfo.ts +++ b/packages/hydrojudge/src/sysinfo.ts @@ -55,11 +55,3 @@ export async function get() { info.stack = await stackSize(); return info; } - -declare global { - namespace NodeJS { - interface Global { - reqCount: number, - } - } -} diff --git a/packages/hydrooj/bin/hydrooj.js b/packages/hydrooj/bin/hydrooj.js index 517870bbd..86fa87f84 100755 --- a/packages/hydrooj/bin/hydrooj.js +++ b/packages/hydrooj/bin/hydrooj.js @@ -5,12 +5,12 @@ require('source-map-support').install(); const os = require('os'); const path = require('path'); -const cluster = require('cluster'); const fs = require('fs-extra'); const argv = require('cac')().parse(); const child = require('child_process'); const esbuild = require('esbuild'); +if (!process.env.NODE_APP_INSTANCE) process.env.NODE_APP_INSTANCE = '0'; const major = +process.version.split('.')[0].split('v')[1]; const minor = +process.version.split('.')[1]; @@ -55,75 +55,66 @@ function buildUrl(opts) { return mongourl; } -if (!cluster.isMaster) { +const hydroPath = path.resolve(os.homedir(), '.hydro'); +fs.ensureDirSync(hydroPath); +const addonPath = path.resolve(hydroPath, 'addon.json'); +if (!fs.existsSync(addonPath)) fs.writeFileSync(addonPath, '[]'); +let addons = JSON.parse(fs.readFileSync(addonPath).toString()); + +if (argv.args[0] === 'db') { + const dbConfig = fs.readFileSync(path.resolve(hydroPath, 'config.json'), 'utf-8'); + const url = buildUrl(JSON.parse(dbConfig)); + return child.spawn('mongo', [url], { stdio: 'inherit' }); +} + +if (!addons.includes('@hydrooj/ui-default')) { + try { + const ui = argv.options.ui || '@hydrooj/ui-default'; + require.resolve(ui); + addons.push(ui); + } catch (e) { + console.error('Please also install @hydrooj/ui-default'); + } +} + +if (argv.args[0] && argv.args[0] !== 'cli') { + const operation = argv.args[0]; + const arg1 = argv.args[1]; + const arg2 = argv.args[2]; + if (operation === 'addon') { + if (arg1 === 'create') { + fs.mkdirSync('/root/addon'); + child.execSync('yarn init -y', { cwd: '/root/addon' }); + fs.mkdirSync('/root/addon/templates'); + fs.mkdirSync('/root/addon/locales'); + fs.mkdirSync('/root/addon/public'); + addons.push('/root/addon'); + } else if (arg1 === 'add') { + for (let i = 0; i < addons.length; i++) { + if (addons[i] === arg2) { + addons.splice(i, 1); + break; + } + } + addons.push(arg2); + } else if (arg1 === 'remove') { + for (let i = 0; i < addons.length; i++) { + if (addons[i] === arg2) { + addons.splice(i, 1); + break; + } + } + } + addons = Array.from(new Set(addons)); + console.log('Current Addons: ', addons); + fs.writeFileSync(addonPath, JSON.stringify(addons, null, 2)); + } +} else { const hydro = require('../src/loader'); - // Forked by hydro - hydro.load().catch((e) => { + addons = Array.from(new Set(addons)); + for (const addon of addons) hydro.addon(addon); + (argv.args[0] === 'cli' ? hydro.loadCli : hydro.load)().catch((e) => { console.error(e); process.exit(1); }); -} else { - const hydroPath = path.resolve(os.homedir(), '.hydro'); - fs.ensureDirSync(hydroPath); - const addonPath = path.resolve(hydroPath, 'addon.json'); - if (!fs.existsSync(addonPath)) fs.writeFileSync(addonPath, '[]'); - let addons = JSON.parse(fs.readFileSync(addonPath).toString()); - - if (argv.args[0] === 'db') { - const dbConfig = fs.readFileSync(path.resolve(hydroPath, 'config.json'), 'utf-8'); - const url = buildUrl(JSON.parse(dbConfig)); - return child.spawn('mongo', [url], { stdio: 'inherit' }); - } - - if (!addons.includes('@hydrooj/ui-default')) { - try { - const ui = argv.options.ui || '@hydrooj/ui-default'; - require.resolve(ui); - addons.push(ui); - } catch (e) { - console.error('Please also install @hydrooj/ui-default'); - } - } - - if (argv.args[0] && argv.args[0] !== 'cli') { - const operation = argv.args[0]; - const arg1 = argv.args[1]; - const arg2 = argv.args[2]; - if (operation === 'addon') { - if (arg1 === 'create') { - fs.mkdirSync('/root/addon'); - child.execSync('yarn init -y', { cwd: '/root/addon' }); - fs.mkdirSync('/root/addon/templates'); - fs.mkdirSync('/root/addon/locales'); - fs.mkdirSync('/root/addon/public'); - addons.push('/root/addon'); - } else if (arg1 === 'add') { - for (let i = 0; i < addons.length; i++) { - if (addons[i] === arg2) { - addons.splice(i, 1); - break; - } - } - addons.push(arg2); - } else if (arg1 === 'remove') { - for (let i = 0; i < addons.length; i++) { - if (addons[i] === arg2) { - addons.splice(i, 1); - break; - } - } - } - addons = Array.from(new Set(addons)); - console.log('Current Addons: ', addons); - fs.writeFileSync(addonPath, JSON.stringify(addons, null, 2)); - } - } else { - const hydro = require('../src/loader'); - addons = Array.from(new Set(addons)); - for (const addon of addons) hydro.addon(addon); - (argv.args[0] === 'cli' ? hydro.loadCli : hydro.load)().catch((e) => { - console.error(e); - process.exit(1); - }); - } } diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json index 8934bafb2..2b3431e7d 100644 --- a/packages/hydrooj/package.json +++ b/packages/hydrooj/package.json @@ -1,6 +1,6 @@ { "name": "hydrooj", - "version": "2.33.18", + "version": "2.34.0", "bin": "bin/hydrooj.js", "main": "src/loader", "module": "src/loader", diff --git a/packages/hydrooj/src/entry/master.ts b/packages/hydrooj/src/entry/master.ts deleted file mode 100644 index 2f5e4fcd0..000000000 --- a/packages/hydrooj/src/entry/master.ts +++ /dev/null @@ -1,84 +0,0 @@ -/* eslint-disable no-await-in-loop */ -/* eslint-disable import/no-dynamic-require */ -import os from 'os'; -import path from 'path'; -import fs from 'fs-extra'; -import cac from 'cac'; -import { builtinModel } from './common'; -import { Entry } from '../loader'; -import { Logger } from '../logger'; -import options from '../options'; -import * as bus from '../service/bus'; -import db from '../service/db'; - -const argv = cac().parse(); -const logger = new Logger('entry/master'); -const tmpdir = path.resolve(os.tmpdir(), 'hydro'); -const lockfile = path.resolve(tmpdir, 'lock.json'); - -export async function load(call: Entry) { - fs.ensureDirSync(tmpdir); - if (fs.existsSync(lockfile) && !argv.options.ignorelock) { - try { - const file = require(lockfile); - process.kill(file.pid, 0); - logger.error(`Lockfile exists, pid=${file.pid}`); - process.exit(1); - } catch { - // Invalid lockfile or process not exist - } - } - const context = { - addons: global.addons, - pid: process.pid, - ppid: process.ppid, - }; - await fs.writeFile(lockfile, JSON.stringify(context)); - require('../lib/i18n'); - require('../utils'); - require('../error'); - const config = require('../options')(); - if (!config) { - logger.info('Starting setup'); - await call({ entry: 'setup', newProcess: true }).catch((err) => { - logger.error('Cannot start setup process.', err); - process.exit(1); - }); - } - bus.once('app/exit', () => { - fs.removeSync(lockfile); - fs.removeSync(path.resolve(os.tmpdir(), 'hydro', 'public')); - }); - const opts = options(); - await db.start(opts); - const modelSystem = require('../model/system'); - await modelSystem.runConfig(); - if (process.env.MINIO_ACCESS_KEY && process.env.MINIO_SECRET_KEY) { - await modelSystem.set('file.accessKey', process.env.MINIO_ACCESS_KEY); - await modelSystem.set('file.secretKey', process.env.MINIO_SECRET_KEY); - } - const storage = require('../service/storage'); - await storage.start(); - require('../service/monitor'); - for (const i of builtinModel) require(`../model/${i}`); - const scripts = require('../upgrade').default; - let dbVer = (await modelSystem.get('db.ver')) ?? 0; - const isFresh = !dbVer; - const expected = scripts.length; - while (dbVer < expected) { - logger.info('Upgrading database: from %d to %d', dbVer, expected); - const func = scripts[dbVer]; - if (typeof func !== 'function' || (isFresh && func.toString().includes('_FRESH_INSTALL_IGNORE'))) { - dbVer++; - continue; - } - const result = await func(); - if (!result) break; - dbVer++; - await modelSystem.set('db.ver', dbVer); - } - await bus.serial('app/started'); - return typeof argv.options.worker === 'number' - ? argv.options.worker - : await modelSystem.get('server.worker'); -} diff --git a/packages/hydrooj/src/entry/setup.ts b/packages/hydrooj/src/entry/setup.ts index 3bd5f19d7..0f9c1d2fe 100644 --- a/packages/hydrooj/src/entry/setup.ts +++ b/packages/hydrooj/src/entry/setup.ts @@ -106,13 +106,7 @@ async function post(ctx: Context) { fs.writeFileSync(path.resolve(os.homedir(), '.hydro', 'config.json'), JSON.stringify({ host, port, name, username, password, })); - ctx.body = `

This page will reload in 3 secs.

- `; - setTimeout(() => process.exit(0), 500); + ctx.body = '

Done! Please restart Hydro.

'; } catch (e) { ctx.body = `Error connecting to database: ${e.message}\n${e.stack}`; } diff --git a/packages/hydrooj/src/entry/worker.ts b/packages/hydrooj/src/entry/worker.ts index 53a226617..00ee22cf1 100644 --- a/packages/hydrooj/src/entry/worker.ts +++ b/packages/hydrooj/src/entry/worker.ts @@ -1,5 +1,8 @@ /* eslint-disable no-await-in-loop */ /* eslint-disable import/no-dynamic-require */ +import os from 'os'; +import fs from 'fs-extra'; +import path from 'path'; import cac from 'cac'; import { locale, template, lib, service, model, handler, script, setting, uistatic, @@ -12,17 +15,22 @@ import { Logger } from '../logger'; const logger = new Logger('loader/worker'); const detail = cac().parse().options.loaderDetail; +const tmpdir = path.resolve(os.tmpdir(), 'hydro'); export async function load() { + fs.ensureDirSync(tmpdir); + require('../lib/i18n'); + require('../utils'); + require('../error'); + const config = require('../options')(); + if (!config) { + logger.info('Starting setup'); + return require('./setup').load(); + } const pending = global.addons; const fail = []; const active = []; if (detail) logger.info('start'); - require('../lib/i18n'); - require('../utils'); - require('../error'); - require('../options'); - if (detail) logger.info('finish: options'); await Promise.all([ locale(pending, fail), template(pending, fail), @@ -80,9 +88,23 @@ export async function load() { if (detail) logger.info('finish: bus.serial(start)'); await server.start(); if (detail) logger.info('finish: server.start'); - setInterval(() => { - process.send({ event: 'stat', count: global.Hydro.stat.reqCount }); - global.Hydro.stat.reqCount = 0; - }, 30 * 1000); + if (process.env.NODE_APP_INSTANCE === '0') { + const scripts = require('../upgrade').default; + let dbVer = (await modelSystem.get('db.ver')) ?? 0; + const isFresh = !dbVer; + const expected = scripts.length; + while (dbVer < expected) { + const func = scripts[dbVer]; + if (typeof func !== 'function' || (isFresh && func.toString().includes('_FRESH_INSTALL_IGNORE'))) { + dbVer++; + continue; + } + logger.info('Upgrading database: from %d to %d', dbVer, expected); + const result = await func(); + if (!result) break; + dbVer++; + await modelSystem.set('db.ver', dbVer); + } + } return { active, fail }; } diff --git a/packages/hydrooj/src/handler/judge.ts b/packages/hydrooj/src/handler/judge.ts index ada873303..1d27c3de8 100644 --- a/packages/hydrooj/src/handler/judge.ts +++ b/packages/hydrooj/src/handler/judge.ts @@ -151,11 +151,11 @@ class JudgeConnectionHandler extends ConnectionHandler { if (this.processing) return; let t; while (!t) { - // eslint-disable-next-line no-await-in-loop - await sleep(100); if (this.closed) return; // eslint-disable-next-line no-await-in-loop t = await task.getFirst(this.query); + // eslint-disable-next-line no-await-in-loop + if (!t) await sleep(500); } this.send({ task: t }); this.processing = t; diff --git a/packages/hydrooj/src/handler/manage.ts b/packages/hydrooj/src/handler/manage.ts index a7781e7be..34c8d85d0 100644 --- a/packages/hydrooj/src/handler/manage.ts +++ b/packages/hydrooj/src/handler/manage.ts @@ -89,8 +89,7 @@ class SystemDashboardHandler extends SystemHandler { } async postRestart() { - const count = system.get('server.worker'); - process.send({ event: 'restart', payload: [count] }); + // TODO handle restart this.back(); } } diff --git a/packages/hydrooj/src/init.ts b/packages/hydrooj/src/init.ts index cedac2b6a..fb3785435 100644 --- a/packages/hydrooj/src/init.ts +++ b/packages/hydrooj/src/init.ts @@ -1,17 +1,16 @@ -import cluster from 'cluster'; - const versionNum = +process.version.replace(/v/gim, '').split('.')[0]; if (versionNum < 10) throw new Error('NodeJS >=10.4 required'); -else if (versionNum < 14 && cluster.isMaster) { +else if (versionNum < 14 && process.env.NODE_APP_INSTANCE === '0') { console.warn('NodeJS version <14, startup performance will be impacted.'); } + +console.log('Process', process.pid, 'running as', process.env.NODE_APP_INSTANCE === '0' ? 'master' : 'worker'); if (!global.Hydro) { global.Hydro = { version: { node: process.version, hydrooj: require('hydrooj/package.json').version, }, - stat: { reqCount: 0 }, handler: {}, // @ts-ignore service: {}, diff --git a/packages/hydrooj/src/interface.ts b/packages/hydrooj/src/interface.ts index ea224291b..2a67d2cc3 100644 --- a/packages/hydrooj/src/interface.ts +++ b/packages/hydrooj/src/interface.ts @@ -33,7 +33,6 @@ export interface SystemKeys { 'server.url': string, 'server.xff': string, 'server.xhost': string, - 'server.worker': number, 'server.port': number, 'server.language': string, 'limit.problem_files_max': number, @@ -529,6 +528,13 @@ export interface FileNode { meta?: ItemBucketMetadata, } +export interface EventDoc { + ack: string[]; + event: string; + payload: string; + expire: Date; +} + export interface Collections { 'blacklist': BlacklistDoc; 'contest': Tdoc; @@ -548,6 +554,7 @@ export interface Collections { 'task': Task; 'storage': FileNode; 'oplog': OplogDoc; + 'event': EventDoc; 'opcount': any; 'fs.chunks': any; 'fs.files': any; @@ -638,7 +645,6 @@ export interface HydroGlobal { Logger: typeof import('./logger').Logger, logger: typeof import('./logger').logger, locales: Record>, - isFirstWorker: boolean, } declare global { diff --git a/packages/hydrooj/src/loader.ts b/packages/hydrooj/src/loader.ts index d20f5e9bf..117d82115 100644 --- a/packages/hydrooj/src/loader.ts +++ b/packages/hydrooj/src/loader.ts @@ -6,13 +6,11 @@ import './init'; import './interface'; import os from 'os'; import path from 'path'; -import cluster from 'cluster'; import fs from 'fs-extra'; import './utils'; import cac from 'cac'; import { Logger } from './logger'; import './ui'; -import * as bus from './service/bus'; export * from './interface'; const argv = cac().parse(); @@ -23,83 +21,6 @@ if (argv.options.debug) { const logger = new Logger('loader'); logger.debug('%o', argv); -async function fork(args: string[] = []) { - const _args = process.argv.slice(2); - _args.push(...args, `--addons=${Buffer.from(JSON.stringify(global.addons)).toString('base64')}`); - cluster.setupMaster({ args: _args }); - return cluster.fork(); -} - -interface EntryConfig { - entry: string, - newProcess?: boolean, -} - -async function entry(config: EntryConfig) { - if (config.entry) { - if (config.newProcess) { - const p = await fork([`--entry=${config.entry}`]); - await new Promise((resolve, reject) => { - p.on('exit', (code, signal) => { - if (code === 0) resolve(null); - else reject(signal); - }); - p.on('error', (err: Error) => { - p.kill(); - reject(err); - }); - }); - } else { - const loader = require(`./entry/${config.entry}`); - return await loader.load(entry, global.addons); - } - } - return null; -} - -export type Entry = typeof entry; - -async function stopWorker() { - cluster.disconnect(); -} - -async function startWorker(cnt: number, createFirst = true) { - if (argv.options.single) { - await entry({ entry: 'worker' }); - } else { - await fork(createFirst ? ['--firstWorker'] : undefined); - for (let i = 1; i < cnt; i++) await fork(); - } -} - -async function reload(count = 1) { - logger.info('Reloading'); - await stopWorker(); - logger.info('Worker stopped'); - await startWorker(count); -} - -const shell = new Logger('shell'); -async function executeCommand(input: string) { - input = input.trim(); - // Clear the stack - setImmediate(async () => { - if (input === 'restart') return reload(); - if (input === 'exit' || input === 'quit' || input === 'shutdown') { - return process.kill(process.pid, 'SIGINT'); - } - try { - // eslint-disable-next-line no-eval - shell.info(await eval(input)); - } catch (e) { - shell.warn(e); - } - return true; - }); -} - -bus.on('message/reload', reload); -bus.on('message/run', executeCommand); process.on('unhandledRejection', logger.error); process.on('uncaughtException', logger.error); @@ -124,51 +45,11 @@ export function addon(addonPath: string, prepend = false) { export async function load() { addon(path.resolve(__dirname, '..'), true); Error.stackTraceLimit = 50; - if (cluster.isMaster || argv.options.startAsMaster) { - logger.info(`Master ${process.pid} Starting`); - const cnt = await entry({ entry: 'master' }); - logger.info('Master started'); - cluster.on('exit', (worker, code, signal) => { - logger.warn(`Worker ${worker.process.pid} ${worker.id} exit: ${code} ${signal}`); - if (code) startWorker(1); - }); - cluster.on('disconnect', (worker) => { - logger.info(`Worker ${worker.process.pid} ${worker.id} disconnected`); - }); - cluster.on('listening', (worker, address) => { - logger.success(`Worker ${worker.process.pid} ${worker.id} listening at `, address); - }); - cluster.on('online', (worker) => { - logger.success(`Worker ${worker.process.pid} ${worker.id} is online`); - }); - await startWorker(cnt); - } else { - global.addons = JSON.parse(Buffer.from(argv.options.addons as string, 'base64').toString()); - logger.info('%o', global.addons); - if (argv.options.entry) { - logger.info(`Worker ${process.pid} Starting as ${argv.options.entry}`); - await entry({ entry: argv.options.entry }); - logger.success(`Worker ${process.pid} Started as ${argv.options.entry}`); - } else { - if (argv.options.firstWorker) global.Hydro.isFirstWorker = true; - else global.Hydro.isFirstWorker = false; - logger.info(`Worker ${process.pid} Starting`); - await entry({ entry: 'worker' }); - logger.success(`Worker ${process.pid} Started`); - } - } + require('./entry/worker').load(); if (global.gc) global.gc(); } export async function loadCli() { - await entry({ entry: 'cli' }); + require('./entry/cli').load(); process.kill(process.pid, 'SIGINT'); } - -if (argv.options.pandora || require.main === module) { - const func = argv.args[0] === 'cli' ? load : loadCli; - func().catch((e) => { - logger.error(e); - process.exit(1); - }); -} diff --git a/packages/hydrooj/src/logger.ts b/packages/hydrooj/src/logger.ts index 19b233637..e9c4bedbc 100644 --- a/packages/hydrooj/src/logger.ts +++ b/packages/hydrooj/src/logger.ts @@ -78,13 +78,11 @@ export class Logger { }); } const msg = `${prefix} ${this.displayName} ${type.translate('en').format({ _inspect: true, payload })}`; - if (process.send) process.send({ event: 'message/log', payload: [msg] }); - else global.Hydro.service.bus.parallel('message/log', msg); + process.stdout.write(`${msg}\n`); return true; } const msg = `${prefix} ${this.displayName} ${this.format(...args)}`; - if (process.send) process.send({ event: 'message/log', payload: [msg] }); - else global.Hydro.service.bus.parallel('message/log', msg); + process.stdout.write(`${msg}\n`); return true; }; } diff --git a/packages/hydrooj/src/model/contest.ts b/packages/hydrooj/src/model/contest.ts index 81d2a3086..2439b695d 100644 --- a/packages/hydrooj/src/model/contest.ts +++ b/packages/hydrooj/src/model/contest.ts @@ -113,12 +113,11 @@ const acm: ContestRule = { for (const [rank, tsdoc] of rankedTsdocs) { const tsddict = {}; for (const item of tsdoc.detail || []) tsddict[item.pid] = item; - const row: ScoreboardRow = []; - row.push( + const row: ScoreboardRow = [ { type: 'string', value: rank.toString() }, { type: 'user', value: udict[tsdoc.uid].uname, raw: tsdoc.uid }, { type: 'string', value: tsdoc.accept || 0 }, - ); + ]; if (isExport) { row.push( { type: 'string', value: tsdoc.time || 0.0 }, @@ -133,9 +132,11 @@ const acm: ContestRule = { const colTime = accept ? doc.time : '-'; const colTimeStr = accept ? misc.formatSeconds(colTime) : '-'; if (isExport) { - row.push({ type: 'string', value: colAccepted }); - row.push({ type: 'string', value: colTime }); - row.push({ type: 'string', value: colTimeStr }); + row.push( + { type: 'string', value: colAccepted }, + { type: 'string', value: colTime }, + { type: 'string', value: colTimeStr }, + ); } else { row.push({ type: 'record', diff --git a/packages/hydrooj/src/model/setting.ts b/packages/hydrooj/src/model/setting.ts index bacc039fa..df8fc2f2f 100644 --- a/packages/hydrooj/src/model/setting.ts +++ b/packages/hydrooj/src/model/setting.ts @@ -175,7 +175,6 @@ SystemSetting( Setting('setting_server', 'server.center', 'https://hydro.undefined.moe:8443/center', 'text', 'server.center', '', FLAG_HIDDEN), Setting('setting_server', 'server.name', 'Hydro', 'text', 'server.name', 'Server Name'), Setting('setting_server', 'server.displayName', 'Hydro', 'text', 'server.name', 'Server Name (Global Display)', FLAG_PRO), - Setting('setting_server', 'server.worker', 1, 'number', 'server.worker', 'Server Workers Number', FLAG_PRO), Setting('setting_server', 'server.url', '/', 'text', 'server.url', 'Server BaseURL'), Setting('setting_server', 'server.cdn', '/', 'text', 'server.cdn', 'CDN Prefix'), Setting('setting_server', 'server.port', 8888, 'number', 'server.port', 'Server Port'), diff --git a/packages/hydrooj/src/model/task.ts b/packages/hydrooj/src/model/task.ts index 6d7b1345c..a6b5633af 100644 --- a/packages/hydrooj/src/model/task.ts +++ b/packages/hydrooj/src/model/task.ts @@ -1,4 +1,5 @@ import moment from 'moment-timezone'; +import { hostname } from 'os'; import { FilterQuery, ObjectID } from 'mongodb'; import { sleep } from '@hydrooj/utils/lib/utils'; import { BaseService, Task } from '../interface'; @@ -8,6 +9,7 @@ import * as bus from '../service/bus'; const logger = new Logger('model/task'); const coll = db.collection('task'); +const collEvent = db.collection('event'); async function getFirst(query: FilterQuery) { const q = { ...query }; @@ -45,7 +47,7 @@ class Consumer { await this.func(res); this.running = null; // eslint-disable-next-line no-await-in-loop - } else await sleep(100); + } else await sleep(1000); } catch (err) { logger.error(err); if (this.destoryOnError) this.destory(); @@ -145,6 +147,29 @@ bus.once('app/started', async () => { interval: [1, 'day'], }); } + await collEvent.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }); + const id = hostname() + process.env.NODE_APP_INSTANCE; + (async () => { + // eslint-disable-next-line no-constant-condition + while (true) { + // eslint-disable-next-line no-await-in-loop + const res = await collEvent.findOneAndUpdate( + { ack: { $not: { $elemMatch: { $eq: id } } } }, + { $push: { ack: id } }, + ); + if (res.value) bus.parallel(res.value.event, ...JSON.parse(res.value.payload)); + // eslint-disable-next-line no-await-in-loop + else await sleep(100); + } + })(); +}); +bus.on('bus/broadcast', (event, payload) => { + collEvent.insertOne({ + ack: [], + event, + payload: JSON.stringify(payload), + expire: new Date(Date.now() + 10000), + }); }); export default TaskModel; diff --git a/packages/hydrooj/src/service/bus.ts b/packages/hydrooj/src/service/bus.ts index 0ddaff6d7..2fc139f5b 100644 --- a/packages/hydrooj/src/service/bus.ts +++ b/packages/hydrooj/src/service/bus.ts @@ -1,6 +1,4 @@ /* eslint-disable no-await-in-loop */ -import cluster, { Worker } from 'cluster'; -import serialize from 'serialize-javascript'; import cac from 'cac'; import type { Db, FilterQuery, ObjectID, OnlyFieldsOfType, @@ -39,15 +37,11 @@ export interface EventMap extends Record { 'app/load/service': () => VoidReturn 'app/exit': () => VoidReturn - 'message/log': (message: string) => VoidReturn - 'message/reload': (count: number) => VoidReturn - 'message/run': (command: string) => VoidReturn - 'database/connect': (db: Db) => void 'database/config': () => VoidReturn 'system/setting': (args: Record) => VoidReturn - + 'bus/broadcast': (event: keyof EventMap, ...args: any[]) => VoidReturn 'monitor/update': (type: 'server' | 'judge', $set: any) => VoidReturn 'user/message': (uid: number, mdoc: MessageDoc) => void @@ -153,9 +147,9 @@ export function off(name: K, listener: EventMap[K]) { export async function parallel(name: K, ...args: Parameters): Promise { const tasks: Promise[] = []; - if (argv.options.showBus && name !== 'message/log') logger.debug('parallel: %s %o', name, args); + if (argv.options.showBus) logger.debug('parallel: %s %o', name, args); for (const callback of _hooks[name] || []) { - if (argv.options.busDetail && name !== 'message/log') logger.debug(callback.toString()); + if (argv.options.busDetail) logger.debug(callback.toString()); tasks.push(callback.apply(this, args)); } await Promise.all(tasks); @@ -166,7 +160,7 @@ export function emit(name: K, ...args: Parameters(name: K, ...args: Parameters): Promise { - if (argv.options.showBus && name !== 'message/log') logger.debug('serial: %s %o', name, args); + if (argv.options.showBus) logger.debug('serial: %s %o', name, args); const hooks = Array.from(_hooks[name] || []); for (const callback of hooks) { if (argv.options.busDetail) logger.debug(callback.toString()); @@ -175,7 +169,7 @@ export async function serial(name: K, ...args: Paramet } export async function bail(name: K, ...args: Parameters): Promise> { - if (argv.options.showBus && name !== 'message/log') logger.debug('bail: %s %o', name, args); + if (argv.options.showBus) logger.debug('bail: %s %o', name, args); const hooks = Array.from(_hooks[name] || []); for (const callback of hooks) { let result = callback.apply(this, args); @@ -186,37 +180,9 @@ export async function bail(name: K, ...args: Parameter } export function broadcast(event: K, ...payload: Parameters) { - // Process forked by pm2 would also have process.send - if (process.send && !cluster.isMaster) { - process.send({ - event: 'bus', - eventName: event, - payload: serialize(payload), - }); - } else parallel(event, ...payload); + return parallel('bus/broadcast', event, payload); } -async function messageHandler(worker: Worker, msg: any) { - if (!msg) msg = worker; - // eslint-disable-next-line no-eval - if (typeof msg.payload === 'string') msg.payload = eval(msg.payload); - if (msg.event) { - if (msg.event === 'bus') { - if (cluster.isMaster) { - for (const i in cluster.workers) { - cluster.workers[i].send(msg); - } - } - emit(msg.eventName, ...msg.payload); - } else if (msg.event === 'stat') { - global.Hydro.stat.reqCount += msg.count; - } else await emit(msg.event, ...msg.payload); - } -} - -process.on('message', messageHandler); -cluster.on('message', messageHandler); - global.Hydro.service.bus = { addListener, bail, broadcast, emit, on, off, once, parallel, prependListener, removeListener, serial, }; diff --git a/packages/hydrooj/src/service/monitor.ts b/packages/hydrooj/src/service/monitor.ts index ba54d7473..ee899d81e 100644 --- a/packages/hydrooj/src/service/monitor.ts +++ b/packages/hydrooj/src/service/monitor.ts @@ -1,4 +1,3 @@ -import cluster from 'cluster'; import crypto from 'crypto'; import superagent from 'superagent'; import { dump } from 'js-yaml'; @@ -57,7 +56,7 @@ export async function feedback(): Promise<[string, StatusUpdate]> { .then((res) => { if (res.body.updateUrl) system.set('server.center', res.body.updateUrl); }) - .catch(() => logger.warn('Cannot connect to hydro center.')); + .catch(() => logger.debug('Cannot connect to hydro center.')); return [mid, $update]; } @@ -66,7 +65,7 @@ export async function update() { const $set = { ...$update, updateAt: new Date(), - reqCount: global.Hydro.stat.reqCount, + reqCount: 0, }; await bus.serial('monitor/update', 'server', $set); await coll.updateOne( @@ -74,7 +73,6 @@ export async function update() { { $set }, { upsert: true }, ); - global.Hydro.stat.reqCount = 0; } export async function updateJudge(args) { @@ -87,7 +85,7 @@ export async function updateJudge(args) { ); } -if (cluster.isMaster) { +if (process.env.NODE_APP_INSTANCE === '0') { bus.on('app/started', async () => { sysinfo.get().then((info) => { coll.updateOne( diff --git a/packages/hydrooj/src/service/server.ts b/packages/hydrooj/src/service/server.ts index 446aee8b9..fc3224498 100644 --- a/packages/hydrooj/src/service/server.ts +++ b/packages/hydrooj/src/service/server.ts @@ -671,9 +671,8 @@ async function bail(name: string, ...args: any[]) { } async function handle(ctx, HandlerClass, checker) { - global.Hydro.stat.reqCount++; const args = { - domainId: 'system', ...ctx.params, ...ctx.query, ...ctx.request.body, + domainId: 'system', ...ctx.params, ...ctx.query, ...ctx.request.body, __start: Date.now(), }; const h = new HandlerClass(ctx); h.args = args; @@ -887,6 +886,7 @@ export function start() { app.use(router.routes()).use(router.allowedMethods()); server.listen(argv.options.port || port); logger.success('Server listening at: %d', argv.options.port || port); + if (process.send) process.send('ready'); // PM2 0-sec-down reload started = true; } diff --git a/packages/hydrooj/src/ui.ts b/packages/hydrooj/src/ui.ts index b294ea07a..fc3de52ec 100644 --- a/packages/hydrooj/src/ui.ts +++ b/packages/hydrooj/src/ui.ts @@ -1,5 +1,5 @@ /* eslint-disable @typescript-eslint/no-shadow */ -import cluster from 'cluster'; +import { Logger } from './logger'; import * as bus from './service/bus'; export namespace Progress { @@ -36,21 +36,28 @@ async function terminate() { process.exit(hasError ? 1 : 0); } process.on('SIGINT', terminate); -if (cluster.isMaster) { - bus.on('message/log', (message) => { - process.stdout.write(`${message}\n`); + +const shell = new Logger('shell'); +async function executeCommand(input: string) { + input = input.trim(); + // Clear the stack + setImmediate(async () => { + if (input === 'exit' || input === 'quit' || input === 'shutdown') { + return process.kill(process.pid, 'SIGINT'); + } + try { + // eslint-disable-next-line no-eval + shell.info(await eval(input)); + } catch (e) { + shell.warn(e); + } + return true; }); - if (process.env.NODE_ENV !== 'test') { - process.stdin.setEncoding('utf-8'); - if (process.stdin.setRawMode) process.stdin.setRawMode(false); - process.stdin.on('data', (buf) => { - const input = buf.toString(); - if (input[0] === '@') { - for (const i in cluster.workers) { - cluster.workers[i].send({ event: 'message/run', payload: [input.substr(1, input.length - 1)] }); - break; - } - } else bus.parallel('message/run', input); - }); - } } + +process.stdin.setEncoding('utf-8'); +if (process.stdin.setRawMode) process.stdin.setRawMode(false); +process.stdin.on('data', (buf) => { + const input = buf.toString(); + executeCommand(input); +}); diff --git a/packages/hydrooj/src/utils.ts b/packages/hydrooj/src/utils.ts index 64ce8729e..ac78c1c15 100644 --- a/packages/hydrooj/src/utils.ts +++ b/packages/hydrooj/src/utils.ts @@ -1,11 +1,5 @@ -import cluster from 'cluster'; import type { Logger } from './logger'; -if (!cluster.worker) { - // @ts-ignore - cluster.worker = { id: 0 }; -} - export function buildProjection(fields: string[]): Record { const o = {}; for (const k of fields) o[k] = 1; diff --git a/packages/hydrooj/test/db.ts b/packages/hydrooj/test/db.ts index 942ea750c..297921316 100644 --- a/packages/hydrooj/test/db.ts +++ b/packages/hydrooj/test/db.ts @@ -1,4 +1,3 @@ -import cluster from 'cluster'; import 'hydrooj/src/loader'; import * as bus from 'hydrooj/src/service/bus'; @@ -23,5 +22,4 @@ export async function dispose() { const db = require('hydrooj/src/service/db'); await db.stop(); bus.emit('app/exit'); - for (const key in cluster.workers) cluster.workers[key].destroy('SIGINT'); } diff --git a/packages/ui-default/backendlib/template.js b/packages/ui-default/backendlib/template.js index 3e2afee25..1d2f1718a 100644 --- a/packages/ui-default/backendlib/template.js +++ b/packages/ui-default/backendlib/template.js @@ -122,6 +122,8 @@ env.addGlobal('static_url', (assetName) => { }); // eslint-disable-next-line no-eval env.addGlobal('eval', eval); +env.addGlobal('Date', Date); +env.addGlobal('process', process); env.addGlobal('global', global); env.addGlobal('typeof', (o) => typeof o); env.addGlobal('datetimeSpan', misc.datetimeSpan); diff --git a/packages/ui-default/package.json b/packages/ui-default/package.json index 198b09d1d..f206b6b84 100644 --- a/packages/ui-default/package.json +++ b/packages/ui-default/package.json @@ -1,6 +1,6 @@ { "name": "@hydrooj/ui-default", - "version": "4.19.5", + "version": "4.20.0", "author": "undefined ", "license": "AGPL-3.0", "main": "hydro.js", diff --git a/packages/ui-default/templates/partials/footer.html b/packages/ui-default/templates/partials/footer.html index 7ab8370bb..9d9a2a067 100644 --- a/packages/ui-default/templates/partials/footer.html +++ b/packages/ui-default/templates/partials/footer.html @@ -74,6 +74,7 @@ {%- for html in (model.system.get('ui-default.footer_extra_html')|default('')).split('\n') -%}
  • {{ html|safe }}
    • {%- endfor -%} +
  • Worker {{ process.env.NODE_APP_INSTANCE }} in {{ Date.now() - handler.args.__start }}ms
  • Powered by Hydro v{{ global.Hydro.version.hydrooj }}
    • {% if model.system.get('server.pro') %}
  • Professional Version
    • diff --git a/packages/vjudge/package.json b/packages/vjudge/package.json index f6628363b..95a7051d4 100644 --- a/packages/vjudge/package.json +++ b/packages/vjudge/package.json @@ -1,6 +1,6 @@ { "name": "@hydrooj/vjudge", - "version": "1.2.3", + "version": "1.3.0", "description": "Submit problems to remote oj", "main": "package.json", "repository": "https://github.com/hydro-dev/Hydro.git", diff --git a/packages/vjudge/src/model.ts b/packages/vjudge/src/model.ts index 5ef2e886c..b0440716c 100644 --- a/packages/vjudge/src/model.ts +++ b/packages/vjudge/src/model.ts @@ -91,7 +91,7 @@ class Service { async function loadAccounts() { // Only start a single daemon - if (!global.Hydro.isFirstWorker) return; + if (process.env.NODE_APP_INSTANCE !== '0') return; const accounts = await coll.find().toArray(); for (const account of accounts) { if (!providers[account.type]) continue;