feat(core): Add basic execution concurrency limits in main mode #8458

netroy · 2024-01-26T12:26:54Z

This PR implements a very rudimentary concurrency limit in main mode, to be a bit more consistent with queue mode, and also to prevent the application from thrashing the event-loop when under high load.

Ticket: N8N-7052

Review / Merge checklist

PR title and summary are descriptive
Docs updated or follow-up ticket created.
Tests included

ivov

Nice work, this is going to be so useful!

Quick pass, not tested yet.

ivov · 2024-02-07T08:27:22Z

packages/cli/src/ActiveExecutions.ts

+		executionStatus: ExecutionStatus,
+	) {
+		// Wait here in-case execution concurrency limit is reached
+		await this.getQueue(executionData.executionMode).enqueue(executionId);


I'd've expected all this to be encapsulated? As in, a ConcurrencyService that contains the two queues and is smart enough to identify manual and production executions without being told what kind they are, e.g. this.concurrencyService.enqueue(execution)

ivov · 2024-02-07T08:27:45Z

packages/cli/src/ActiveExecutions.ts

@@ -21,6 +24,11 @@ import { Logger } from '@/Logger';

 @Service()
 export class ActiveExecutions {
+	private queues = {
+		manual: new ConcurrencyQueue(config.getEnv('executions.manualConcurrency')),
+		others: new ConcurrencyQueue(config.getEnv('executions.concurrency')),


Or is there a reason to keep it intentionally vague?

Suggested change

others: new ConcurrencyQueue(config.getEnv('executions.concurrency')),

production: new ConcurrencyQueue(config.getEnv('executions.concurrency')),

ivov · 2024-02-07T08:31:46Z

packages/cli/src/ActiveExecutions.ts

+	getRunningExecutionIds() {
+		const executions = Object.entries(this.activeExecutions);
+		return executions.filter(([, value]) => value.status === 'running').map(([id]) => id);
+	}
+


Are we using this?

ivov · 2024-02-07T08:32:48Z

packages/cli/src/ConcurrencyQueue.ts

@@ -0,0 +1,30 @@
+export class ConcurrencyQueue {
+	private waiting: Array<[string, () => void]> = [];


readonly

waiting is an execution status, is there a better term? paused? rateLimited? throttled? onHold?

This is a nice use case for labeled tuples: Array<[executionId: string, resolveFn: () => void]> = [];

ivov · 2024-02-07T08:33:08Z

packages/cli/src/ConcurrencyQueue.ts

+
+	constructor(private capacity: number) {}
+
+	async enqueue(id: string) {


Suggested change

async enqueue(id: string) {

async enqueue(executionId: string) {

ivov · 2024-02-07T08:48:45Z

packages/cli/test/unit/ConcurrencyQueue.test.ts

+});
+
+describe('ConcurrencyQueue', () => {
+	test('should limit concurrency', async () => {


We're testing state at various Ts but we're not testing that the concurrency limit is being enforced?

ivov · 2024-02-07T08:49:39Z

packages/editor-ui/src/plugins/i18n/locales/en.json

@@ -518,6 +518,7 @@
 	"executionDetails.confirmMessage.headline": "Delete Execution?",
 	"executionDetails.confirmMessage.message": "Are you sure that you want to delete the current execution?",
 	"executionDetails.deleteExecution": "Delete this execution",
+	"executionDetails.enqueuedMessage": "Execution hasn't started yet.",


As a user, I'd like to know why the execution hasn't started - if there's a problem or if this is behaving as intended, especially if I'm unaware of the concurrency capacity and I might be seeing this message for a while. Execution was rate-limited? Might need product input.

ivov · 2024-02-07T08:49:54Z

packages/editor-ui/src/plugins/i18n/locales/en.json

@@ -558,6 +559,7 @@
 	"executionsList.confirmMessage.message": "Are you sure that you want to delete the {numSelected} selected execution(s)?",
 	"executionsList.clearSelection": "Clear selection",
 	"executionsList.error": "Failed",
+	"executionsList.enqueued": "Enqueued",


A user not on queue mode might be confused by the reference to an implementation detail. Rate limited? Might need product input.

ivov · 2024-02-07T08:50:16Z

packages/cli/src/WorkflowRunner.ts

@@ -162,22 +200,18 @@ export class WorkflowRunner {
 			this.activeExecutions.attachResponsePromise(executionId, responsePromise);
 		}

-		if (this.executionsMode === 'queue' && data.executionMode !== 'manual') {
+		const runInMainProcess = this.executionsMode !== 'queue' || data.executionMode === 'manual';
+		if (!runInMainProcess) {


Let's flip for readability?

ivov · 2024-02-07T08:54:39Z

packages/cli/test/unit/ConcurrencyQueue.test.ts

+afterAll(() => {
+	jest.useRealTimers();
+});


Is this needed? If we don't do this, fake timers will bleed over to other suites?

netroy · 2024-02-08T15:15:35Z

jumping back on this now.

ivov · 2024-02-09T08:46:18Z

More ideas:

Prevent users from setting any concurrency limit lower than 1. As an exception, consider supporting -1 for unlimited concurrency, with a warning.
Add Public API endpoints to read queue size and update the production concurrency limit, so cloud can dynamically adjust it for heavy users in lower plans.
Priority queuing for selected workflows?
Warn on startup if concurrency limit is too high for system specs.
Add a timestamp to the queued item and on resuming warn if the time elapsed between enqueueing and resuming exceeded a threshold.

…xecution modes

netroy · 2024-06-03T17:31:27Z

closing in favor of #9453

n8n-assistant bot added core Enhancement outside /nodes-base and /editor-ui n8n team Authored by the n8n team labels Jan 26, 2024

netroy force-pushed the rudimentary-concurrency branch from b3d9a5c to 11b7b49 Compare January 26, 2024 12:52

RicardoE105 marked this pull request as ready for review January 27, 2024 12:33

netroy force-pushed the rudimentary-concurrency branch from 11b7b49 to f681b8d Compare January 27, 2024 21:02

netroy marked this pull request as draft January 28, 2024 05:27

netroy force-pushed the rudimentary-concurrency branch 2 times, most recently from 0b27aed to 0502e77 Compare January 29, 2024 20:19

netroy mentioned this pull request Jan 29, 2024

refactor(core): Couple of refactors on WorkflowRunner and ActiveExecutions (no-changelog) #8487

Merged

1 task

netroy changed the base branch from master to refactor-WorkflowRunner January 29, 2024 21:59

netroy force-pushed the rudimentary-concurrency branch from 0502e77 to 6a11f4a Compare January 29, 2024 22:00

netroy force-pushed the refactor-WorkflowRunner branch 3 times, most recently from 5686ad5 to f350b14 Compare January 30, 2024 12:07

netroy force-pushed the rudimentary-concurrency branch from 6a11f4a to bc61aa4 Compare January 30, 2024 12:11

netroy force-pushed the refactor-WorkflowRunner branch from c1c2bbc to 8f28c36 Compare February 6, 2024 12:59

netroy force-pushed the rudimentary-concurrency branch 2 times, most recently from 7b546d0 to 1e1ea50 Compare February 6, 2024 15:52

Base automatically changed from refactor-WorkflowRunner to master February 6, 2024 17:09

netroy force-pushed the rudimentary-concurrency branch from 1e1ea50 to fea3475 Compare February 6, 2024 19:26

netroy marked this pull request as ready for review February 6, 2024 19:26

netroy changed the title ~~feat(core): Make execution concurrency limits consistent across all execution modes~~ feat(core): Add basic execution concurrency limits in main mode Feb 6, 2024

netroy requested a review from krynble February 6, 2024 19:27

ivov reviewed Feb 7, 2024

View reviewed changes

netroy force-pushed the rudimentary-concurrency branch 2 times, most recently from 767c1ce to f4baf87 Compare February 20, 2024 14:40

netroy force-pushed the rudimentary-concurrency branch 2 times, most recently from 5e191f7 to 3406f40 Compare February 28, 2024 16:40

feat(core): Make execution concurrency limits consistent across all e…

0ef7883

…xecution modes

netroy force-pushed the rudimentary-concurrency branch from 3406f40 to 0ef7883 Compare April 17, 2024 09:27

address some PR comments

a1e1541

ivov mentioned this pull request May 17, 2024

perf(core): Introduce concurrency control for main mode #9453

Merged

netroy closed this Jun 3, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(core): Add basic execution concurrency limits in main mode #8458

feat(core): Add basic execution concurrency limits in main mode #8458

netroy commented Jan 26, 2024 •

edited

Loading

ivov left a comment

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

ivov Feb 7, 2024

netroy commented Feb 8, 2024

ivov commented Feb 9, 2024 •

edited

Loading

netroy commented Jun 3, 2024

	others: new ConcurrencyQueue(config.getEnv('executions.concurrency')),
	production: new ConcurrencyQueue(config.getEnv('executions.concurrency')),

		@@ -0,0 +1,30 @@
		export class ConcurrencyQueue {
		private waiting: Array<[string, () => void]> = [];


		constructor(private capacity: number) {}

		async enqueue(id: string) {

	async enqueue(id: string) {
	async enqueue(executionId: string) {

feat(core): Add basic execution concurrency limits in main mode #8458

feat(core): Add basic execution concurrency limits in main mode #8458

Conversation

netroy commented Jan 26, 2024 • edited Loading

Review / Merge checklist

ivov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

netroy commented Feb 8, 2024

ivov commented Feb 9, 2024 • edited Loading

netroy commented Jun 3, 2024

netroy commented Jan 26, 2024 •

edited

Loading

ivov commented Feb 9, 2024 •

edited

Loading