fade22015f
* Adds new features table to top of v4 upgrade guide * Adds wait idempotency to wait-until, wait-for, and wait-for-token pages * Adds new priority docs page and updates the v4 upgrade guide * Adds new task lifecycle hooks * Removes the message about requiring tasks to be exported * Adds new global lifecycle hooks section * Moves sections from upgrade guide into the table * Adds hidden task page * Improves the global lifecycle hooks section * Updates middleware and locals section * Adds new useWaitToken page to the react hooks section * Adds a new ai.tool section * Moves Docker (legacy) page into self-hosting section * Removes known issues from v4 upgrade guide * Replace “toolTask” with “ai.tool” in the Streams page example * Renames guide to “Migrating from v3” and adds redirect * Remove references to v4 * Removes changelog from migration guide * The installation guide now references `@latest update` * Changes all references from `/sdk/v3` to `/sdk` * Updates @v4-beta to @latest * Fixed broken link * Fixes broken link * Adds an upgrade to v4 using AI section * Fixes 2 broken links * Adds an entry for targetting preview branches * Updates the run statuses * Adds boolean helpers section to the runs and realtime pages * Updates the concurrency page * Updates the test page to include the new options * Adds SDK and curl options for the preview branch targeting * Updates new bulk actions page * Remove the releasing concurrency section * Got rid of some more @v4-beta mentions * Improved rate limit docs * Improved migrating docs * Removed commented sections of the docs * useWaitToken hook * Fixed the description * Fix for missing test image --------- Co-authored-by: Matt Aitken <matt@mattaitken.com> Co-authored-by: Dan <8297864+D-K-P@users.noreply.github.com>
311 lines
11 KiB
Plaintext
311 lines
11 KiB
Plaintext
---
|
|
title: "Concurrency & Queues"
|
|
description: "Configure what you want to happen when there is more than one run at a time."
|
|
---
|
|
|
|
When you trigger a task, it isn't executed immediately. Instead, the task [run](/runs) is placed into a queue for execution. By default, each task gets its own queue with unbounded concurrency—meaning the task runs as soon as resources are available, subject only to the overall concurrency limits of your environment. If you need more control (for example, to limit concurrency or share limits across multiple tasks), you can define a custom queue as described later in this document.
|
|
|
|
Controlling concurrency is useful when you have a task that can't be run concurrently, or when you want to limit the number of runs to avoid overloading a resource.
|
|
|
|
It's important to note that only actively executing runs count towards concurrency limits. Runs that are delayed or waiting in a queue do not consume concurrency slots until they begin execution.
|
|
|
|
## Default concurrency
|
|
|
|
By default, all tasks have an unbounded concurrency limit, limited only by the overall concurrency limits of your environment. This means that each task could possibly "fill up" the entire
|
|
concurrency limit of your environment.
|
|
|
|
Each individual queue has a maximum concurrency limit equal to your environment's base concurrency limit. If you don't explicitly set a queue's concurrency limit, it will default to your environment's base concurrency limit.
|
|
|
|
<Note>
|
|
Your environment has a base concurrency limit and a burstable limit (default burst factor of 2.0x the base limit). Individual queues are limited by the base concurrency limit, not the burstable limit. For example, if your base limit is 10, your environment can burst up to 20 concurrent runs, but any single queue can have at most 10 concurrent runs. If you're a paying customer you can request higher limits by [contacting us](https://www.trigger.dev/contact).
|
|
</Note>
|
|
|
|
## Setting task concurrency
|
|
|
|
You can set the concurrency limit for a task by setting the `concurrencyLimit` property on the task's queue. This limits the number of runs that can be executing at any one time:
|
|
|
|
```ts /trigger/one-at-a-time.ts
|
|
// This task will only run one at a time
|
|
export const oneAtATime = task({
|
|
id: "one-at-a-time",
|
|
queue: {
|
|
concurrencyLimit: 1,
|
|
},
|
|
run: async (payload) => {
|
|
//...
|
|
},
|
|
});
|
|
```
|
|
|
|
This is useful if you need to control access to a shared resource, like a database or an API that has rate limits.
|
|
|
|
## Sharing concurrency between tasks
|
|
|
|
As well as putting queue settings directly on a task, you can define a queue and reuse it across multiple tasks. This allows you to share the same concurrency limit:
|
|
|
|
```ts /trigger/queue.ts
|
|
export const myQueue = queue({
|
|
name: "my-queue",
|
|
concurrencyLimit: 1,
|
|
});
|
|
|
|
export const task1 = task({
|
|
id: "task-1",
|
|
queue: myQueue,
|
|
run: async (payload: { message: string }) => {
|
|
// ...
|
|
},
|
|
});
|
|
|
|
export const task2 = task({
|
|
id: "task-2",
|
|
queue: myQueue,
|
|
run: async (payload: { message: string }) => {
|
|
// ...
|
|
},
|
|
});
|
|
```
|
|
|
|
In this example, `task1` and `task2` share the same queue, so only one of them can run at a time.
|
|
|
|
## Setting the concurrency when you trigger a run
|
|
|
|
When you trigger a task you can override the concurrency limit. This is really useful if you sometimes have high priority runs.
|
|
|
|
The task:
|
|
|
|
```ts /trigger/override-concurrency.ts
|
|
export const generatePullRequest = task({
|
|
id: "generate-pull-request",
|
|
queue: {
|
|
//normally when triggering this task it will be limited to 1 run at a time
|
|
concurrencyLimit: 1,
|
|
},
|
|
run: async (payload) => {
|
|
//todo generate a PR using OpenAI
|
|
},
|
|
});
|
|
```
|
|
|
|
Triggering from your backend and overriding the concurrency:
|
|
|
|
```ts app/api/push/route.ts
|
|
import { generatePullRequest } from "~/trigger/override-concurrency";
|
|
|
|
export async function POST(request: Request) {
|
|
const data = await request.json();
|
|
|
|
if (data.branch === "main") {
|
|
//trigger the task, with a different queue
|
|
const handle = await generatePullRequest.trigger(data, {
|
|
queue: {
|
|
//the "main-branch" queue will have a concurrency limit of 10
|
|
//this triggered run will use that queue
|
|
name: "main-branch", // Make sure to change the queue name or the task concurrency limit will be updated
|
|
concurrencyLimit: 10,
|
|
},
|
|
});
|
|
|
|
return Response.json(handle);
|
|
} else {
|
|
//triggered with the default (concurrency of 1)
|
|
const handle = await generatePullRequest.trigger(data);
|
|
return Response.json(handle);
|
|
}
|
|
}
|
|
```
|
|
|
|
## Concurrency keys and per-tenant queuing
|
|
|
|
If you're building an application where you want to run tasks for your users, you might want a separate queue for each of your users (or orgs, projects, etc.).
|
|
|
|
You can do this by using `concurrencyKey`. It creates a separate queue for each value of the key.
|
|
|
|
Your backend code:
|
|
|
|
```ts app/api/pr/route.ts
|
|
import { generatePullRequest } from "~/trigger/override-concurrency";
|
|
|
|
export async function POST(request: Request) {
|
|
const data = await request.json();
|
|
|
|
if (data.isFreeUser) {
|
|
//free users can only have 1 PR generated at a time
|
|
const handle = await generatePullRequest.trigger(data, {
|
|
queue: {
|
|
//every free user gets a queue with a concurrency limit of 1
|
|
name: "free-users",
|
|
concurrencyLimit: 1,
|
|
},
|
|
concurrencyKey: data.userId,
|
|
});
|
|
|
|
//return a success response with the handle
|
|
return Response.json(handle);
|
|
} else {
|
|
//trigger the task, with a different queue
|
|
const handle = await generatePullRequest.trigger(data, {
|
|
queue: {
|
|
//every paid user gets a queue with a concurrency limit of 10
|
|
name: "paid-users",
|
|
concurrencyLimit: 10,
|
|
},
|
|
concurrencyKey: data.userId,
|
|
});
|
|
|
|
//return a success response with the handle
|
|
return Response.json(handle);
|
|
}
|
|
}
|
|
```
|
|
|
|
## Concurrency and subtasks
|
|
|
|
When you trigger a task that has subtasks, the subtasks will not inherit the concurrency settings of the parent task. Unless otherwise specified, subtasks will run on their own queue
|
|
|
|
```ts /trigger/subtasks.ts
|
|
export const parentTask = task({
|
|
id: "parent-task",
|
|
run: async (payload) => {
|
|
//trigger a subtask
|
|
await subtask.triggerAndWait(payload);
|
|
},
|
|
});
|
|
|
|
// This subtask will run on its own queue
|
|
export const subtask = task({
|
|
id: "subtask",
|
|
run: async (payload) => {
|
|
//...
|
|
},
|
|
});
|
|
```
|
|
|
|
## Waits and concurrency
|
|
|
|
With our [task checkpoint system](/how-it-works#the-checkpoint-resume-system), tasks can wait at various waitpoints (like waiting for subtasks to complete, delays, or external events). The way this system interacts with the concurrency system is important to understand.
|
|
|
|
Concurrency is only released when a run reaches a waitpoint and is checkpointed. When a run is checkpointed, it transitions to the `WAITING` state and releases its concurrency slot back to both the queue and the environment, allowing other runs to execute or resume.
|
|
|
|
This means that:
|
|
- Only actively executing runs count towards concurrency limits
|
|
- Runs in the `WAITING` state (checkpointed at waitpoints) do not consume concurrency slots
|
|
- You can have more runs in the `WAITING` state than your queue's concurrency limit
|
|
- When a waiting run resumes (e.g., when a subtask completes), it must re-acquire a concurrency slot
|
|
|
|
For example, if you have a queue with a `concurrencyLimit` of 1:
|
|
- You can only have exactly 1 run executing at a time
|
|
- You may have multiple runs in the `WAITING` state that belong to that queue
|
|
- When the executing run reaches a waitpoint and checkpoints, it releases its slot
|
|
- The next queued run can then begin execution
|
|
|
|
<Note>
|
|
We sometimes refer to the parent task as the "parent" and the subtask as the "child". Subtask and
|
|
child task are used interchangeably. We apologize for the confusion.
|
|
</Note>
|
|
|
|
### Waiting for a subtask on a different queue
|
|
|
|
When a parent task triggers and waits for a subtask on a different queue, the parent task will checkpoint and release its concurrency slot once it reaches the wait point. This prevents environment deadlocks where all concurrency slots would be occupied by waiting tasks.
|
|
|
|
```ts /trigger/waiting.ts
|
|
export const parentTask = task({
|
|
id: "parent-task",
|
|
queue: {
|
|
concurrencyLimit: 1,
|
|
},
|
|
run: async (payload) => {
|
|
//trigger a subtask and wait for it to complete
|
|
await subtask.triggerAndWait(payload);
|
|
// The parent task checkpoints here and releases its concurrency slot
|
|
// allowing other tasks to execute while waiting
|
|
},
|
|
});
|
|
|
|
export const subtask = task({
|
|
id: "subtask",
|
|
run: async (payload) => {
|
|
//...
|
|
},
|
|
});
|
|
```
|
|
|
|
When the parent task reaches the `triggerAndWait` call, it checkpoints and transitions to the `WAITING` state, releasing its concurrency slot back to both its queue and the environment. Once the subtask completes, the parent task will resume and re-acquire a concurrency slot.
|
|
|
|
### Waiting for a subtask on the same queue
|
|
|
|
When a parent task and subtask share the same queue, the checkpointing behavior ensures that recursive task execution can proceed without deadlocks, up to the queue's concurrency limit.
|
|
|
|
```ts /trigger/waiting-same-queue.ts
|
|
export const myQueue = queue({
|
|
name: "my-queue",
|
|
concurrencyLimit: 1,
|
|
});
|
|
|
|
export const parentTask = task({
|
|
id: "parent-task",
|
|
queue: myQueue,
|
|
run: async (payload) => {
|
|
//trigger a subtask and wait for it to complete
|
|
await subtask.triggerAndWait(payload);
|
|
},
|
|
});
|
|
|
|
export const subtask = task({
|
|
id: "subtask",
|
|
queue: myQueue,
|
|
run: async (payload) => {
|
|
//...
|
|
},
|
|
});
|
|
```
|
|
|
|
When the parent task checkpoints at the `triggerAndWait` call, it releases its concurrency slot back to the queue, allowing the subtask to execute. Once the subtask completes, the parent task will resume.
|
|
|
|
However, you can only have recursive waits up to your queue's concurrency limit. If you exceed this limit, you will receive a `RECURSIVE_WAIT_DEADLOCK` error:
|
|
|
|
```ts /trigger/deadlock.ts
|
|
export const myQueue = queue({
|
|
name: "my-queue",
|
|
concurrencyLimit: 1,
|
|
});
|
|
|
|
export const parentTask = task({
|
|
id: "parent-task",
|
|
queue: myQueue,
|
|
run: async (payload) => {
|
|
await subtask.triggerAndWait(payload);
|
|
},
|
|
});
|
|
|
|
export const subtask = task({
|
|
id: "subtask",
|
|
queue: myQueue,
|
|
run: async (payload) => {
|
|
await subsubtask.triggerAndWait(payload); // This will cause a deadlock
|
|
},
|
|
});
|
|
|
|
export const subsubtask = task({
|
|
id: "subsubtask",
|
|
queue: myQueue,
|
|
run: async (payload) => {
|
|
//...
|
|
},
|
|
});
|
|
```
|
|
|
|
This results in a `RECURSIVE_WAIT_DEADLOCK` error because the queue can only support one level of recursive waiting with a concurrency limit of 1:
|
|
|
|

|
|
|
|
### Mitigating recursive wait deadlocks
|
|
|
|
To avoid recursive wait deadlocks when using shared queues:
|
|
|
|
1. **Increase the queue's concurrency limit** to allow more levels of recursive waiting
|
|
2. **Use different queues** for parent and child tasks to eliminate the possibility of deadlock
|
|
3. **Design task hierarchies** to minimize deep recursive waiting patterns
|
|
|
|
Remember that the number of recursive waits you can have on a shared queue is limited by that queue's concurrency limit.
|