升级queue为task

hydro/handler/judge.js

@@ -4,14 +4,12 @@ const problem = require('../model/problem');
 const builtin = require('../model/builtin');
 const contest = require('../model/contest');
 const user = require('../model/user');
+const task = require('../model/task');
 const bus = require('../service/bus');
-const queue = require('../service/queue');
 const {
     Route, Handler, Connection, ConnectionHandler,
 } = require('../service/server');
 
-queue.assert('judge');
-
 async function _postJudge(rdoc) {
     const accept = rdoc.status === builtin.STATUS_ACCEPTED;
     bus.publish('record_change', rdoc);
@@ -57,6 +55,7 @@ async function next(body) {
     rdoc = await record.update(body.rid, $set, $push);
     bus.publish('record_change', rdoc);
 }
+
 async function end(body) {
     let rdoc = await record.get(body.rid);
     const $set = {};
@@ -92,34 +91,10 @@ class JudgeHandler extends Handler {
     async get({ check = false }) {
         if (check) return;
         const tasks = [];
-        let rid = await queue.get('judge', false);
-        while (rid) {
-            const rdoc = await record.get(rid); // eslint-disable-line no-await-in-loop
-            let task;
-            if (!rdoc.input) {
-                // eslint-disable-next-line no-await-in-loop
-                const pdoc = await problem.getById(rdoc.pid);
-                task = {
-                    event: 'judge',
-                    rid,
-                    type: 0,
-                    pid: rdoc.pid,
-                    data: pdoc.data,
-                    lang: rdoc.lang,
-                    code: rdoc.code,
-                };
-            } else {
-                task = {
-                    event: 'pretest',
-                    rid,
-                    type: 0,
-                    input: rdoc.input,
-                    lang: rdoc.lang,
-                    code: rdoc.code,
-                };
-            }
-            tasks.push(task);
-            rid = await queue.get('judge', false); // eslint-disable-line no-await-in-loop
+        let t = await task.getFirst({ type: 'judge' });
+        while (t) {
+            tasks.push(t);
+            t = await task.getFirst({ type: 'judge' }); // eslint-disable-line no-await-in-loop
         }
         this.response.body = { tasks };
     }
@@ -133,6 +108,7 @@ class JudgeHandler extends Handler {
         await end(this.request.body);
     }
 }
+
 class JudgeConnectionHandler extends ConnectionHandler {
     async prepare() {
         this.checkPerm(PERM_JUDGE);
@@ -143,40 +119,16 @@ class JudgeConnectionHandler extends ConnectionHandler {
         else if (msg.key === 'end') {
             await end({ judger: this.user._id, ...msg });
             this.processing = null;
-            const rid = await queue.get('judge');
-            const rdoc = await record.get(rid);
-            let task;
-            if (!rdoc.input) {
-                // eslint-disable-next-line no-await-in-loop
-                const pdoc = await problem.getById(rdoc.pid);
-                task = {
-                    event: 'judge',
-                    rid,
-                    type: 0,
-                    pid: rdoc.pid,
-                    data: pdoc.data,
-                    lang: rdoc.lang,
-                    code: rdoc.code,
-                };
-            } else {
-                task = {
-                    event: 'pretest',
-                    rid,
-                    type: 0,
-                    input: rdoc.input,
-                    lang: rdoc.lang,
-                    code: rdoc.code,
-                };
-            }
-            this.send({ task });
-            this.processing = task.rid;
+            const t = await task.getFirst({ type: 'judge' });
+            this.send({ task: t });
+            this.processing = t;
         }
     }
 
     async cleanup() {
         if (this.processing) {
             await record.reset(this.processing);
-            queue.push('judge', this.processing);
+            task.add(this.processing);
         }
     }
 }
@@ -187,5 +139,5 @@ async function apply() {
 }
 
 global.Hydro.handler.judge = module.exports = {
-    JudgeHandler, JudgeConnectionHandler, apply,
+    JudgeHandler, JudgeConnectionHandler, apply, next, end,
 };

hydro/handler/problem.js

@@ -11,7 +11,6 @@ const {
     Route, Connection, Handler, ConnectionHandler,
 } = require('../service/server');
 const gridfs = require('../service/gridfs');
-const queue = require('../service/queue');
 const {
     NoProblemError, ProblemDataNotFoundError, BadRequestError,
     SolutionNotFoundError,
@@ -25,8 +24,6 @@ const {
     PERM_REPLY_PROBLEM_SOLUTION, PERM_LOGGEDIN,
 } = require('../permission');
 
-queue.assert('judge');
-
 class ProblemHandler extends Handler {
     async _prepare() {
         this.checkPerm(PERM_VIEW_PROBLEM);
@@ -122,7 +119,7 @@ class ProblemSubmitHandler extends ProblemDetailHandler {
         const rid = await record.add({
             uid: this.uid, lang, code, pid: this.pdoc._id,
         });
-        await queue.push('judge', rid);
+        await record.judge(rid);
         this.user.nSubmit++;
         this.response.body = { rid };
         this.response.redirect = `/r/${rid}`;
@@ -135,7 +132,7 @@ class ProblemPretestHandler extends ProblemDetailHandler {
         const rid = await record.add({
             uid: this.uid, lang, code, pid: this.pdoc._id, input,
         });
-        await queue.push('judge', rid);
+        await record.judge(rid);
         this.response.body = { rid };
     }
 }

hydro/handler/record.js

@@ -8,7 +8,6 @@ const record = require('../model/record');
 const contest = require('../model/contest');
 const user = require('../model/user');
 const bus = require('../service/bus');
-const queue = require('../service/queue');
 const {
     Route, Handler, Connection, ConnectionHandler,
 } = require('../service/server');
@@ -69,10 +68,7 @@ class RecordRejudgeHandler extends Handler {
     async post({ rid }) {
         this.checkPerm(PERM_REJUDGE);
         const rdoc = await record.get(rid);
-        if (rdoc) {
-            await record.reset(rid);
-            await queue.push('judge', rid);
-        }
+        if (rdoc) await record.rejudge(rid);
         this.back();
     }
 }

hydro/loader.js

@@ -26,8 +26,8 @@ async function handler() {
     for (const i of active) {
         if (i.handler) {
             console.time(`Handler init: ${i.id}`);
-            const module = {}; // eslint-disable-line no-unused-vars
-            const exports = {}; // eslint-disable-line no-unused-vars
+            const exports = {};
+            const module = { exports }; // eslint-disable-line no-unused-vars
             eval(i.handler);
             console.timeEnd(`Handler init: ${i.id}`);
         }
@@ -56,15 +56,28 @@ async function model() {
     for (const i of active) {
         if (i.model) {
             console.time(`Model init: ${i.id}`);
-            const module = {};
             const exports = {}; // eslint-disable-line no-unused-vars
-            eval(i.handler);
+            const module = { exports };
+            eval(i.model);
             if ((module.exports || {}).index) await module.exports.index();
             console.timeEnd(`Model init: ${i.id}`);
         }
     }
 }
 
+async function service() {
+    for (const i of active) {
+        if (i.service) {
+            console.time(`Service init: ${i.id}`);
+            const exports = {}; // eslint-disable-line no-unused-vars
+            const module = { exports };
+            eval(i.service);
+            if ((module.exports || {}).init) await module.exports.init();
+            console.timeEnd(`Service init: ${i.id}`);
+        }
+    }
+}
+
 async function load() {
     global.Hydro = {
         handler: {},
@@ -99,8 +112,8 @@ async function load() {
     for (const i of builtinLib) require(`./lib/${i}`);
     await locale();
     require('./service/gridfs');
-    require('./service/queue');
     const server = require('./service/server');
+    await service();
     const builtinModel = [
         'blacklist', 'builtin', 'contest', 'discussion', 'message',
         'opcount', 'problem', 'record', 'setting', 'solution',
@@ -122,6 +135,9 @@ async function load() {
     }
     const notfound = require('./handler/notfound');
     await notfound.apply();
+    for (const i in global.Hydro.service) {
+        if (global.Hydro.service[i].postInit) await global.Hydro.service[i].postInit();
+    }
     server.start();
 }
 

hydro/model/index.js

@@ -10,4 +10,7 @@ module.exports = {
     token: require('./token'),
     training: require('./training'),
     user: require('./user'),
+    task: require('./task'),
+    message: require('./message'),
+    setting: require('./setting'),
 };

hydro/model/record.js

@@ -1,7 +1,9 @@
 const _ = require('lodash');
 const { ObjectID } = require('bson');
-const { RecordNotFoundError } = require('../error');
 const { STATUS_WAITING } = require('./builtin').STATUS;
+const task = require('./task');
+const problem = require('./problem');
+const { RecordNotFoundError } = require('../error');
 const db = require('../service/db.js');
 
 const coll = db.collection('record');
@@ -75,6 +77,35 @@ function getUserInProblemMulti(uid, pid) {
     return coll.find({ owner: uid, pid });
 }
 
+async function judge(rid) {
+    const rdoc = await get(rid);
+    const pdoc = await problem.getById(rdoc.pid);
+    await task.add({
+        event: 'judge',
+        rid,
+        type: 0,
+        pid: rdoc.pid,
+        data: pdoc.data,
+        lang: rdoc.lang,
+        code: rdoc.code,
+    });
+}
+
+async function rejudge(rid) {
+    await reset(rid);
+    const rdoc = await get(rid);
+    const pdoc = await problem.getById(rdoc.pid);
+    await task.add({
+        event: 'judge',
+        rid,
+        type: 0,
+        pid: rdoc.pid,
+        data: pdoc.data,
+        lang: rdoc.lang,
+        code: rdoc.code,
+    });
+}
+
 module.exports = {
     add,
     get,
@@ -84,4 +115,6 @@ module.exports = {
     reset,
     getList,
     getUserInProblemMulti,
+    judge,
+    rejudge,
 };

hydro/model/task.js

@@ -0,0 +1,30 @@
+const db = require('../service/db.js');
+
+const coll = db.collection('task');
+
+async function add(task) {
+    const res = await coll.insertOne(task);
+    return res.insertedId;
+}
+
+function get(_id) {
+    return coll.findOne({ _id });
+}
+
+function count(query) {
+    return coll.find(query).count();
+}
+
+function del(_id) {
+    return coll.deleteOne({ _id });
+}
+
+async function getFirst(query) {
+    const res = await coll.find(query).sort('_id', 1).limit(1).toArray();
+    await coll.deleteOne({ _id: res[0]._id });
+    return res[0];
+}
+
+module.exports = {
+    add, get, del, count, getFirst,
+};

hydro/service/queue.js

@@ -1,27 +0,0 @@
-const q = {};
-module.exports = {
-    async connect() { }, // eslint-disable-line no-empty-function
-    async assert(queue) {
-        q[queue] = q[queue] || [];
-    },
-    async push(queue, data) {
-        q[queue].push(data);
-    },
-    async get(queue, wait = true) {
-        if (wait) {
-            while (!q[queue].length) {
-                await new Promise((resolve) => { // eslint-disable-line no-await-in-loop
-                    setTimeout(resolve, 100);
-                });
-            }
-            const data = q[queue][0];
-            q[queue].splice(0, 1);
-            return data;
-        }
-        if (!q[queue].length) return null;
-
-        const data = q[queue][0];
-        q[queue].splice(0, 1);
-        return data;
-    },
-};