重いのでジョブキュー無効化

This commit is contained in:
syuilo 2019-02-10 11:44:08 +09:00
parent 7ca073aafd
commit 310371658b
No known key found for this signature in database
GPG key ID: BDC4C49D06AB9D69
6 changed files with 48 additions and 36 deletions

View file

@ -5,9 +5,9 @@ program
.version(pkg.version) .version(pkg.version)
.option('--no-daemons', 'Disable daemon processes (for debbuging)') .option('--no-daemons', 'Disable daemon processes (for debbuging)')
.option('--disable-clustering', 'Disable clustering') .option('--disable-clustering', 'Disable clustering')
.option('--disable-ap-queue', 'Disable creating job queue related to ap')
.option('--disable-queue', 'Disable job queue processing') .option('--disable-queue', 'Disable job queue processing')
.option('--only-queue', 'Pocessing job queue only') .option('--only-server', 'Run server only (without job queue)')
.option('--only-queue', 'Pocessing job queue only (without server)')
.option('--quiet', 'Suppress all logs') .option('--quiet', 'Suppress all logs')
.option('--verbose', 'Enable all logs') .option('--verbose', 'Enable all logs')
.option('--with-log-time', 'Include timestamp for each logs') .option('--with-log-time', 'Include timestamp for each logs')
@ -15,8 +15,7 @@ program
.option('--color', 'This option is a dummy for some external program\'s (e.g. forever) issue.') .option('--color', 'This option is a dummy for some external program\'s (e.g. forever) issue.')
.parse(process.argv); .parse(process.argv);
/*if (process.env.MK_DISABLE_AP_QUEUE)*/ program.disableApQueue = true; /*if (process.env.MK_DISABLE_QUEUE)*/ program.disableQueue = true;
if (process.env.MK_DISABLE_QUEUE) program.disableQueue = true;
if (process.env.MK_ONLY_QUEUE) program.onlyQueue = true; if (process.env.MK_ONLY_QUEUE) program.onlyQueue = true;
export { program }; export { program };

View file

@ -8,12 +8,13 @@ import handler from './processors';
import { queueLogger } from './logger'; import { queueLogger } from './logger';
const enableQueue = !program.disableQueue; const enableQueue = !program.disableQueue;
const enableQueueProcessing = !program.onlyServer && enableQueue;
const queueAvailable = config.redis != null; const queueAvailable = config.redis != null;
const queue = initializeQueue(); const queue = initializeQueue();
function initializeQueue() { function initializeQueue() {
if (queueAvailable) { if (queueAvailable && enableQueue) {
return new Queue('misskey', { return new Queue('misskey', {
redis: { redis: {
port: config.redis.port, port: config.redis.port,
@ -42,7 +43,7 @@ export function deliver(user: ILocalUser, content: any, to: any) {
to to
}; };
if (queueAvailable && !program.disableApQueue) { if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data) return queue.createJob(data)
.retries(8) .retries(8)
.backoff('exponential', 1000) .backoff('exponential', 1000)
@ -59,7 +60,7 @@ export function processInbox(activity: any, signature: httpSignature.IParsedSign
signature signature
}; };
if (queueAvailable && !program.disableApQueue) { if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data) return queue.createJob(data)
.retries(3) .retries(3)
.backoff('exponential', 500) .backoff('exponential', 500)
@ -70,47 +71,59 @@ export function processInbox(activity: any, signature: httpSignature.IParsedSign
} }
export function createExportNotesJob(user: ILocalUser) { export function createExportNotesJob(user: ILocalUser) {
if (!queueAvailable) throw 'queue unavailable'; const data = {
return queue.createJob({
type: 'exportNotes', type: 'exportNotes',
user: user user: user
}) };
.save();
if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data).save();
} else {
return handler({ data }, () => {});
}
} }
export function createExportFollowingJob(user: ILocalUser) { export function createExportFollowingJob(user: ILocalUser) {
if (!queueAvailable) throw 'queue unavailable'; const data = {
return queue.createJob({
type: 'exportFollowing', type: 'exportFollowing',
user: user user: user
}) };
.save();
if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data).save();
} else {
return handler({ data }, () => {});
}
} }
export function createExportMuteJob(user: ILocalUser) { export function createExportMuteJob(user: ILocalUser) {
if (!queueAvailable) throw 'queue unavailable'; const data = {
return queue.createJob({
type: 'exportMute', type: 'exportMute',
user: user user: user
}) };
.save();
if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data).save();
} else {
return handler({ data }, () => {});
}
} }
export function createExportBlockingJob(user: ILocalUser) { export function createExportBlockingJob(user: ILocalUser) {
if (!queueAvailable) throw 'queue unavailable'; const data = {
return queue.createJob({
type: 'exportBlocking', type: 'exportBlocking',
user: user user: user
}) };
.save();
if (queueAvailable && enableQueueProcessing) {
return queue.createJob(data).save();
} else {
return handler({ data }, () => {});
}
} }
export default function() { export default function() {
if (queueAvailable && enableQueue) { if (queueAvailable && enableQueueProcessing) {
queue.process(128, handler); queue.process(128, handler);
queueLogger.succ('Processing started'); queueLogger.succ('Processing started');
} }

View file

@ -48,7 +48,7 @@ export async function exportBlocking(job: bq.Job, done: any): Promise<void> {
if (blockings.length === 0) { if (blockings.length === 0) {
ended = true; ended = true;
job.reportProgress(100); if (job.reportProgress) job.reportProgress(100);
break; break;
} }
@ -74,7 +74,7 @@ export async function exportBlocking(job: bq.Job, done: any): Promise<void> {
blockerId: user._id, blockerId: user._id,
}); });
job.reportProgress(exportedCount / total); if (job.reportProgress) job.reportProgress(exportedCount / total);
} }
stream.end(); stream.end();

View file

@ -48,7 +48,7 @@ export async function exportFollowing(job: bq.Job, done: any): Promise<void> {
if (followings.length === 0) { if (followings.length === 0) {
ended = true; ended = true;
job.reportProgress(100); if (job.reportProgress) job.reportProgress(100);
break; break;
} }
@ -74,7 +74,7 @@ export async function exportFollowing(job: bq.Job, done: any): Promise<void> {
followerId: user._id, followerId: user._id,
}); });
job.reportProgress(exportedCount / total); if (job.reportProgress) job.reportProgress(exportedCount / total);
} }
stream.end(); stream.end();

View file

@ -48,7 +48,7 @@ export async function exportMute(job: bq.Job, done: any): Promise<void> {
if (mutes.length === 0) { if (mutes.length === 0) {
ended = true; ended = true;
job.reportProgress(100); if (job.reportProgress) job.reportProgress(100);
break; break;
} }
@ -74,7 +74,7 @@ export async function exportMute(job: bq.Job, done: any): Promise<void> {
muterId: user._id, muterId: user._id,
}); });
job.reportProgress(exportedCount / total); if (job.reportProgress) job.reportProgress(exportedCount / total);
} }
stream.end(); stream.end();

View file

@ -58,7 +58,7 @@ export async function exportNotes(job: bq.Job, done: any): Promise<void> {
if (notes.length === 0) { if (notes.length === 0) {
ended = true; ended = true;
job.reportProgress(100); if (job.reportProgress) job.reportProgress(100);
break; break;
} }
@ -83,7 +83,7 @@ export async function exportNotes(job: bq.Job, done: any): Promise<void> {
userId: user._id, userId: user._id,
}); });
job.reportProgress(exportedNotesCount / total); if (job.reportProgress) job.reportProgress(exportedNotesCount / total);
} }
await new Promise((res, rej) => { await new Promise((res, rej) => {