Concurrency Control in Hatchet Tasks

Hatchet provides powerful concurrency control features to help you manage the execution of your tasks. This is particularly useful when you have tasks that may be triggered frequently or have long-running steps, and you want to limit the number of concurrent executions to prevent overloading your system, ensure fairness, or avoid race conditions.

Why use concurrency control?

There are several reasons why you might want to use concurrency control in your Hatchet tasks:

1. Fairness: When you have multiple clients or users triggering tasks, concurrency control can help ensure fair access to resources. By limiting the number of concurrent runs per client or user, you can prevent a single client from monopolizing the system and ensure that all clients get a fair share of the available resources.

2. Resource management: If your task steps are resource-intensive (e.g., they make external API calls or perform heavy computations), running too many instances concurrently can overload your system. By limiting concurrency, you can ensure your system remains stable and responsive.

3. Avoiding race conditions: If your task steps modify shared resources, running multiple instances concurrently can lead to race conditions and inconsistent data. Concurrency control helps you avoid these issues by ensuring only a limited number of instances run at a time.

4. Compliance with external service limits: If your task steps interact with external services that have rate limits, concurrency control can help you stay within those limits and avoid being throttled or blocked.

5. Spike Protection: When you have tasks that are triggered by external events, such as webhooks or user actions, you may experience spikes in traffic that can overwhelm your system. Concurrency control can help you manage these spikes by limiting the number of concurrent runs and queuing new runs until resources become available.

Available Strategies:

• GROUP_ROUND_ROBIN: Distribute task instances across available slots in a round-robin fashion based on the key function.
• CANCEL_IN_PROGRESS: Cancel the currently running task instances for the same concurrency key to free up slots for the new instance.
• CANCEL_NEWEST: Cancel the newest task instance for the same concurrency key to free up slots for the new instance.

We’re always open to adding more strategies to fit your needs. Join our discord to let us know.

Setting concurrency on workers

In addition to setting concurrency limits at the task level, you can also control concurrency at the worker level by passing the slots option when creating a new Worker instance:

class WorkflowInput(BaseModel):
    group: str


concurrency_limit_rr_workflow = hatchet.workflow(
    name="ConcurrencyDemoWorkflowRR",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    ),
    input_validator=WorkflowInput,
)

export const simpleConcurrency = hatchet.workflow<SimpleInput, SimpleOutput>({
  name: 'simple-concurrency',
  concurrency: {
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    expression: 'input.GroupKey',
  },
});

var maxRuns int32 = 1
strategy := types.GroupRoundRobin

concurrency := factory.NewTask(
	create.StandaloneTask{
		Name: "simple-concurrency",
		Concurrency: []*types.Concurrency{
			{
				Expression:    "input.GroupKey",
				MaxRuns:       &maxRuns,
				LimitStrategy: &strategy,
			},
		},
	}, func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet,
)

This example will only let 1 run in each group run at a given time to fairly distribute the load across the workers.

Group Round Robin

How it works

When a new task instance is triggered, the GROUP_ROUND_ROBIN strategy will:

1. Determine the group that the instance belongs to based on the key function defined in the task’s concurrency configuration.
2. Check if there are any available slots for the instance’s group based on the slots limit of available workers.
3. If a slot is available, the new task instance starts executing immediately.
4. If no slots are available, the new task instance is added to a queue for its group.
5. When a running task instance completes and a slot becomes available for a group, the next queued instance for that group (in round-robin order) is dequeued and starts executing.

This strategy ensures that task instances are processed fairly across different groups, preventing any one group from monopolizing the available resources. It also helps to reduce latency for instances within each group, as they are processed in a round-robin fashion rather than strictly in the order they were triggered.

When to use GROUP_ROUND_ROBIN

The GROUP_ROUND_ROBIN strategy is particularly useful in scenarios where:

• You have multiple clients or users triggering task instances, and you want to ensure fair resource allocation among them.
• You want to process instances within each group in a round-robin fashion to minimize latency and ensure that no single instance within a group is starved for resources.
• You have long-running task instances and want to avoid one group’s instances monopolizing the available slots.

Keep in mind that the GROUP_ROUND_ROBIN strategy may not be suitable for all use cases, especially those that require strict ordering or prioritization of the most recent events.

Cancel In Progress

How it works

When a new task instance is triggered, the CANCEL_IN_PROGRESS strategy will:

1. Determine the group that the instance belongs to based on the key function defined in the task’s concurrency configuration.
2. Check if there are any available slots for the instance’s group based on the maxRuns limit of available workers.
3. If a slot is available, the new task instance starts executing immediately.
4. If there are no available slots, currently running task instances for the same concurrency key are cancelled to free up slots for the new instance.
5. The new task instance starts executing immediately.

When to use Cancel In Progress

The CANCEL_IN_PROGRESS strategy is particularly useful in scenarios where:

• You have long-running task instances that may become stale or irrelevant if newer instances are triggered.
• You want to prioritize processing the most recent data or events, even if it means canceling older task instances.
• You have resource-intensive tasks where it’s more efficient to cancel an in-progress instance and start a new one than to wait for the old instance to complete.
• Your user UI allows for multiple inputs, but only the most recent is relevant (i.e. chat messages, form submissions, etc.).

Cancel Newest

How it works

The CANCEL_NEWEST strategy is similar to CANCEL_IN_PROGRESS, but it cancels the newly enqueued run instead of the oldest.

When to use CANCEL_NEWEST

The CANCEL_NEWEST strategy is particularly useful in scenarios where:

• You want to allow in progress runs to complete before starting new work.
• You have long-running task instances and want to avoid one group’s instances monopolizing the available slots.

Multiple concurrency strategies

You can also combine multiple concurrency strategies to create a more complex concurrency control system. For example, you can use one group key to represent a specific team, and another group to represent a specific resource in that team, giving you more control over the rate at which tasks are executed.

class WorkflowInput(BaseModel):
    name: str
    digit: str


concurrency_workflow_level_workflow = hatchet.workflow(
    name="ConcurrencyWorkflowManyKeys",
    input_validator=WorkflowInput,
    concurrency=[
        ConcurrencyExpression(
            expression="input.digit",
            max_runs=DIGIT_MAX_RUNS,
            limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
        ),
        ConcurrencyExpression(
            expression="input.name",
            max_runs=NAME_MAX_RUNS,
            limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
        ),
    ],
)

export const multipleConcurrencyKeys = hatchet.workflow<SimpleInput, SimpleOutput>({
  name: 'simple-concurrency',
  concurrency: [
    {
      maxRuns: 1,
      limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
      expression: 'input.Tier',
    },
    {
      maxRuns: 1,
      limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
      expression: 'input.Account',
    },
  ],
});

strategy := types.GroupRoundRobin
var maxRuns int32 = 20

concurrency := factory.NewTask(
	create.StandaloneTask{
		Name: "simple-concurrency",
		Concurrency: []*types.Concurrency{
			{
				Expression:    "input.Tier",
				MaxRuns:       &maxRuns,
				LimitStrategy: &strategy,
			},
			{
				Expression:    "input.Account",
				MaxRuns:       &maxRuns,
				LimitStrategy: &strategy,
			},
		},
	}, func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet,
)