Weishen/Hydro
2
0 Fork 0
mirror of https://gitcode.com/gh_mirrors/hy/Hydro.git synced 2026-10-20 01:38:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

core&judge: allow concurrency setting within single connection

Browse Source
This commit is contained in:
undefined committed 2023-11-27 23:59:02 +08:00
1 parent b122a2bda9
commit 599a177f46
2 files changed
+29 -12
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/hydrojudge/src/hosts/hydro.ts
+5 -2
View File
@@ -178,8 +178,11 @@ export default class Hydro {
Authorization: `Bearer ${this.config.cookie.split('sid=')[1].split(';')[0]}`, Authorization: `Bearer ${this.config.cookie.split('sid=')[1].split(';')[0]}`,
}, },
}); });
const content = this.config.minPriority !== undefined const config: { prio?: number, concurrency?: number } = {};
? `{"key":"prio","prio":${this.config.minPriority}}` if (this.config.minPriority !== undefined) config.prio = this.config.minPriority;
if (this.config.concurrency !== undefined) config.concurrency = this.config.concurrency;
const content = Object.keys(config).length
? JSON.stringify({ key: 'config', ...config })
: '{"key":"ping"}'; : '{"key":"ping"}';
setInterval(() => this.ws?.send?.(content), 30000); setInterval(() => this.ws?.send?.(content), 30000);
this.ws.on('message', (data) => { this.ws.on('message', (data) => {
packages/hydrooj/src/handler/judge.ts
+24 -10
View File
@@ -3,7 +3,7 @@ import yaml from 'js-yaml';
import { omit } from 'lodash'; import { omit } from 'lodash';
import { ObjectId } from 'mongodb'; import { ObjectId } from 'mongodb';
import { import {
JudgeResultBody, ProblemConfigFile, RecordDoc, TestCase, JudgeResultBody, ProblemConfigFile, RecordDoc, Task, TestCase,
} from '../interface'; } from '../interface';
import { Logger } from '../logger'; import { Logger } from '../logger';
import * as builtin from '../model/builtin'; import * as builtin from '../model/builtin';
@@ -191,11 +191,12 @@ export class SubmissionDataDownloadHandler extends Handler {
class JudgeConnectionHandler extends ConnectionHandler { class JudgeConnectionHandler extends ConnectionHandler {
category = '#judge'; category = '#judge';
processing: any = null; processing: Task[] = [];
closed = false; closed = false;
query: any = { type: 'judge' }; query: any = { type: 'judge' };
rdocs: Record<string, RecordDoc> = {}; rdocs: Record<string, RecordDoc> = {};
ip: string; ip: string;
concurrency = 1;
async prepare() { async prepare() {
logger.info('Judge daemon connected from ', this.request.ip); logger.info('Judge daemon connected from ', this.request.ip);
@@ -211,7 +212,7 @@ class JudgeConnectionHandler extends ConnectionHandler {
} }
async newTask() { async newTask() {
if (this.processing) return; if (this.processing.length >= this.concurrency) return;
let t; let t;
let rdoc: RecordDoc; let rdoc: RecordDoc;
while (!t) { while (!t) {
@@ -225,7 +226,7 @@ class JudgeConnectionHandler extends ConnectionHandler {
} }
this.send({ task: { ...rdoc, ...t } }); this.send({ task: { ...rdoc, ...t } });
this.rdocs[rdoc._id.toHexString()] = rdoc; this.rdocs[rdoc._id.toHexString()] = rdoc;
this.processing = t; this.processing.push(t);
const $set = { status: builtin.STATUS.STATUS_FETCHED }; const $set = { status: builtin.STATUS.STATUS_FETCHED };
rdoc = await record.update(t.domainId, t.rid, $set, {}); rdoc = await record.update(t.domainId, t.rid, $set, {});
bus.broadcast('record/change', rdoc, $set, {}); bus.broadcast('record/change', rdoc, $set, {});
@@ -243,7 +244,7 @@ class JudgeConnectionHandler extends ConnectionHandler {
if (msg.key === 'next') await next({ ...msg, rdoc }); if (msg.key === 'next') await next({ ...msg, rdoc });
if (msg.key === 'end') { if (msg.key === 'end') {
if (!msg.nop) await end({ judger: this.user._id, ...msg, rdoc }).catch((e) => logger.error(e)); if (!msg.nop) await end({ judger: this.user._id, ...msg, rdoc }).catch((e) => logger.error(e));
this.processing = null; this.processing = this.processing.filter((t) => t.rid.toHexString() !== msg.rid);
delete this.rdocs[msg.rid]; delete this.rdocs[msg.rid];
await this.newTask(); await this.newTask();
} }
@@ -253,16 +254,29 @@ class JudgeConnectionHandler extends ConnectionHandler {
this.query.priority = { $gt: msg.prio }; this.query.priority = { $gt: msg.prio };
} else if (msg.key === 'lang' && msg.lang instanceof Array && msg.lang.every((i) => typeof i === 'string')) { } else if (msg.key === 'lang' && msg.lang instanceof Array && msg.lang.every((i) => typeof i === 'string')) {
this.query.lang = { $in: msg.lang }; this.query.lang = { $in: msg.lang };
} else if (msg.key === 'config') {
if (Number.isSafeInteger(msg.prio)) {
this.query.priority = { $gt: msg.prio };
}
if (Number.isSafeInteger(msg.concurrency) && msg.concurrency > 0) {
const old = this.concurrency;
this.concurrency = msg.concurrency;
if (old <= this.concurrency) {
for (let i = old; i < this.concurrency; i++) {
await this.newTask(); // eslint-disable-line no-await-in-loop
}
}
}
} }
} }
async cleanup() { async cleanup() {
logger.info('Judge daemon disconnected from ', this.request.ip);
if (this.processing) {
await record.reset(this.processing.domainId, this.processing.rid, false);
await task.add(this.processing);
}
this.closed = true; this.closed = true;
logger.info('Judge daemon disconnected from ', this.request.ip);
await Promise.all(this.processing.map(async (t) => {
await record.reset(t.domainId, t.rid, false);
return await task.add(t);
}));
} }
} }