mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
## What changed?
Add a ExecutionQueueScheduler to serialize tasks from a busy workflow
execution. Currently, all workflow tasks go through FIFO scheduler. We
have a per-workflow-execution lock that must be aquired in each history
task. But if a single execution has large number of tasks, each of these
tasks compete for this lock and create large number of retries. By
Adding this new scheduler, we can serialize task processing from such
busy workflows. These tasks are routed to a new WorkflowQueueScheduler
when lock contention is detected in a workflow. This will create a new
queue for that workflow. Additional tasks for this workflow will be then
routed to this new queue. This queue will be cleaned up after a few
seconds of inactivity from that workflow execution. We have added a
ExecutionAwareScheduler which will manage this routing of workflow tasks
to either FIFO scheduler or this new WQ Scheduler.
```
Task → InterleavedWeightedRoundRobinScheduler
↓
ExecutionAwareScheduler
↓
┌─────┴─────┐
↓ ↓
FIFOScheduler ExecutionQueueScheduler
(normal path) (contended executions)
```
This new scheduler is only enabled when
history.taskSchedulerEnableExecutionQueueScheduler is enabled. The
number of workflow queues created in this scheduler will be controlled
by config history.taskSchedulerWorkflowQueueSchedulerQueueSize.
Tasks are routed to FIFOScheduler(Like the way it was before this
change) if number of queues reaches this value.
A new set of goroutines is spawned for each queue in this new scheduler.
This is fine here as we don’t expect more than a few hundred hot
workflows per history host. This simplifies the design for this
scheduler.
## Why?
To reduce workflow lock contention and wasted history CPU when tasks are
competing for workflow lock.
## Benchmark Results
Collected by running 2,000 parallel activities from a single workflow
with a 5ms lock timeout to trigger contention. EQS queue concurrency =
2, max queues = 500.
### Task Routing & Failures
| Metric | **EQS ENABLED (C=2)** | **EQS DISABLED** | Improvement |
| --- | --- | --- | --- |
| EQS Tasks Submitted | 1,953 | 0 | - |
| EQS Tasks Completed | 1,940 | 0 | - |
| EQS Tasks Failed | 13 | 0 | - |
| EQS Tasks Aborted | 0 | 0 | - |
| EQS Submit Rejected | 0 | 0 | - |
| FIFO Tasks Completed | 65 | 2,051 | - |
| **Total Failures** | **66** | **6,382** | **97x fewer failures** |
### End-to-End Task Latency (task_latency_queue)
| Percentile | **EQS ENABLED (C=2)** | **EQS DISABLED** | Improvement |
| --- | --- | --- | --- |
| Average | 426ms | 908ms | **2.1x faster** |
| P90 | 668ms | **1,599ms** | **2.4x faster** |
| P99 | 729ms | **3,982ms** | **5.5x faster** |
| Max | 736ms | **6,773ms** | **9.2x faster** |
### Runtime
| Metric | **EQS ENABLED (C=2)** | **EQS DISABLED** | Improvement |
| --- | --- | --- | --- |
| Test Time | 4.6s | 8.9s | **1.9x faster** |
## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [x] added new unit test(s)
- [x] added new functional test(s)
143 lines
4.4 KiB
Go
143 lines
4.4 KiB
Go
// Code generated by MockGen. DO NOT EDIT.
|
|
// Source: scheduler.go
|
|
//
|
|
// Generated by this command:
|
|
//
|
|
// mockgen -package queues -source scheduler.go -destination scheduler_mock.go
|
|
//
|
|
|
|
// Package queues is a generated GoMock package.
|
|
package queues
|
|
|
|
import (
|
|
reflect "reflect"
|
|
|
|
gomock "go.uber.org/mock/gomock"
|
|
)
|
|
|
|
// MockBusyWorkflowHandler is a mock of BusyWorkflowHandler interface.
|
|
type MockBusyWorkflowHandler struct {
|
|
ctrl *gomock.Controller
|
|
recorder *MockBusyWorkflowHandlerMockRecorder
|
|
isgomock struct{}
|
|
}
|
|
|
|
// MockBusyWorkflowHandlerMockRecorder is the mock recorder for MockBusyWorkflowHandler.
|
|
type MockBusyWorkflowHandlerMockRecorder struct {
|
|
mock *MockBusyWorkflowHandler
|
|
}
|
|
|
|
// NewMockBusyWorkflowHandler creates a new mock instance.
|
|
func NewMockBusyWorkflowHandler(ctrl *gomock.Controller) *MockBusyWorkflowHandler {
|
|
mock := &MockBusyWorkflowHandler{ctrl: ctrl}
|
|
mock.recorder = &MockBusyWorkflowHandlerMockRecorder{mock}
|
|
return mock
|
|
}
|
|
|
|
// EXPECT returns an object that allows the caller to indicate expected use.
|
|
func (m *MockBusyWorkflowHandler) EXPECT() *MockBusyWorkflowHandlerMockRecorder {
|
|
return m.recorder
|
|
}
|
|
|
|
// HandleBusyWorkflow mocks base method.
|
|
func (m *MockBusyWorkflowHandler) HandleBusyWorkflow(arg0 Executable) bool {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "HandleBusyWorkflow", arg0)
|
|
ret0, _ := ret[0].(bool)
|
|
return ret0
|
|
}
|
|
|
|
// HandleBusyWorkflow indicates an expected call of HandleBusyWorkflow.
|
|
func (mr *MockBusyWorkflowHandlerMockRecorder) HandleBusyWorkflow(arg0 any) *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HandleBusyWorkflow", reflect.TypeOf((*MockBusyWorkflowHandler)(nil).HandleBusyWorkflow), arg0)
|
|
}
|
|
|
|
// MockScheduler is a mock of Scheduler interface.
|
|
type MockScheduler struct {
|
|
ctrl *gomock.Controller
|
|
recorder *MockSchedulerMockRecorder
|
|
isgomock struct{}
|
|
}
|
|
|
|
// MockSchedulerMockRecorder is the mock recorder for MockScheduler.
|
|
type MockSchedulerMockRecorder struct {
|
|
mock *MockScheduler
|
|
}
|
|
|
|
// NewMockScheduler creates a new mock instance.
|
|
func NewMockScheduler(ctrl *gomock.Controller) *MockScheduler {
|
|
mock := &MockScheduler{ctrl: ctrl}
|
|
mock.recorder = &MockSchedulerMockRecorder{mock}
|
|
return mock
|
|
}
|
|
|
|
// EXPECT returns an object that allows the caller to indicate expected use.
|
|
func (m *MockScheduler) EXPECT() *MockSchedulerMockRecorder {
|
|
return m.recorder
|
|
}
|
|
|
|
// Start mocks base method.
|
|
func (m *MockScheduler) Start() {
|
|
m.ctrl.T.Helper()
|
|
m.ctrl.Call(m, "Start")
|
|
}
|
|
|
|
// Start indicates an expected call of Start.
|
|
func (mr *MockSchedulerMockRecorder) Start() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockScheduler)(nil).Start))
|
|
}
|
|
|
|
// Stop mocks base method.
|
|
func (m *MockScheduler) Stop() {
|
|
m.ctrl.T.Helper()
|
|
m.ctrl.Call(m, "Stop")
|
|
}
|
|
|
|
// Stop indicates an expected call of Stop.
|
|
func (mr *MockSchedulerMockRecorder) Stop() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockScheduler)(nil).Stop))
|
|
}
|
|
|
|
// Submit mocks base method.
|
|
func (m *MockScheduler) Submit(arg0 Executable) {
|
|
m.ctrl.T.Helper()
|
|
m.ctrl.Call(m, "Submit", arg0)
|
|
}
|
|
|
|
// Submit indicates an expected call of Submit.
|
|
func (mr *MockSchedulerMockRecorder) Submit(arg0 any) *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Submit", reflect.TypeOf((*MockScheduler)(nil).Submit), arg0)
|
|
}
|
|
|
|
// TaskChannelKeyFn mocks base method.
|
|
func (m *MockScheduler) TaskChannelKeyFn() TaskChannelKeyFn {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "TaskChannelKeyFn")
|
|
ret0, _ := ret[0].(TaskChannelKeyFn)
|
|
return ret0
|
|
}
|
|
|
|
// TaskChannelKeyFn indicates an expected call of TaskChannelKeyFn.
|
|
func (mr *MockSchedulerMockRecorder) TaskChannelKeyFn() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TaskChannelKeyFn", reflect.TypeOf((*MockScheduler)(nil).TaskChannelKeyFn))
|
|
}
|
|
|
|
// TrySubmit mocks base method.
|
|
func (m *MockScheduler) TrySubmit(arg0 Executable) bool {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "TrySubmit", arg0)
|
|
ret0, _ := ret[0].(bool)
|
|
return ret0
|
|
}
|
|
|
|
// TrySubmit indicates an expected call of TrySubmit.
|
|
func (mr *MockSchedulerMockRecorder) TrySubmit(arg0 any) *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TrySubmit", reflect.TypeOf((*MockScheduler)(nil).TrySubmit), arg0)
|
|
}
|