mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-03-05 20:50:17 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Lazyload lifecycle manager

Browse source
This commit is contained in:
Iván Ovejero 2024-11-07 13:45:26 +01:00
parent e4532a92b5
commit 647bd61d05
No known key found for this signature in database
2 changed files with 29 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
packages/cli/src/runners/runner-lifecycle-manager.ts
View file

@ -16,6 +16,10 @@ export type RunnerLifecycleEventMap = {
@Service() @Service()
export class RunnerLifecycleEvents extends TypedEmitter<RunnerLifecycleEventMap> {} export class RunnerLifecycleEvents extends TypedEmitter<RunnerLifecycleEventMap> {}
/**
* Responsible for launching a task runner if none available
* and shutting it down if idle for too long.
*/
@Service() @Service()
export class RunnerLifecycleManager { export class RunnerLifecycleManager {
private state: 'stopped' | 'starting' | 'running' | 'stopping' = 'stopped'; private state: 'stopped' | 'starting' | 'running' | 'stopping' = 'stopped';

38
packages/cli/src/runners/task-broker.service.ts
View file

@ -1,3 +1,4 @@
import { TaskRunnersConfig } from '@n8n/config';
import type { import type {
BrokerMessage, BrokerMessage,
RequesterMessage, RequesterMessage,
@ -6,12 +7,11 @@ import type {
} from '@n8n/task-runner'; } from '@n8n/task-runner';
import { ApplicationError, ensureError } from 'n8n-workflow'; import { ApplicationError, ensureError } from 'n8n-workflow';
import { nanoid } from 'nanoid'; import { nanoid } from 'nanoid';
import { Service } from 'typedi'; import Container, { Service } from 'typedi';
import { Logger } from '@/logging/logger.service'; import { Logger } from '@/logging/logger.service';
import { TaskRejectError } from './errors'; import { TaskRejectError } from './errors';
import { RunnerLifecycleManager } from './runner-lifecycle-manager';
export interface TaskRunner { export interface TaskRunner {
id: string; id: string;
@ -81,7 +81,7 @@ export class TaskBroker {
constructor( constructor(
private readonly logger: Logger, private readonly logger: Logger,
private readonly lifecycleManager: RunnerLifecycleManager, private readonly runnerConfig: TaskRunnersConfig,
) {} ) {}
expireTasks() { expireTasks() {
@ -557,17 +557,29 @@ export class TaskBroker {
} }
} }
async taskRequested(request: TaskRequest) { private get shouldManageLifecycle() {
try { return (
await this.lifecycleManager.ensureRunnerAvailable(); this.runnerConfig.mode === 'internal_childprocess' ||
} catch (e) { this.runnerConfig.mode === 'internal_launcher'
const error = ensureError(e); );
this.logger.error('Failed to start task runner', { error }); }
this.handleRunnerReject(request.requestId, `Task runner unavailable: ${error.message}`);
return;
}
this.lifecycleManager.updateLastActivityTime(); async taskRequested(request: TaskRequest) {
if (this.shouldManageLifecycle) {
const { RunnerLifecycleManager } = await import('@/runners/runner-lifecycle-manager');
const lifecycleManager = Container.get(RunnerLifecycleManager);
try {
await lifecycleManager.ensureRunnerAvailable();
} catch (e) {
const error = ensureError(e);
this.logger.error('Failed to start task runner', { error });
this.handleRunnerReject(request.requestId, `Task runner unavailable: ${error.message}`);
return;
}
lifecycleManager.updateLastActivityTime();
}
this.pendingTaskRequests.push(request); this.pendingTaskRequests.push(request);
this.settleTasks(); this.settleTasks();