mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
zeekay
5a601d081b
Full module path rebrand from Temporal to Hanzo Tasks: - go.mod module: github.com/hanzoai/tasks - All internal Go imports updated (1906 files) - cmd/server renamed to cmd/tasksd - Binary name: tasksd (was temporal-server) - Docker images: hanzoai/tasks (was temporalio/server) - README rewritten with Hanzo Tasks branding - LLM.md created with integration notes - Makefile, goreleaser, CI workflows updated - go.temporal.io/sdk and go.temporal.io/api imports preserved (separate repos) Fork of Temporal.io (MIT license).
68 lines
2.3 KiB
Go
68 lines
2.3 KiB
Go
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination stream_receiver_flow_controller_mock.go
|
|
package replication
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
enumsspb "github.com/hanzoai/tasks/api/enums/v1"
|
|
"github.com/hanzoai/tasks/service/history/configs"
|
|
)
|
|
|
|
type (
|
|
FlowControlSignalProvider func() *FlowControlSignal
|
|
|
|
// FlowControlSignal holds signals to make flow control decision, more signalsProvider can be added here i.e. total persistence rps, cpu usage etc.
|
|
FlowControlSignal struct {
|
|
taskTrackingCount int
|
|
lastSlowSubmission time.Time
|
|
}
|
|
|
|
// FlowControlInfo holds the flow control command and an optional cause string for logging when command is PAUSE.
|
|
FlowControlInfo struct {
|
|
Command enumsspb.ReplicationFlowControlCommand
|
|
Cause string
|
|
}
|
|
|
|
ReceiverFlowController interface {
|
|
GetFlowControlInfo(priority enumsspb.TaskPriority) FlowControlInfo
|
|
}
|
|
streamReceiverFlowControllerImpl struct {
|
|
signalsProvider map[enumsspb.TaskPriority]FlowControlSignalProvider
|
|
config *configs.Config
|
|
}
|
|
)
|
|
|
|
func NewReceiverFlowControl(signals map[enumsspb.TaskPriority]FlowControlSignalProvider, config *configs.Config) *streamReceiverFlowControllerImpl {
|
|
return &streamReceiverFlowControllerImpl{
|
|
signalsProvider: signals,
|
|
config: config,
|
|
}
|
|
}
|
|
|
|
func (s *streamReceiverFlowControllerImpl) GetFlowControlInfo(priority enumsspb.TaskPriority) FlowControlInfo {
|
|
if signal, ok := s.signalsProvider[priority]; ok {
|
|
signalData := signal()
|
|
limit := s.config.ReplicationReceiverMaxOutstandingTaskCount()
|
|
|
|
if signalData.taskTrackingCount > limit {
|
|
return FlowControlInfo{
|
|
Command: enumsspb.REPLICATION_FLOW_CONTROL_COMMAND_PAUSE,
|
|
Cause: fmt.Sprintf("task count %d > limit %d", signalData.taskTrackingCount, limit),
|
|
}
|
|
}
|
|
|
|
if s.config.EnableReplicationReceiverSlowSubmissionFlowControl() {
|
|
now := time.Now()
|
|
slowSubmissionWindow := now.Add(-s.config.ReplicationReceiverSlowSubmissionWindow())
|
|
if signalData.lastSlowSubmission.After(slowSubmissionWindow) {
|
|
return FlowControlInfo{
|
|
Command: enumsspb.REPLICATION_FLOW_CONTROL_COMMAND_PAUSE,
|
|
Cause: fmt.Sprintf("slow submission within window (last at %v)", signalData.lastSlowSubmission),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return FlowControlInfo{Command: enumsspb.REPLICATION_FLOW_CONTROL_COMMAND_RESUME}
|
|
}
|