Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Forward polls from sticky partitions to high-priority backlog (#8925)

Browse Source 
## What changed?
- Add "ephemeral data" propagated alongside user data.
- Include information about which priority levels have significant
backlog in ephemeral data (currently only propagated down the tree, not
up)
- When sticky partitions see that normal partitions have significant
backlog, set up poll forwarders to forward to those partitions.
- The poll forwarders use min priority to ensure they only get
higher-priority tasks than available local tasks.

## Why?
Without this functionality, sticky queues interfere with priority
dispatch by keeping most of the pollers working on sticky tasks and
potentially starving high-priority tasks that appear on the normal
queue. With the default values of the new settings, sticky pollers
should notice high-priority normal tasks within 10 seconds.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
- Clients have to accept that polls on sticky may return normal tasks. I
believe all SDKs are fine with this.
- Increased overhead of user data + ephemeral data.
This commit is contained in:
David Reiss
2026-01-15 15:01:05 -08:00
committed by GitHub
parent e61179270f
commit 08f90d3960
27 changed files with 1183 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
api/matchingservice/v1/request_response.pb.go
View File

@@ -2282,8 +2282,10 @@ type GetTaskQueueUserDataRequest struct {
// If the requester has no data, it should set this to 0.
// This value must not be set to a negative number (note that our linter suggests avoiding uint64).
LastKnownUserDataVersion int64 `protobuf:"varint,3,opt,name=last_known_user_data_version,json=lastKnownUserDataVersion,proto3" json:"last_known_user_data_version,omitempty"`
// If set and last_known_user_data_version is the current version, block until new data is
// available (or timeout).
// Same for ephemeral data.
LastKnownEphemeralDataVersion int64 `protobuf:"varint,7,opt,name=last_known_ephemeral_data_version,json=lastKnownEphemeralDataVersion,proto3" json:"last_known_ephemeral_data_version,omitempty"`
// If set and last_known_{user_data,ephemeral_data}_version is the current version,
// block until new data is available (or timeout).
WaitNewData bool `protobuf:"varint,4,opt,name=wait_new_data,json=waitNewData,proto3" json:"wait_new_data,omitempty"`
// If set, do not load task queue if unloaded. (Returns FailedPrecondition error in that case.)
OnlyIfLoaded bool `protobuf:"varint,6,opt,name=only_if_loaded,json=onlyIfLoaded,proto3" json:"only_if_loaded,omitempty"`
@@ -2349,6 +2351,13 @@ func (x *GetTaskQueueUserDataRequest) GetLastKnownUserDataVersion() int64 {
return 0
}
func (x *GetTaskQueueUserDataRequest) GetLastKnownEphemeralDataVersion() int64 {
if x != nil {
return x.LastKnownEphemeralDataVersion
}
return 0
}
func (x *GetTaskQueueUserDataRequest) GetWaitNewData() bool {
if x != nil {
return x.WaitNewData
@@ -2368,6 +2377,7 @@ type GetTaskQueueUserDataResponse struct {
// Versioned user data, set if the task queue has user data and the request's last_known_user_data_version is less
// than the version cached in the root partition.
UserData *v111.VersionedTaskQueueUserData `protobuf:"bytes,2,opt,name=user_data,json=userData,proto3" json:"user_data,omitempty"`
EphemeralData *v18.VersionedEphemeralData `protobuf:"bytes,3,opt,name=ephemeral_data,json=ephemeralData,proto3" json:"ephemeral_data,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -2409,6 +2419,13 @@ func (x *GetTaskQueueUserDataResponse) GetUserData() *v111.VersionedTaskQueueUse
return nil
}
func (x *GetTaskQueueUserDataResponse) GetEphemeralData() *v18.VersionedEphemeralData {
if x != nil {
return x.EphemeralData
}
return nil
}
type SyncDeploymentUserDataRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
NamespaceId string `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
@@ -5527,17 +5544,19 @@ const file_temporal_server_api_matchingservice_v1_request_response_proto_rawDesc
"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12_\n" +
"\arequest\x18\x02 \x01(\v2E.temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityRequestR\arequest\"\x8b\x01\n" +
"%GetWorkerBuildIdCompatibilityResponse\x12b\n" +
"\bresponse\x18\x01 \x01(\v2F.temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityResponseR\bresponse\"\xb7\x02\n" +
"\bresponse\x18\x01 \x01(\v2F.temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityResponseR\bresponse\"\x81\x03\n" +
"\x1bGetTaskQueueUserDataRequest\x12!\n" +
"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1d\n" +
"\n" +
"task_queue\x18\x02 \x01(\tR\ttaskQueue\x12L\n" +
"\x0ftask_queue_type\x18\x05 \x01(\x0e2$.temporal.api.enums.v1.TaskQueueTypeR\rtaskQueueType\x12>\n" +
"\x1clast_known_user_data_version\x18\x03 \x01(\x03R\x18lastKnownUserDataVersion\x12\"\n" +
"\x1clast_known_user_data_version\x18\x03 \x01(\x03R\x18lastKnownUserDataVersion\x12H\n" +
"!last_known_ephemeral_data_version\x18\a \x01(\x03R\x1dlastKnownEphemeralDataVersion\x12\"\n" +
"\rwait_new_data\x18\x04 \x01(\bR\vwaitNewData\x12$\n" +
"\x0eonly_if_loaded\x18\x06 \x01(\bR\fonlyIfLoaded\"\x81\x01\n" +
"\x0eonly_if_loaded\x18\x06 \x01(\bR\fonlyIfLoaded\"\xe2\x01\n" +
"\x1cGetTaskQueueUserDataResponse\x12[\n" +
"\tuser_data\x18\x02 \x01(\v2>.temporal.server.api.persistence.v1.VersionedTaskQueueUserDataR\buserDataJ\x04\b\x01\x10\x02\"\xf4\x06\n" +
"\tuser_data\x18\x02 \x01(\v2>.temporal.server.api.persistence.v1.VersionedTaskQueueUserDataR\buserData\x12_\n" +
"\x0eephemeral_data\x18\x03 \x01(\v28.temporal.server.api.taskqueue.v1.VersionedEphemeralDataR\rephemeralDataJ\x04\b\x01\x10\x02\"\xf4\x06\n" +
"\x1dSyncDeploymentUserDataRequest\x12!\n" +
"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1d\n" +
"\n" +
@@ -5845,29 +5864,30 @@ var file_temporal_server_api_matchingservice_v1_request_response_proto_goTypes =
(*v1.GetWorkerBuildIdCompatibilityRequest)(nil), // 122: temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityRequest
(*v1.GetWorkerBuildIdCompatibilityResponse)(nil), // 123: temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityResponse
(*v111.VersionedTaskQueueUserData)(nil), // 124: temporal.server.api.persistence.v1.VersionedTaskQueueUserData
(*v110.DeploymentVersionData)(nil), // 125: temporal.server.api.deployment.v1.DeploymentVersionData
(*v112.RoutingConfig)(nil), // 126: temporal.api.deployment.v1.RoutingConfig
(*v111.TaskQueueUserData)(nil), // 127: temporal.server.api.persistence.v1.TaskQueueUserData
(*v113.Request)(nil), // 128: temporal.api.nexus.v1.Request
(*v113.HandlerError)(nil), // 129: temporal.api.nexus.v1.HandlerError
(*v113.Response)(nil), // 130: temporal.api.nexus.v1.Response
(*v1.PollNexusTaskQueueRequest)(nil), // 131: temporal.api.workflowservice.v1.PollNexusTaskQueueRequest
(*v1.PollNexusTaskQueueResponse)(nil), // 132: temporal.api.workflowservice.v1.PollNexusTaskQueueResponse
(*v1.RespondNexusTaskCompletedRequest)(nil), // 133: temporal.api.workflowservice.v1.RespondNexusTaskCompletedRequest
(*v1.RespondNexusTaskFailedRequest)(nil), // 134: temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest
(*v111.NexusEndpointSpec)(nil), // 135: temporal.server.api.persistence.v1.NexusEndpointSpec
(*v111.NexusEndpointEntry)(nil), // 136: temporal.server.api.persistence.v1.NexusEndpointEntry
(*v1.RecordWorkerHeartbeatRequest)(nil), // 137: temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest
(*v1.ListWorkersRequest)(nil), // 138: temporal.api.workflowservice.v1.ListWorkersRequest
(*v114.WorkerInfo)(nil), // 139: temporal.api.worker.v1.WorkerInfo
(*v1.UpdateTaskQueueConfigRequest)(nil), // 140: temporal.api.workflowservice.v1.UpdateTaskQueueConfigRequest
(*v14.TaskQueueConfig)(nil), // 141: temporal.api.taskqueue.v1.TaskQueueConfig
(*v1.DescribeWorkerRequest)(nil), // 142: temporal.api.workflowservice.v1.DescribeWorkerRequest
(v115.FairnessState)(0), // 143: temporal.server.api.enums.v1.FairnessState
(*v14.TaskQueueStats)(nil), // 144: temporal.api.taskqueue.v1.TaskQueueStats
(*v18.TaskQueueVersionInfoInternal)(nil), // 145: temporal.server.api.taskqueue.v1.TaskQueueVersionInfoInternal
(*v1.UpdateWorkerBuildIdCompatibilityRequest)(nil), // 146: temporal.api.workflowservice.v1.UpdateWorkerBuildIdCompatibilityRequest
(*v110.WorkerDeploymentVersionData)(nil), // 147: temporal.server.api.deployment.v1.WorkerDeploymentVersionData
(*v18.VersionedEphemeralData)(nil), // 125: temporal.server.api.taskqueue.v1.VersionedEphemeralData
(*v110.DeploymentVersionData)(nil), // 126: temporal.server.api.deployment.v1.DeploymentVersionData
(*v112.RoutingConfig)(nil), // 127: temporal.api.deployment.v1.RoutingConfig
(*v111.TaskQueueUserData)(nil), // 128: temporal.server.api.persistence.v1.TaskQueueUserData
(*v113.Request)(nil), // 129: temporal.api.nexus.v1.Request
(*v113.HandlerError)(nil), // 130: temporal.api.nexus.v1.HandlerError
(*v113.Response)(nil), // 131: temporal.api.nexus.v1.Response
(*v1.PollNexusTaskQueueRequest)(nil), // 132: temporal.api.workflowservice.v1.PollNexusTaskQueueRequest
(*v1.PollNexusTaskQueueResponse)(nil), // 133: temporal.api.workflowservice.v1.PollNexusTaskQueueResponse
(*v1.RespondNexusTaskCompletedRequest)(nil), // 134: temporal.api.workflowservice.v1.RespondNexusTaskCompletedRequest
(*v1.RespondNexusTaskFailedRequest)(nil), // 135: temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest
(*v111.NexusEndpointSpec)(nil), // 136: temporal.server.api.persistence.v1.NexusEndpointSpec
(*v111.NexusEndpointEntry)(nil), // 137: temporal.server.api.persistence.v1.NexusEndpointEntry
(*v1.RecordWorkerHeartbeatRequest)(nil), // 138: temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest
(*v1.ListWorkersRequest)(nil), // 139: temporal.api.workflowservice.v1.ListWorkersRequest
(*v114.WorkerInfo)(nil), // 140: temporal.api.worker.v1.WorkerInfo
(*v1.UpdateTaskQueueConfigRequest)(nil), // 141: temporal.api.workflowservice.v1.UpdateTaskQueueConfigRequest
(*v14.TaskQueueConfig)(nil), // 142: temporal.api.taskqueue.v1.TaskQueueConfig
(*v1.DescribeWorkerRequest)(nil), // 143: temporal.api.workflowservice.v1.DescribeWorkerRequest
(v115.FairnessState)(0), // 144: temporal.server.api.enums.v1.FairnessState
(*v14.TaskQueueStats)(nil), // 145: temporal.api.taskqueue.v1.TaskQueueStats
(*v18.TaskQueueVersionInfoInternal)(nil), // 146: temporal.server.api.taskqueue.v1.TaskQueueVersionInfoInternal
(*v1.UpdateWorkerBuildIdCompatibilityRequest)(nil), // 147: temporal.api.workflowservice.v1.UpdateWorkerBuildIdCompatibilityRequest
(*v110.WorkerDeploymentVersionData)(nil), // 148: temporal.server.api.deployment.v1.WorkerDeploymentVersionData
}
var file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs = []int32{
88, // 0: temporal.server.api.matchingservice.v1.PollWorkflowTaskQueueRequest.poll_request:type_name -> temporal.api.workflowservice.v1.PollWorkflowTaskQueueRequest
@@ -5949,60 +5969,61 @@ var file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs =
123, // 76: temporal.server.api.matchingservice.v1.GetWorkerBuildIdCompatibilityResponse.response:type_name -> temporal.api.workflowservice.v1.GetWorkerBuildIdCompatibilityResponse
111, // 77: temporal.server.api.matchingservice.v1.GetTaskQueueUserDataRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
124, // 78: temporal.server.api.matchingservice.v1.GetTaskQueueUserDataResponse.user_data:type_name -> temporal.server.api.persistence.v1.VersionedTaskQueueUserData
111, // 79: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.task_queue_types:type_name -> temporal.api.enums.v1.TaskQueueType
125, // 80: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.update_version_data:type_name -> temporal.server.api.deployment.v1.DeploymentVersionData
113, // 81: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.forget_version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
126, // 82: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.update_routing_config:type_name -> temporal.api.deployment.v1.RoutingConfig
86, // 83: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.upsert_versions_data:type_name -> temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.UpsertVersionsDataEntry
127, // 84: temporal.server.api.matchingservice.v1.ApplyTaskQueueUserDataReplicationEventRequest.user_data:type_name -> temporal.server.api.persistence.v1.TaskQueueUserData
115, // 85: temporal.server.api.matchingservice.v1.ForceLoadTaskQueuePartitionRequest.task_queue_partition:type_name -> temporal.server.api.taskqueue.v1.TaskQueuePartition
111, // 86: temporal.server.api.matchingservice.v1.ForceUnloadTaskQueueRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
115, // 87: temporal.server.api.matchingservice.v1.ForceUnloadTaskQueuePartitionRequest.task_queue_partition:type_name -> temporal.server.api.taskqueue.v1.TaskQueuePartition
124, // 88: temporal.server.api.matchingservice.v1.UpdateTaskQueueUserDataRequest.user_data:type_name -> temporal.server.api.persistence.v1.VersionedTaskQueueUserData
127, // 89: temporal.server.api.matchingservice.v1.ReplicateTaskQueueUserDataRequest.user_data:type_name -> temporal.server.api.persistence.v1.TaskQueueUserData
93, // 90: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
128, // 91: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.request:type_name -> temporal.api.nexus.v1.Request
107, // 92: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.forward_info:type_name -> temporal.server.api.taskqueue.v1.TaskForwardInfo
129, // 93: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.handler_error:type_name -> temporal.api.nexus.v1.HandlerError
130, // 94: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.response:type_name -> temporal.api.nexus.v1.Response
87, // 95: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.request_timeout:type_name -> temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.Timeout
131, // 96: temporal.server.api.matchingservice.v1.PollNexusTaskQueueRequest.request:type_name -> temporal.api.workflowservice.v1.PollNexusTaskQueueRequest
78, // 97: temporal.server.api.matchingservice.v1.PollNexusTaskQueueRequest.conditions:type_name -> temporal.server.api.matchingservice.v1.PollConditions
132, // 98: temporal.server.api.matchingservice.v1.PollNexusTaskQueueResponse.response:type_name -> temporal.api.workflowservice.v1.PollNexusTaskQueueResponse
93, // 99: temporal.server.api.matchingservice.v1.RespondNexusTaskCompletedRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
133, // 100: temporal.server.api.matchingservice.v1.RespondNexusTaskCompletedRequest.request:type_name -> temporal.api.workflowservice.v1.RespondNexusTaskCompletedRequest
93, // 101: temporal.server.api.matchingservice.v1.RespondNexusTaskFailedRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
134, // 102: temporal.server.api.matchingservice.v1.RespondNexusTaskFailedRequest.request:type_name -> temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest
135, // 103: temporal.server.api.matchingservice.v1.CreateNexusEndpointRequest.spec:type_name -> temporal.server.api.persistence.v1.NexusEndpointSpec
136, // 104: temporal.server.api.matchingservice.v1.CreateNexusEndpointResponse.entry:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
135, // 105: temporal.server.api.matchingservice.v1.UpdateNexusEndpointRequest.spec:type_name -> temporal.server.api.persistence.v1.NexusEndpointSpec
136, // 106: temporal.server.api.matchingservice.v1.UpdateNexusEndpointResponse.entry:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
136, // 107: temporal.server.api.matchingservice.v1.ListNexusEndpointsResponse.entries:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
137, // 108: temporal.server.api.matchingservice.v1.RecordWorkerHeartbeatRequest.heartbeart_request:type_name -> temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest
138, // 109: temporal.server.api.matchingservice.v1.ListWorkersRequest.list_request:type_name -> temporal.api.workflowservice.v1.ListWorkersRequest
139, // 110: temporal.server.api.matchingservice.v1.ListWorkersResponse.workers_info:type_name -> temporal.api.worker.v1.WorkerInfo
140, // 111: temporal.server.api.matchingservice.v1.UpdateTaskQueueConfigRequest.update_taskqueue_config:type_name -> temporal.api.workflowservice.v1.UpdateTaskQueueConfigRequest
141, // 112: temporal.server.api.matchingservice.v1.UpdateTaskQueueConfigResponse.updated_taskqueue_config:type_name -> temporal.api.taskqueue.v1.TaskQueueConfig
142, // 113: temporal.server.api.matchingservice.v1.DescribeWorkerRequest.request:type_name -> temporal.api.workflowservice.v1.DescribeWorkerRequest
139, // 114: temporal.server.api.matchingservice.v1.DescribeWorkerResponse.worker_info:type_name -> temporal.api.worker.v1.WorkerInfo
111, // 115: temporal.server.api.matchingservice.v1.UpdateFairnessStateRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
143, // 116: temporal.server.api.matchingservice.v1.UpdateFairnessStateRequest.fairness_state:type_name -> temporal.server.api.enums.v1.FairnessState
111, // 117: temporal.server.api.matchingservice.v1.CheckTaskQueueVersionMembershipRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
113, // 118: temporal.server.api.matchingservice.v1.CheckTaskQueueVersionMembershipRequest.version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
91, // 119: temporal.server.api.matchingservice.v1.PollWorkflowTaskQueueResponse.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
111, // 120: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesRequest.VersionTaskQueue.type:type_name -> temporal.api.enums.v1.TaskQueueType
111, // 121: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.type:type_name -> temporal.api.enums.v1.TaskQueueType
144, // 122: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.stats:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
82, // 123: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.stats_by_priority_key:type_name -> temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.StatsByPriorityKeyEntry
144, // 124: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.StatsByPriorityKeyEntry.value:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
145, // 125: temporal.server.api.matchingservice.v1.DescribeTaskQueuePartitionResponse.VersionsInfoInternalEntry.value:type_name -> temporal.server.api.taskqueue.v1.TaskQueueVersionInfoInternal
146, // 126: temporal.server.api.matchingservice.v1.UpdateWorkerBuildIdCompatibilityRequest.ApplyPublicRequest.request:type_name -> temporal.api.workflowservice.v1.UpdateWorkerBuildIdCompatibilityRequest
147, // 127: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.UpsertVersionsDataEntry.value:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersionData
128, // [128:128] is the sub-list for method output_type
128, // [128:128] is the sub-list for method input_type
128, // [128:128] is the sub-list for extension type_name
128, // [128:128] is the sub-list for extension extendee
0, // [0:128] is the sub-list for field type_name
125, // 79: temporal.server.api.matchingservice.v1.GetTaskQueueUserDataResponse.ephemeral_data:type_name -> temporal.server.api.taskqueue.v1.VersionedEphemeralData
111, // 80: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.task_queue_types:type_name -> temporal.api.enums.v1.TaskQueueType
126, // 81: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.update_version_data:type_name -> temporal.server.api.deployment.v1.DeploymentVersionData
113, // 82: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.forget_version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
127, // 83: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.update_routing_config:type_name -> temporal.api.deployment.v1.RoutingConfig
86, // 84: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.upsert_versions_data:type_name -> temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.UpsertVersionsDataEntry
128, // 85: temporal.server.api.matchingservice.v1.ApplyTaskQueueUserDataReplicationEventRequest.user_data:type_name -> temporal.server.api.persistence.v1.TaskQueueUserData
115, // 86: temporal.server.api.matchingservice.v1.ForceLoadTaskQueuePartitionRequest.task_queue_partition:type_name -> temporal.server.api.taskqueue.v1.TaskQueuePartition
111, // 87: temporal.server.api.matchingservice.v1.ForceUnloadTaskQueueRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
115, // 88: temporal.server.api.matchingservice.v1.ForceUnloadTaskQueuePartitionRequest.task_queue_partition:type_name -> temporal.server.api.taskqueue.v1.TaskQueuePartition
124, // 89: temporal.server.api.matchingservice.v1.UpdateTaskQueueUserDataRequest.user_data:type_name -> temporal.server.api.persistence.v1.VersionedTaskQueueUserData
128, // 90: temporal.server.api.matchingservice.v1.ReplicateTaskQueueUserDataRequest.user_data:type_name -> temporal.server.api.persistence.v1.TaskQueueUserData
93, // 91: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
129, // 92: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.request:type_name -> temporal.api.nexus.v1.Request
107, // 93: temporal.server.api.matchingservice.v1.DispatchNexusTaskRequest.forward_info:type_name -> temporal.server.api.taskqueue.v1.TaskForwardInfo
130, // 94: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.handler_error:type_name -> temporal.api.nexus.v1.HandlerError
131, // 95: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.response:type_name -> temporal.api.nexus.v1.Response
87, // 96: temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.request_timeout:type_name -> temporal.server.api.matchingservice.v1.DispatchNexusTaskResponse.Timeout
132, // 97: temporal.server.api.matchingservice.v1.PollNexusTaskQueueRequest.request:type_name -> temporal.api.workflowservice.v1.PollNexusTaskQueueRequest
78, // 98: temporal.server.api.matchingservice.v1.PollNexusTaskQueueRequest.conditions:type_name -> temporal.server.api.matchingservice.v1.PollConditions
133, // 99: temporal.server.api.matchingservice.v1.PollNexusTaskQueueResponse.response:type_name -> temporal.api.workflowservice.v1.PollNexusTaskQueueResponse
93, // 100: temporal.server.api.matchingservice.v1.RespondNexusTaskCompletedRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
134, // 101: temporal.server.api.matchingservice.v1.RespondNexusTaskCompletedRequest.request:type_name -> temporal.api.workflowservice.v1.RespondNexusTaskCompletedRequest
93, // 102: temporal.server.api.matchingservice.v1.RespondNexusTaskFailedRequest.task_queue:type_name -> temporal.api.taskqueue.v1.TaskQueue
135, // 103: temporal.server.api.matchingservice.v1.RespondNexusTaskFailedRequest.request:type_name -> temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest
136, // 104: temporal.server.api.matchingservice.v1.CreateNexusEndpointRequest.spec:type_name -> temporal.server.api.persistence.v1.NexusEndpointSpec
137, // 105: temporal.server.api.matchingservice.v1.CreateNexusEndpointResponse.entry:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
136, // 106: temporal.server.api.matchingservice.v1.UpdateNexusEndpointRequest.spec:type_name -> temporal.server.api.persistence.v1.NexusEndpointSpec
137, // 107: temporal.server.api.matchingservice.v1.UpdateNexusEndpointResponse.entry:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
137, // 108: temporal.server.api.matchingservice.v1.ListNexusEndpointsResponse.entries:type_name -> temporal.server.api.persistence.v1.NexusEndpointEntry
138, // 109: temporal.server.api.matchingservice.v1.RecordWorkerHeartbeatRequest.heartbeart_request:type_name -> temporal.api.workflowservice.v1.RecordWorkerHeartbeatRequest
139, // 110: temporal.server.api.matchingservice.v1.ListWorkersRequest.list_request:type_name -> temporal.api.workflowservice.v1.ListWorkersRequest
140, // 111: temporal.server.api.matchingservice.v1.ListWorkersResponse.workers_info:type_name -> temporal.api.worker.v1.WorkerInfo
141, // 112: temporal.server.api.matchingservice.v1.UpdateTaskQueueConfigRequest.update_taskqueue_config:type_name -> temporal.api.workflowservice.v1.UpdateTaskQueueConfigRequest
142, // 113: temporal.server.api.matchingservice.v1.UpdateTaskQueueConfigResponse.updated_taskqueue_config:type_name -> temporal.api.taskqueue.v1.TaskQueueConfig
143, // 114: temporal.server.api.matchingservice.v1.DescribeWorkerRequest.request:type_name -> temporal.api.workflowservice.v1.DescribeWorkerRequest
140, // 115: temporal.server.api.matchingservice.v1.DescribeWorkerResponse.worker_info:type_name -> temporal.api.worker.v1.WorkerInfo
111, // 116: temporal.server.api.matchingservice.v1.UpdateFairnessStateRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
144, // 117: temporal.server.api.matchingservice.v1.UpdateFairnessStateRequest.fairness_state:type_name -> temporal.server.api.enums.v1.FairnessState
111, // 118: temporal.server.api.matchingservice.v1.CheckTaskQueueVersionMembershipRequest.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
113, // 119: temporal.server.api.matchingservice.v1.CheckTaskQueueVersionMembershipRequest.version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
91, // 120: temporal.server.api.matchingservice.v1.PollWorkflowTaskQueueResponse.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
111, // 121: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesRequest.VersionTaskQueue.type:type_name -> temporal.api.enums.v1.TaskQueueType
111, // 122: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.type:type_name -> temporal.api.enums.v1.TaskQueueType
145, // 123: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.stats:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
82, // 124: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.stats_by_priority_key:type_name -> temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.StatsByPriorityKeyEntry
145, // 125: temporal.server.api.matchingservice.v1.DescribeVersionedTaskQueuesResponse.VersionTaskQueue.StatsByPriorityKeyEntry.value:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
146, // 126: temporal.server.api.matchingservice.v1.DescribeTaskQueuePartitionResponse.VersionsInfoInternalEntry.value:type_name -> temporal.server.api.taskqueue.v1.TaskQueueVersionInfoInternal
147, // 127: temporal.server.api.matchingservice.v1.UpdateWorkerBuildIdCompatibilityRequest.ApplyPublicRequest.request:type_name -> temporal.api.workflowservice.v1.UpdateWorkerBuildIdCompatibilityRequest
148, // 128: temporal.server.api.matchingservice.v1.SyncDeploymentUserDataRequest.UpsertVersionsDataEntry.value:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersionData
129, // [129:129] is the sub-list for method output_type
129, // [129:129] is the sub-list for method input_type
129, // [129:129] is the sub-list for extension type_name
129, // [129:129] is the sub-list for extension extendee
0, // [0:129] is the sub-list for field type_name
}
func init() { file_temporal_server_api_matchingservice_v1_request_response_proto_init() }

74
api/taskqueue/v1/message.go-helpers.pb.go
View File

@@ -300,3 +300,77 @@ func (this *TaskForwardInfo) Equal(that interface{}) bool {
return proto.Equal(this, that1)
}
// Marshal an object of type EphemeralData to the protobuf v3 wire format
func (val *EphemeralData) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type EphemeralData from the protobuf v3 wire format
func (val *EphemeralData) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *EphemeralData) Size() int {
return proto.Size(val)
}
// Equal returns whether two EphemeralData values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *EphemeralData) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *EphemeralData
switch t := that.(type) {
case *EphemeralData:
that1 = t
case EphemeralData:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type VersionedEphemeralData to the protobuf v3 wire format
func (val *VersionedEphemeralData) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type VersionedEphemeralData from the protobuf v3 wire format
func (val *VersionedEphemeralData) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *VersionedEphemeralData) Size() int {
return proto.Size(val)
}
// Equal returns whether two VersionedEphemeralData values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *VersionedEphemeralData) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *VersionedEphemeralData
switch t := that.(type) {
case *VersionedEphemeralData:
that1 = t
case VersionedEphemeralData:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}

285
api/taskqueue/v1/message.pb.go
View File

@@ -684,6 +684,214 @@ func (x *TaskForwardInfo) GetDispatchVersionSet() string {
return ""
}
// EphemeralData is data that we want to propagate among task queue partitions, but is not persisted.
// Ephemeral data is propagated alongside "task queue user data", but while user data applies to a
// task queue family (all queues with the same name, across types), ephemeral data applies only to
// one type at a time.
type EphemeralData struct {
state protoimpl.MessageState `protogen:"open.v1"`
Partition []*EphemeralData_ByPartition `protobuf:"bytes,1,rep,name=partition,proto3" json:"partition,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *EphemeralData) Reset() {
*x = EphemeralData{}
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *EphemeralData) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*EphemeralData) ProtoMessage() {}
func (x *EphemeralData) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use EphemeralData.ProtoReflect.Descriptor instead.
func (*EphemeralData) Descriptor() ([]byte, []int) {
return file_temporal_server_api_taskqueue_v1_message_proto_rawDescGZIP(), []int{8}
}
func (x *EphemeralData) GetPartition() []*EphemeralData_ByPartition {
if x != nil {
return x.Partition
}
return nil
}
type VersionedEphemeralData struct {
state protoimpl.MessageState `protogen:"open.v1"`
Data *EphemeralData `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
Version int64 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *VersionedEphemeralData) Reset() {
*x = VersionedEphemeralData{}
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *VersionedEphemeralData) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*VersionedEphemeralData) ProtoMessage() {}
func (x *VersionedEphemeralData) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use VersionedEphemeralData.ProtoReflect.Descriptor instead.
func (*VersionedEphemeralData) Descriptor() ([]byte, []int) {
return file_temporal_server_api_taskqueue_v1_message_proto_rawDescGZIP(), []int{9}
}
func (x *VersionedEphemeralData) GetData() *EphemeralData {
if x != nil {
return x.Data
}
return nil
}
func (x *VersionedEphemeralData) GetVersion() int64 {
if x != nil {
return x.Version
}
return 0
}
type EphemeralData_ByVersion struct {
state protoimpl.MessageState `protogen:"open.v1"`
// Key for this data. Data for the unversioned queue has no version field present.
// All following fields are data associated with this versioned queue.
Version *v12.WorkerDeploymentVersion `protobuf:"bytes,1,opt,name=version,proto3" json:"version,omitempty"`
// This is a bit field of priority levels that have "significant" backlog (defined by
// the server configuration). Priority key k corresponds to 1<<k.
BacklogPriorityLevels int64 `protobuf:"varint,2,opt,name=backlog_priority_levels,json=backlogPriorityLevels,proto3" json:"backlog_priority_levels,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *EphemeralData_ByVersion) Reset() {
*x = EphemeralData_ByVersion{}
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *EphemeralData_ByVersion) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*EphemeralData_ByVersion) ProtoMessage() {}
func (x *EphemeralData_ByVersion) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[11]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use EphemeralData_ByVersion.ProtoReflect.Descriptor instead.
func (*EphemeralData_ByVersion) Descriptor() ([]byte, []int) {
return file_temporal_server_api_taskqueue_v1_message_proto_rawDescGZIP(), []int{8, 0}
}
func (x *EphemeralData_ByVersion) GetVersion() *v12.WorkerDeploymentVersion {
if x != nil {
return x.Version
}
return nil
}
func (x *EphemeralData_ByVersion) GetBacklogPriorityLevels() int64 {
if x != nil {
return x.BacklogPriorityLevels
}
return 0
}
type EphemeralData_ByPartition struct {
state protoimpl.MessageState `protogen:"open.v1"`
Partition int32 `protobuf:"varint,1,opt,name=partition,proto3" json:"partition,omitempty"`
Version []*EphemeralData_ByVersion `protobuf:"bytes,2,rep,name=version,proto3" json:"version,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *EphemeralData_ByPartition) Reset() {
*x = EphemeralData_ByPartition{}
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[12]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *EphemeralData_ByPartition) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*EphemeralData_ByPartition) ProtoMessage() {}
func (x *EphemeralData_ByPartition) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[12]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use EphemeralData_ByPartition.ProtoReflect.Descriptor instead.
func (*EphemeralData_ByPartition) Descriptor() ([]byte, []int) {
return file_temporal_server_api_taskqueue_v1_message_proto_rawDescGZIP(), []int{8, 1}
}
func (x *EphemeralData_ByPartition) GetPartition() int32 {
if x != nil {
return x.Partition
}
return 0
}
func (x *EphemeralData_ByPartition) GetVersion() []*EphemeralData_ByVersion {
if x != nil {
return x.Version
}
return nil
}
var File_temporal_server_api_taskqueue_v1_message_proto protoreflect.FileDescriptor
const file_temporal_server_api_taskqueue_v1_message_proto_rawDesc = "" +
@@ -742,7 +950,18 @@ const file_temporal_server_api_taskqueue_v1_message_proto_rawDesc = "" +
"taskSource\x12Z\n" +
"\rredirect_info\x18\x03 \x01(\v25.temporal.server.api.taskqueue.v1.BuildIdRedirectInfoR\fredirectInfo\x12*\n" +
"\x11dispatch_build_id\x18\x04 \x01(\tR\x0fdispatchBuildId\x120\n" +
"\x14dispatch_version_set\x18\x05 \x01(\tR\x12dispatchVersionSetB2Z0go.temporal.io/server/api/taskqueue/v1;taskqueueb\x06proto3"
"\x14dispatch_version_set\x18\x05 \x01(\tR\x12dispatchVersionSet\"\x89\x03\n" +
"\rEphemeralData\x12Y\n" +
"\tpartition\x18\x01 \x03(\v2;.temporal.server.api.taskqueue.v1.EphemeralData.ByPartitionR\tpartition\x1a\x99\x01\n" +
"\tByVersion\x12T\n" +
"\aversion\x18\x01 \x01(\v2:.temporal.server.api.deployment.v1.WorkerDeploymentVersionR\aversion\x126\n" +
"\x17backlog_priority_levels\x18\x02 \x01(\x03R\x15backlogPriorityLevels\x1a\x80\x01\n" +
"\vByPartition\x12\x1c\n" +
"\tpartition\x18\x01 \x01(\x05R\tpartition\x12S\n" +
"\aversion\x18\x02 \x03(\v29.temporal.server.api.taskqueue.v1.EphemeralData.ByVersionR\aversion\"w\n" +
"\x16VersionedEphemeralData\x12C\n" +
"\x04data\x18\x01 \x01(\v2/.temporal.server.api.taskqueue.v1.EphemeralDataR\x04data\x12\x18\n" +
"\aversion\x18\x02 \x01(\x03R\aversionB2Z0go.temporal.io/server/api/taskqueue/v1;taskqueueb\x06proto3"
var (
file_temporal_server_api_taskqueue_v1_message_proto_rawDescOnce sync.Once
@@ -756,7 +975,7 @@ func file_temporal_server_api_taskqueue_v1_message_proto_rawDescGZIP() []byte {
return file_temporal_server_api_taskqueue_v1_message_proto_rawDescData
}
var file_temporal_server_api_taskqueue_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
var file_temporal_server_api_taskqueue_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
var file_temporal_server_api_taskqueue_v1_message_proto_goTypes = []any{
(*TaskVersionDirective)(nil), // 0: temporal.server.api.taskqueue.v1.TaskVersionDirective
(*FairLevel)(nil), // 1: temporal.server.api.taskqueue.v1.FairLevel
@@ -766,40 +985,48 @@ var file_temporal_server_api_taskqueue_v1_message_proto_goTypes = []any{
(*TaskQueuePartition)(nil), // 5: temporal.server.api.taskqueue.v1.TaskQueuePartition
(*BuildIdRedirectInfo)(nil), // 6: temporal.server.api.taskqueue.v1.BuildIdRedirectInfo
(*TaskForwardInfo)(nil), // 7: temporal.server.api.taskqueue.v1.TaskForwardInfo
nil, // 8: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry
(*emptypb.Empty)(nil), // 9: google.protobuf.Empty
(v1.VersioningBehavior)(0), // 10: temporal.api.enums.v1.VersioningBehavior
(*v11.Deployment)(nil), // 11: temporal.api.deployment.v1.Deployment
(*v12.WorkerDeploymentVersion)(nil), // 12: temporal.server.api.deployment.v1.WorkerDeploymentVersion
(*v13.TaskIdBlock)(nil), // 13: temporal.api.taskqueue.v1.TaskIdBlock
(*v13.PollerInfo)(nil), // 14: temporal.api.taskqueue.v1.PollerInfo
(*v13.TaskQueueStats)(nil), // 15: temporal.api.taskqueue.v1.TaskQueueStats
(v1.TaskQueueType)(0), // 16: temporal.api.enums.v1.TaskQueueType
(v14.TaskSource)(0), // 17: temporal.server.api.enums.v1.TaskSource
(*EphemeralData)(nil), // 8: temporal.server.api.taskqueue.v1.EphemeralData
(*VersionedEphemeralData)(nil), // 9: temporal.server.api.taskqueue.v1.VersionedEphemeralData
nil, // 10: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry
(*EphemeralData_ByVersion)(nil), // 11: temporal.server.api.taskqueue.v1.EphemeralData.ByVersion
(*EphemeralData_ByPartition)(nil), // 12: temporal.server.api.taskqueue.v1.EphemeralData.ByPartition
(*emptypb.Empty)(nil), // 13: google.protobuf.Empty
(v1.VersioningBehavior)(0), // 14: temporal.api.enums.v1.VersioningBehavior
(*v11.Deployment)(nil), // 15: temporal.api.deployment.v1.Deployment
(*v12.WorkerDeploymentVersion)(nil), // 16: temporal.server.api.deployment.v1.WorkerDeploymentVersion
(*v13.TaskIdBlock)(nil), // 17: temporal.api.taskqueue.v1.TaskIdBlock
(*v13.PollerInfo)(nil), // 18: temporal.api.taskqueue.v1.PollerInfo
(*v13.TaskQueueStats)(nil), // 19: temporal.api.taskqueue.v1.TaskQueueStats
(v1.TaskQueueType)(0), // 20: temporal.api.enums.v1.TaskQueueType
(v14.TaskSource)(0), // 21: temporal.server.api.enums.v1.TaskSource
}
var file_temporal_server_api_taskqueue_v1_message_proto_depIdxs = []int32{
9, // 0: temporal.server.api.taskqueue.v1.TaskVersionDirective.use_assignment_rules:type_name -> google.protobuf.Empty
10, // 1: temporal.server.api.taskqueue.v1.TaskVersionDirective.behavior:type_name -> temporal.api.enums.v1.VersioningBehavior
11, // 2: temporal.server.api.taskqueue.v1.TaskVersionDirective.deployment:type_name -> temporal.api.deployment.v1.Deployment
12, // 3: temporal.server.api.taskqueue.v1.TaskVersionDirective.deployment_version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
13, // 0: temporal.server.api.taskqueue.v1.TaskVersionDirective.use_assignment_rules:type_name -> google.protobuf.Empty
14, // 1: temporal.server.api.taskqueue.v1.TaskVersionDirective.behavior:type_name -> temporal.api.enums.v1.VersioningBehavior
15, // 2: temporal.server.api.taskqueue.v1.TaskVersionDirective.deployment:type_name -> temporal.api.deployment.v1.Deployment
16, // 3: temporal.server.api.taskqueue.v1.TaskVersionDirective.deployment_version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
1, // 4: temporal.server.api.taskqueue.v1.InternalTaskQueueStatus.fair_read_level:type_name -> temporal.server.api.taskqueue.v1.FairLevel
1, // 5: temporal.server.api.taskqueue.v1.InternalTaskQueueStatus.fair_ack_level:type_name -> temporal.server.api.taskqueue.v1.FairLevel
13, // 6: temporal.server.api.taskqueue.v1.InternalTaskQueueStatus.task_id_block:type_name -> temporal.api.taskqueue.v1.TaskIdBlock
17, // 6: temporal.server.api.taskqueue.v1.InternalTaskQueueStatus.task_id_block:type_name -> temporal.api.taskqueue.v1.TaskIdBlock
1, // 7: temporal.server.api.taskqueue.v1.InternalTaskQueueStatus.fair_max_read_level:type_name -> temporal.server.api.taskqueue.v1.FairLevel
4, // 8: temporal.server.api.taskqueue.v1.TaskQueueVersionInfoInternal.physical_task_queue_info:type_name -> temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo
14, // 9: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.pollers:type_name -> temporal.api.taskqueue.v1.PollerInfo
18, // 9: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.pollers:type_name -> temporal.api.taskqueue.v1.PollerInfo
2, // 10: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.internal_task_queue_status:type_name -> temporal.server.api.taskqueue.v1.InternalTaskQueueStatus
15, // 11: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.task_queue_stats:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
8, // 12: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.task_queue_stats_by_priority_key:type_name -> temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry
16, // 13: temporal.server.api.taskqueue.v1.TaskQueuePartition.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
17, // 14: temporal.server.api.taskqueue.v1.TaskForwardInfo.task_source:type_name -> temporal.server.api.enums.v1.TaskSource
19, // 11: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.task_queue_stats:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
10, // 12: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.task_queue_stats_by_priority_key:type_name -> temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry
20, // 13: temporal.server.api.taskqueue.v1.TaskQueuePartition.task_queue_type:type_name -> temporal.api.enums.v1.TaskQueueType
21, // 14: temporal.server.api.taskqueue.v1.TaskForwardInfo.task_source:type_name -> temporal.server.api.enums.v1.TaskSource
6, // 15: temporal.server.api.taskqueue.v1.TaskForwardInfo.redirect_info:type_name -> temporal.server.api.taskqueue.v1.BuildIdRedirectInfo
15, // 16: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry.value:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
17, // [17:17] is the sub-list for method output_type
17, // [17:17] is the sub-list for method input_type
17, // [17:17] is the sub-list for extension type_name
17, // [17:17] is the sub-list for extension extendee
0, // [0:17] is the sub-list for field type_name
12, // 16: temporal.server.api.taskqueue.v1.EphemeralData.partition:type_name -> temporal.server.api.taskqueue.v1.EphemeralData.ByPartition
8, // 17: temporal.server.api.taskqueue.v1.VersionedEphemeralData.data:type_name -> temporal.server.api.taskqueue.v1.EphemeralData
19, // 18: temporal.server.api.taskqueue.v1.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKeyEntry.value:type_name -> temporal.api.taskqueue.v1.TaskQueueStats
16, // 19: temporal.server.api.taskqueue.v1.EphemeralData.ByVersion.version:type_name -> temporal.server.api.deployment.v1.WorkerDeploymentVersion
11, // 20: temporal.server.api.taskqueue.v1.EphemeralData.ByPartition.version:type_name -> temporal.server.api.taskqueue.v1.EphemeralData.ByVersion
21, // [21:21] is the sub-list for method output_type
21, // [21:21] is the sub-list for method input_type
21, // [21:21] is the sub-list for extension type_name
21, // [21:21] is the sub-list for extension extendee
0, // [0:21] is the sub-list for field type_name
}
func init() { file_temporal_server_api_taskqueue_v1_message_proto_init() }
@@ -821,7 +1048,7 @@ func file_temporal_server_api_taskqueue_v1_message_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc), len(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 9,
NumMessages: 13,
NumExtensions: 0,
NumServices: 0,
},

6
common/backoff/jitter.go
View File

@@ -1,10 +1,6 @@
package backoff
import (
"math/rand"
)
const fullCoefficient float64 = 1
import "math/rand"
// FullJitter return random number from 0 to input, inclusive, exclusive
func FullJitter[T ~int64 | ~int | ~int32 | ~float64 | ~float32](input T) T {

17
common/dynamicconfig/constants.go
View File

@@ -1253,11 +1253,24 @@ This can help reduce effects of task queue movement.`,
5*time.Minute,
`MatchingGetUserDataRefresh is how often the user data owner refreshes data from persistence.`,
)
MatchingEphemeralDataUpdateInterval = NewTaskQueueDurationSetting(
"matching.ephemeralDataUpdateInterval",
5*time.Second,
`How often to update ephemeral data (e.g. backlog size for forwarding sticky polls).
Set to zero to disable ephemeral data updates.`,
)
MatchingPriorityBacklogForwarding = NewTaskQueueBoolSetting(
"matching.priorityBacklogForwarding",
true,
`Whether to forward polls to partitions with higher-priority backlog.`,
)
MatchingBacklogNegligibleAge = NewTaskQueueDurationSetting(
"matching.backlogNegligibleAge",
5*time.Second,
`MatchingBacklogNegligibleAge if the head of backlog gets older than this we stop sync match and
forwarding to ensure more equal dispatch order among partitions.`,
`MatchingBacklogNegligibleAge is a threshold for negligible vs significant backlogs:
If the head of the backlog is older than this, then we stop sync match and forwarding to ensure
more equal dispatch order among partitions. We also forward sticky polls to partitions with
higher-priority backlog.`,
)
MatchingMaxWaitForPollerBeforeFwd = NewTaskQueueDurationSetting(
"matching.maxWaitForPollerBeforeFwd",

7
proto/internal/temporal/server/api/matchingservice/v1/request_response.proto
View File

@@ -336,8 +336,10 @@ message GetTaskQueueUserDataRequest {
// If the requester has no data, it should set this to 0.
// This value must not be set to a negative number (note that our linter suggests avoiding uint64).
int64 last_known_user_data_version = 3;
// If set and last_known_user_data_version is the current version, block until new data is
// available (or timeout).
// Same for ephemeral data.
int64 last_known_ephemeral_data_version = 7;
// If set and last_known_{user_data,ephemeral_data}_version is the current version,
// block until new data is available (or timeout).
bool wait_new_data = 4;
// If set, do not load task queue if unloaded. (Returns FailedPrecondition error in that case.)
bool only_if_loaded = 6;
@@ -348,6 +350,7 @@ message GetTaskQueueUserDataResponse {
// Versioned user data, set if the task queue has user data and the request's last_known_user_data_version is less
// than the version cached in the root partition.
temporal.server.api.persistence.v1.VersionedTaskQueueUserData user_data = 2;
temporal.server.api.taskqueue.v1.VersionedEphemeralData ephemeral_data = 3;
}
message SyncDeploymentUserDataRequest {

26
proto/internal/temporal/server/api/taskqueue/v1/message.proto
View File

@@ -116,3 +116,29 @@ message TaskForwardInfo {
// Deprecated. [cleanup-old-wv]
string dispatch_version_set = 5;
}
// EphemeralData is data that we want to propagate among task queue partitions, but is not persisted.
// Ephemeral data is propagated alongside "task queue user data", but while user data applies to a
// task queue family (all queues with the same name, across types), ephemeral data applies only to
// one type at a time.
message EphemeralData {
message ByVersion {
// Key for this data. Data for the unversioned queue has no version field present.
// All following fields are data associated with this versioned queue.
temporal.server.api.deployment.v1.WorkerDeploymentVersion version = 1;
// This is a bit field of priority levels that have "significant" backlog (defined by
// the server configuration). Priority key k corresponds to 1<<k.
int64 backlog_priority_levels = 2;
}
message ByPartition {
int32 partition = 1;
repeated ByVersion version = 2;
}
repeated ByPartition partition = 1;
}
message VersionedEphemeralData {
EphemeralData data = 1;
int64 version = 2;
}

19
service/matching/config.go
View File

@@ -14,11 +14,6 @@ import (
"go.temporal.io/server/service/matching/counter"
)
const (
// Maximum value for priority levels.
maxPriorityLevels = 100
)
type (
// Config represents configuration for matching service
Config struct {
@@ -82,6 +77,8 @@ type (
TaskQueueLimitPerBuildId dynamicconfig.IntPropertyFnWithNamespaceFilter
GetUserDataLongPollTimeout dynamicconfig.DurationPropertyFn
GetUserDataRefresh dynamicconfig.DurationPropertyFn
EphemeralDataUpdateInterval dynamicconfig.DurationPropertyFnWithTaskQueueFilter
PriorityBacklogForwarding dynamicconfig.BoolPropertyFnWithTaskQueueFilter
BacklogNegligibleAge dynamicconfig.DurationPropertyFnWithTaskQueueFilter
MaxWaitForPollerBeforeFwd dynamicconfig.DurationPropertyFnWithTaskQueueFilter
QueryPollerUnavailableWindow dynamicconfig.DurationPropertyFn
@@ -144,6 +141,8 @@ type (
taskQueueConfig struct {
forwarderConfig
SyncMatchWaitDuration func() time.Duration
EphemeralDataUpdateInterval func() time.Duration
PriorityBacklogForwarding func() bool
BacklogNegligibleAge func() time.Duration
MaxWaitForPollerBeforeFwd func() time.Duration
QueryPollerUnavailableWindow func() time.Duration
@@ -313,6 +312,8 @@ func NewConfig(
TaskQueueLimitPerBuildId: dynamicconfig.TaskQueuesPerBuildIdLimit.Get(dc),
GetUserDataLongPollTimeout: dynamicconfig.MatchingGetUserDataLongPollTimeout.Get(dc), // Use -10 seconds so that we send back empty response instead of timeout
GetUserDataRefresh: dynamicconfig.MatchingGetUserDataRefresh.Get(dc),
EphemeralDataUpdateInterval: dynamicconfig.MatchingEphemeralDataUpdateInterval.Get(dc),
PriorityBacklogForwarding: dynamicconfig.MatchingPriorityBacklogForwarding.Get(dc),
BacklogNegligibleAge: dynamicconfig.MatchingBacklogNegligibleAge.Get(dc),
MaxWaitForPollerBeforeFwd: dynamicconfig.MatchingMaxWaitForPollerBeforeFwd.Get(dc),
QueryPollerUnavailableWindow: dynamicconfig.QueryPollerUnavailableWindow.Get(dc),
@@ -392,6 +393,12 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
SyncMatchWaitDuration: func() time.Duration {
return config.SyncMatchWaitDuration(ns.String(), taskQueueName, taskType)
},
EphemeralDataUpdateInterval: func() time.Duration {
return config.EphemeralDataUpdateInterval(ns.String(), taskQueueName, taskType)
},
PriorityBacklogForwarding: func() bool {
return config.PriorityBacklogForwarding(ns.String(), taskQueueName, taskType)
},
BacklogNegligibleAge: func() time.Duration {
return config.BacklogNegligibleAge(ns.String(), taskQueueName, taskType)
},
@@ -509,6 +516,6 @@ func (c *taskQueueConfig) clipPriority(priority priorityKey) priorityKey {
func (c *taskQueueConfig) setDefaultPriority(task *internalTask) {
if task.effectivePriority == 0 {
task.effectivePriority = c.DefaultPriorityKey
task.effectivePriority = effectivePriorityFactor * c.DefaultPriorityKey
}
}

2
service/matching/fair_backlog_manager.go
View File

@@ -390,6 +390,6 @@ func (c *fairBacklogManagerImpl) setPriority(task *internalTask) {
c.config.setDefaultPriority(task)
if c.isDraining {
// draining goes before active backlog so we're guaranteed to finish migration
task.effectivePriority -= maxPriorityLevels
task.effectivePriority -= effectivePriorityFactor * maxPriorityLevels
}
}

2
service/matching/matcher.go
View File

@@ -39,7 +39,6 @@ type TaskMatcher struct {
fwdr *Forwarder
metricsHandler metrics.Handler // namespace metric scope
numPartitions func() int // number of task queue partitions
backlogTasksCreateTime map[int64]int // task creation time (unix nanos) -> number of tasks with that time
backlogTasksLock sync.Mutex
lastPoller atomic.Int64 // unix nanos of most recent poll start time
@@ -62,7 +61,6 @@ func newTaskMatcher(config *taskQueueConfig, fwdr *Forwarder, metricsHandler met
taskC: make(chan *internalTask),
queryTaskC: make(chan *internalTask),
closeC: make(chan struct{}),
numPartitions: config.NumReadPartitions,
backlogTasksCreateTime: make(map[int64]int),
}
}

18
service/matching/matcher_data.go
View File

@@ -15,8 +15,16 @@ import (
)
const (
invalidHeapIndex = -13 // use unusual value to stand out in panics
pollForwarderPriority = 1000000 // lower than any other priority. must be > maxPriorityLevels.
invalidHeapIndex = -13 // use unusual value to stand out in panics
maxPriorityLevels = 60 // maximum value for priority levels (fits in a bitfield with a few bits reserved)
effectivePriorityFactor = 10 // multiply priority level by this to leave room for intermediate levels
pollForwarderPriority = 1000000 // lower than any other priority. must be > maxPriorityLevels*effectivePriorityFactor.
)
const (
notPollForwarder pollForwarderType = iota
normalPollForwarder
priorityBacklogPollForwarder
)
// maxTokens is the maximum number of tokens we might consume at a time for simpleLimiter. This
@@ -388,7 +396,9 @@ func (d *matcherData) findMatch(allowForwarding bool) (*internalTask, *waitingPo
// TODO(pri): optimize so it's not O(d*n) worst case
// TODO(pri): this iterates over heap as slice, which isn't quite correct, but okay for now
for _, task := range d.tasks.heap {
if !allowForwarding && task.isPollForwarder() {
// disallow normal poll forwarding when allowForwarding is false, but allow the
// "priority backlog poll forwarders".
if !allowForwarding && task.pollForwarderType == normalPollForwarder {
continue
}
@@ -402,7 +412,7 @@ func (d *matcherData) findMatch(allowForwarding bool) (*internalTask, *waitingPo
continue
} else if poller.isTaskValidator && task.forwardCtx != nil {
continue
} else if mp := poller.minPriority(); mp > 0 && task.effectivePriority > mp {
} else if mp := poller.minPriority(); mp > 0 && task.effectivePriority > effectivePriorityFactor*mp {
// Note the ">" above: "min" priority is a numeric max.
// Also note: this condition will be false for draining tasks since we artifically boost
// their priority above "1". that's inaccurate but it's just a temporary situation.

10
service/matching/matcher_data_test.go
View File

@@ -415,7 +415,7 @@ func (s *MatcherDataSuite) TestOrder() {
t1 := s.newBacklogTaskWithPriority(1, 0, nil, &commonpb.Priority{PriorityKey: 1})
t2 := s.newBacklogTaskWithPriority(2, 0, nil, &commonpb.Priority{PriorityKey: 2})
t3 := s.newBacklogTaskWithPriority(3, 0, nil, &commonpb.Priority{PriorityKey: 3})
tf := newPollForwarderTask()
tf := newPollForwarderTask(pollForwarderPriority, normalPollForwarder)
s.md.EnqueueTaskNoWait(t3)
s.md.EnqueueTaskNoWait(tf)
@@ -437,7 +437,7 @@ func (s *MatcherDataSuite) TestPollForwardSuccess() {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask())
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask(pollForwarderPriority, normalPollForwarder))
// task is woken up with poller to forward
s.NotNil(tres.poller)
// forward succeeded, pass back task
@@ -459,7 +459,7 @@ func (s *MatcherDataSuite) TestPollForwardFailed() {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask())
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask(pollForwarderPriority, normalPollForwarder))
// task is woken up with poller to forward
s.NotNil(tres.poller)
// there's a new task in the meantime
@@ -484,7 +484,7 @@ func (s *MatcherDataSuite) TestPollForwardFailedTimedOut() {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask())
tres := s.md.EnqueueTaskAndWait([]context.Context{ctx}, newPollForwarderTask(pollForwarderPriority, normalPollForwarder))
// task is woken up with poller to forward
s.NotNil(tres.poller)
// there's a new task in the meantime
@@ -847,7 +847,7 @@ func FuzzMatcherData(f *testing.F) {
sleepTime := randms(100)
go func() {
defer pollForwarders.Add(-1)
res := md.EnqueueTaskAndWait(nil, newPollForwarderTask())
res := md.EnqueueTaskAndWait(nil, newPollForwarderTask(pollForwarderPriority, normalPollForwarder))
softassert.That(md.logger, res.ctxErr == nil && res.poller != nil, "")
ts.Sleep(sleepTime)
t := &persistencespb.TaskInfo{

4
service/matching/matching_engine.go
View File

@@ -429,11 +429,13 @@ func (e *matchingEngineImpl) getTaskQueuePartitionManager(
logger, throttledLogger, metricsHandler := e.loggerAndMetricsForPartition(namespaceEntry, partition, tqConfig)
onFatalErr := func(cause unloadCause) { newPM.unloadFromEngine(cause) }
onUserDataChanged := func(to *persistencespb.VersionedTaskQueueUserData) { newPM.userDataChanged(to) }
onEphemeralDataChanged := func(data *taskqueuespb.EphemeralData) { newPM.ephemeralDataChanged(data) }
userDataManager := newUserDataManager(
e.taskManager,
e.matchingRawClient,
onFatalErr,
onUserDataChanged,
onEphemeralDataChanged,
partition,
tqConfig,
logger,
@@ -3023,8 +3025,8 @@ func (e *matchingEngineImpl) recordWorkflowTaskStarted(
// TODO: stop sending ScheduledDeployment. [cleanup-old-wv]
ScheduledDeployment: worker_versioning.DirectiveDeployment(task.event.Data.VersionDirective),
VersionDirective: task.event.Data.VersionDirective,
TaskDispatchRevisionNumber: task.taskDispatchRevisionNumber,
Stamp: task.event.Data.GetStamp(),
TaskDispatchRevisionNumber: task.taskDispatchRevisionNumber,
TargetDeploymentVersion: sentTargetVersion,
}

13
service/matching/physical_task_queue_key.go
View File

@@ -7,6 +7,7 @@ import (
deploymentpb "go.temporal.io/api/deployment/v1"
enumspb "go.temporal.io/api/enums/v1"
deploymentspb "go.temporal.io/server/api/deployment/v1"
"go.temporal.io/server/common/tqid"
"go.temporal.io/server/common/worker_versioning"
)
@@ -166,6 +167,18 @@ func (v PhysicalTaskQueueVersion) Deployment() *deploymentpb.Deployment {
return nil
}
// WorkerDeploymentVersionS returns the internal server api WorkerDeploymentVersion
// (different from the public api WorkerDeploymentVersion).
func (v PhysicalTaskQueueVersion) WorkerDeploymentVersionS() *deploymentspb.WorkerDeploymentVersion {
if len(v.deploymentSeriesName) > 0 {
return &deploymentspb.WorkerDeploymentVersion{
BuildId: v.buildId,
DeploymentName: v.deploymentSeriesName,
}
}
return nil
}
// BuildId returns empty if this is not a Versioning v2 queue.
func (v PhysicalTaskQueueVersion) BuildId() string {
if len(v.deploymentSeriesName) > 0 {

28
service/matching/physical_task_queue_manager.go
View File

@@ -206,6 +206,7 @@ func newPhysicalTaskQueueManager(
config,
queue.partition,
fwdr,
pqMgr.matchingClient,
pqMgr.taskValidator,
pqMgr.logger,
newFairMetricsHandler(taggedMetricsHandler),
@@ -244,6 +245,7 @@ func newPhysicalTaskQueueManager(
config,
queue.partition,
fwdr,
pqMgr.matchingClient,
pqMgr.taskValidator,
pqMgr.logger,
newPriMetricsHandler(taggedMetricsHandler),
@@ -605,7 +607,7 @@ func (c *physicalTaskQueueManagerImpl) LegacyDescribeTaskQueue(includeTaskQueueS
return response
}
func (c *physicalTaskQueueManagerImpl) GetStatsByPriority() map[int32]*taskqueuepb.TaskQueueStats {
func (c *physicalTaskQueueManagerImpl) GetStatsByPriority(includeRates bool) map[int32]*taskqueuepb.TaskQueueStats {
stats := c.backlogMgr.BacklogStatsByPriority()
if m := c.drainBacklogMgr.Load(); m != nil {
@@ -615,15 +617,17 @@ func (c *physicalTaskQueueManagerImpl) GetStatsByPriority() map[int32]*taskqueue
}
}
c.taskTrackerLock.RLock()
defer c.taskTrackerLock.RUnlock()
if includeRates {
c.taskTrackerLock.RLock()
for pri, tt := range c.tasksAdded {
util.GetOrSetNew(stats, int32(pri)).TasksAddRate = tt.rate()
}
for pri, tt := range c.tasksDispatched {
util.GetOrSetNew(stats, int32(pri)).TasksDispatchRate = tt.rate()
}
c.taskTrackerLock.RUnlock()
}
for pri, tt := range c.tasksAdded {
util.GetOrSetNew(stats, int32(pri)).TasksAddRate = tt.rate()
}
for pri, tt := range c.tasksDispatched {
util.GetOrSetNew(stats, int32(pri)).TasksDispatchRate = tt.rate()
}
return stats
}
@@ -845,6 +849,12 @@ func (c *physicalTaskQueueManagerImpl) makePollerScalingDecisionImpl(
}
}
func (c *physicalTaskQueueManagerImpl) UpdateRemotePriorityBacklogs(backlogs remotePriorityBacklogSet) {
if c.priMatcher != nil {
c.priMatcher.UpdateRemotePriorityBacklogs(backlogs)
}
}
func (c *physicalTaskQueueManagerImpl) getOrCreateTaskTracker(
intervals map[priorityKey]*taskTracker,
priorityKey priorityKey,

3
service/matching/physical_task_queue_manager_interface.go
View File

@@ -48,7 +48,7 @@ type (
HasPollerAfter(accessTime time.Time) bool
// LegacyDescribeTaskQueue returns pollers info and legacy TaskQueueStatus for this physical queue
LegacyDescribeTaskQueue(includeTaskQueueStatus bool) *matchingservice.DescribeTaskQueueResponse
GetStatsByPriority() map[int32]*taskqueuepb.TaskQueueStats
GetStatsByPriority(includeRates bool) map[int32]*taskqueuepb.TaskQueueStats
GetInternalTaskQueueStatus() []*taskqueuespb.InternalTaskQueueStatus
UnloadFromPartitionManager(unloadCause)
QueueKey() *PhysicalTaskQueueKey
@@ -57,5 +57,6 @@ type (
MakePollerScalingDecision(ctx context.Context, pollStartTime time.Time) *taskqueuepb.PollerScalingDecision
// GetFairnessWeightOverrides returns current fairness weight overrides for this queue.
GetFairnessWeightOverrides() fairnessWeightOverrides
UpdateRemotePriorityBacklogs(remotePriorityBacklogSet)
}
)

20
service/matching/physical_task_queue_manager_mock.go
View File

@@ -158,17 +158,17 @@ func (mr *MockphysicalTaskQueueManagerMockRecorder) GetInternalTaskQueueStatus()
}
// GetStatsByPriority mocks base method.
func (m *MockphysicalTaskQueueManager) GetStatsByPriority() map[int32]*taskqueue.TaskQueueStats {
func (m *MockphysicalTaskQueueManager) GetStatsByPriority(includeRates bool) map[int32]*taskqueue.TaskQueueStats {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetStatsByPriority")
ret := m.ctrl.Call(m, "GetStatsByPriority", includeRates)
ret0, _ := ret[0].(map[int32]*taskqueue.TaskQueueStats)
return ret0
}
// GetStatsByPriority indicates an expected call of GetStatsByPriority.
func (mr *MockphysicalTaskQueueManagerMockRecorder) GetStatsByPriority() *gomock.Call {
func (mr *MockphysicalTaskQueueManagerMockRecorder) GetStatsByPriority(includeRates any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsByPriority", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).GetStatsByPriority))
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsByPriority", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).GetStatsByPriority), includeRates)
}
// HasPollerAfter mocks base method.
@@ -357,6 +357,18 @@ func (mr *MockphysicalTaskQueueManagerMockRecorder) UpdatePollerInfo(arg0, arg1
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdatePollerInfo", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).UpdatePollerInfo), arg0, arg1)
}
// UpdateRemotePriorityBacklogs mocks base method.
func (m *MockphysicalTaskQueueManager) UpdateRemotePriorityBacklogs(arg0 remotePriorityBacklogSet) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "UpdateRemotePriorityBacklogs", arg0)
}
// UpdateRemotePriorityBacklogs indicates an expected call of UpdateRemotePriorityBacklogs.
func (mr *MockphysicalTaskQueueManagerMockRecorder) UpdateRemotePriorityBacklogs(arg0 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateRemotePriorityBacklogs", reflect.TypeOf((*MockphysicalTaskQueueManager)(nil).UpdateRemotePriorityBacklogs), arg0)
}
// UserDataChanged mocks base method.
func (m *MockphysicalTaskQueueManager) UserDataChanged() {
m.ctrl.T.Helper()

2
service/matching/physical_task_queue_manager_test.go
View File

@@ -80,7 +80,7 @@ func (s *PhysicalTaskQueueManagerTestSuite) SetupTest() {
prtn := s.physicalTaskQueueKey.Partition()
tqConfig := newTaskQueueConfig(prtn.TaskQueue(), engine.config, nsName)
onFatalErr := func(unloadCause) { s.T().Fatal("user data manager called onFatalErr") }
udMgr := newUserDataManager(engine.taskManager, engine.matchingRawClient, onFatalErr, nil, prtn, tqConfig, engine.logger, engine.namespaceRegistry)
udMgr := newUserDataManager(engine.taskManager, engine.matchingRawClient, onFatalErr, nil, nil, prtn, tqConfig, engine.logger, engine.namespaceRegistry)
prtnMgr, err := newTaskQueuePartitionManager(engine, ns, prtn, tqConfig, engine.logger, nil, metrics.NoopMetricsHandler, udMgr)
s.NoError(err)

2
service/matching/pri_backlog_manager.go
View File

@@ -395,6 +395,6 @@ func (c *priBacklogManagerImpl) setPriority(task *internalTask) {
c.config.setDefaultPriority(task)
if c.isDraining {
// draining goes before active backlog so we're guaranteed to finish migration
task.effectivePriority -= maxPriorityLevels
task.effectivePriority -= effectivePriorityFactor * maxPriorityLevels
}
}

38
service/matching/pri_forwarder.go
View File

@@ -182,6 +182,7 @@ func (f *priForwarder) ForwardNexusTask(ctx context.Context, task *internalTask)
}
// ForwardPoll forwards a poll request to parent task queue partition if it exist
// TODO(pri): remove this and update tests to call ForwardPollWithTarget directly
func (f *priForwarder) ForwardPoll(ctx context.Context, pollMetadata *pollMetadata) (*internalTask, error) {
degree := f.cfg.ForwarderMaxChildrenPerNode()
target, err := f.partition.ParentPartition(degree)
@@ -189,25 +190,36 @@ func (f *priForwarder) ForwardPoll(ctx context.Context, pollMetadata *pollMetada
return nil, err
}
return ForwardPollWithTarget(ctx, pollMetadata, f.client, f.partition, target)
}
// ForwardPollWithTarget forwards a poll request to another partition
func ForwardPollWithTarget(
ctx context.Context,
pollMetadata *pollMetadata,
client matchingservice.MatchingServiceClient,
source tqid.Partition,
target *tqid.NormalPartition,
) (*internalTask, error) {
pollerID, _ := ctx.Value(pollerIDKey).(string) // nolint:revive
identity, _ := ctx.Value(identityKey).(string) // nolint:revive
// nolint:exhaustive // there's a default clause
switch f.partition.TaskType() {
switch target.TaskType() {
case enumspb.TASK_QUEUE_TYPE_WORKFLOW:
resp, err := f.client.PollWorkflowTaskQueue(ctx, &matchingservice.PollWorkflowTaskQueueRequest{
NamespaceId: f.partition.TaskQueue().NamespaceId(),
resp, err := client.PollWorkflowTaskQueue(ctx, &matchingservice.PollWorkflowTaskQueueRequest{
NamespaceId: target.TaskQueue().NamespaceId(),
PollerId: pollerID,
PollRequest: &workflowservice.PollWorkflowTaskQueueRequest{
TaskQueue: &taskqueuepb.TaskQueue{
Name: target.RpcName(),
Kind: f.partition.Kind(),
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: identity,
WorkerVersionCapabilities: pollMetadata.workerVersionCapabilities,
DeploymentOptions: pollMetadata.deploymentOptions,
},
ForwardedSource: f.partition.RpcName(),
ForwardedSource: source.RpcName(),
Conditions: pollMetadata.conditions,
})
if err != nil {
@@ -217,20 +229,20 @@ func (f *priForwarder) ForwardPoll(ctx context.Context, pollMetadata *pollMetada
}
return newInternalStartedTask(&startedTaskInfo{workflowTaskInfo: resp}), nil
case enumspb.TASK_QUEUE_TYPE_ACTIVITY:
resp, err := f.client.PollActivityTaskQueue(ctx, &matchingservice.PollActivityTaskQueueRequest{
NamespaceId: f.partition.TaskQueue().NamespaceId(),
resp, err := client.PollActivityTaskQueue(ctx, &matchingservice.PollActivityTaskQueueRequest{
NamespaceId: target.TaskQueue().NamespaceId(),
PollerId: pollerID,
PollRequest: &workflowservice.PollActivityTaskQueueRequest{
TaskQueue: &taskqueuepb.TaskQueue{
Name: target.RpcName(),
Kind: f.partition.Kind(),
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: identity,
TaskQueueMetadata: pollMetadata.taskQueueMetadata,
WorkerVersionCapabilities: pollMetadata.workerVersionCapabilities,
DeploymentOptions: pollMetadata.deploymentOptions,
},
ForwardedSource: f.partition.RpcName(),
ForwardedSource: source.RpcName(),
Conditions: pollMetadata.conditions,
})
if err != nil {
@@ -240,20 +252,20 @@ func (f *priForwarder) ForwardPoll(ctx context.Context, pollMetadata *pollMetada
}
return newInternalStartedTask(&startedTaskInfo{activityTaskInfo: resp}), nil
case enumspb.TASK_QUEUE_TYPE_NEXUS:
resp, err := f.client.PollNexusTaskQueue(ctx, &matchingservice.PollNexusTaskQueueRequest{
NamespaceId: f.partition.TaskQueue().NamespaceId(),
resp, err := client.PollNexusTaskQueue(ctx, &matchingservice.PollNexusTaskQueueRequest{
NamespaceId: target.TaskQueue().NamespaceId(),
PollerId: pollerID,
Request: &workflowservice.PollNexusTaskQueueRequest{
TaskQueue: &taskqueuepb.TaskQueue{
Name: target.RpcName(),
Kind: f.partition.Kind(),
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: identity,
WorkerVersionCapabilities: pollMetadata.workerVersionCapabilities,
DeploymentOptions: pollMetadata.deploymentOptions,
// Namespace is ignored here.
},
ForwardedSource: f.partition.RpcName(),
ForwardedSource: source.RpcName(),
Conditions: pollMetadata.conditions,
})
if err != nil {

137
service/matching/pri_matcher.go
View File

@@ -1,7 +1,9 @@
package matching
import (
"cmp"
"context"
"errors"
"strconv"
"time"
@@ -11,6 +13,7 @@ import (
"go.temporal.io/server/common"
"go.temporal.io/server/common/backoff"
"go.temporal.io/server/common/clock"
"go.temporal.io/server/common/goro"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/primitives/timestamp"
@@ -34,12 +37,14 @@ type priTaskMatcher struct {
partition tqid.Partition
fwdr *priForwarder
client matchingservice.MatchingServiceClient
validator taskValidator
rateLimitManager *rateLimitManager
metricsHandler metrics.Handler // namespace metric scope
logger log.Logger
numPartitions func() int // number of task queue partitions
markAlive func() // function to mark the physical task queue alive
markAlive func() // function to mark the physical task queue alive
priorityBacklogForwarders *goro.KeyedSet[remotePriorityBacklog]
}
type waitingPoller struct {
@@ -59,6 +64,18 @@ type matchResult struct {
ctxErrIdx int // index of context that closed first
}
// remotePriorityBacklog represents the fact that a specific normal partition has a significant
// backlog at a specific priority level. This is used to forward polls to partitions that have
// higher priority backlogs than the partition they arrived at.
type remotePriorityBacklog struct {
partition int32
priority priorityKey
}
type remotePriorityBacklogSet = map[remotePriorityBacklog]struct{}
type pollForwarderType int32
var (
// TODO(pri): old matcher cleanup, move to here
// errNoRecentPoller = status.Error(codes.FailedPrecondition, "no poller seen for task queue recently, worker may be down")
@@ -78,6 +95,7 @@ func newPriTaskMatcher(
config *taskQueueConfig,
partition tqid.Partition,
fwdr *priForwarder,
client matchingservice.MatchingServiceClient,
validator taskValidator,
logger log.Logger,
metricsHandler metrics.Handler,
@@ -85,17 +103,18 @@ func newPriTaskMatcher(
markAlive func(),
) *priTaskMatcher {
tm := &priTaskMatcher{
config: config,
data: newMatcherData(config, logger, clock.NewRealTimeSource(), fwdr != nil, rateLimitManager),
tqCtx: tqCtx,
logger: logger,
metricsHandler: metricsHandler,
partition: partition,
fwdr: fwdr,
validator: validator,
rateLimitManager: rateLimitManager,
numPartitions: config.NumReadPartitions,
markAlive: markAlive,
config: config,
data: newMatcherData(config, logger, clock.NewRealTimeSource(), fwdr != nil, rateLimitManager),
tqCtx: tqCtx,
logger: logger,
metricsHandler: metricsHandler,
partition: partition,
fwdr: fwdr,
client: client,
validator: validator,
rateLimitManager: rateLimitManager,
markAlive: markAlive,
priorityBacklogForwarders: goro.NewKeyedSet[remotePriorityBacklog](tqCtx),
}
return tm
@@ -109,22 +128,34 @@ func (tm *priTaskMatcher) Start() {
lim := quotas.NewDefaultOutgoingRateLimiter(tm.config.ForwarderMaxRatePerSecond)
if tm.fwdr == nil {
// Root doesn't forward. But it does need something to validate tasks.
// Root/sticky doesn't forward. But it does need something to validate tasks.
go tm.validateTasksOnRoot(retrier)
return
}
// Child partitions:
// Non-root normal partitions:
// TODO(pri): ForwarderMaxOutstandingTasks > 1 is not supported: it will cause alternating
// tasks to be sent to the validator, which will make the validator not validate anything.
for range tm.config.ForwarderMaxOutstandingTasks() {
go tm.forwardTasks(lim, retrier)
}
for range tm.config.ForwarderMaxOutstandingPolls() {
go tm.forwardPolls()
if normal, ok := tm.partition.(*tqid.NormalPartition); ok { // always true here, sticky has nil tm.fwdr
degree := tm.config.ForwarderMaxChildrenPerNode()
if parent, err := normal.ParentPartition(degree); err == nil {
for range tm.config.ForwarderMaxOutstandingPolls() {
go tm.forwardPolls(tm.tqCtx, 0, pollForwarderPriority, normalPollForwarder, parent)
}
}
}
}
func (tm *priTaskMatcher) Stop() {}
func (tm *priTaskMatcher) Stop() {
tm.priorityBacklogForwarders.Sync(nil, nil)
}
// TODO(pri): access to retrier is not synchronized
func (tm *priTaskMatcher) forwardTasks(lim quotas.RateLimiter, retrier backoff.Retrier) {
ctxs := []context.Context{tm.tqCtx}
poller := waitingPoller{isTaskForwarder: true}
@@ -160,7 +191,11 @@ func (tm *priTaskMatcher) forwardTask(task *internalTask) (bool, error) {
if task.forwardCtx != nil {
// Use sync match context if we have it (for deadline, headers, etc.)
// TODO(pri): does it make sense to subtract 1s from the context deadline here?
ctx = task.forwardCtx
// Also arrange for this to be canceled on tqCtx closing.
ctx, cancel = context.WithCancel(task.forwardCtx)
stop := context.AfterFunc(tm.tqCtx, cancel)
defer cancel()
defer stop()
} else {
// Task is from local backlog.
@@ -245,10 +280,21 @@ func (tm *priTaskMatcher) validateTasksOnRoot(retrier backoff.Retrier) {
}
}
func (tm *priTaskMatcher) forwardPolls() {
forwarderTask := newPollForwarderTask()
ctxs := []context.Context{tm.tqCtx}
for {
func (tm *priTaskMatcher) forwardPolls(
ctx context.Context,
targetPriority, effectivePriority priorityKey,
ft pollForwarderType,
target *tqid.NormalPartition,
) {
forwarderTask := newPollForwarderTask(effectivePriority, ft)
ctxs := []context.Context{ctx} // ctx should be equal to or child of tm.tqCtx
for ctx.Err() == nil {
if ft == priorityBacklogPollForwarder && !tm.config.PriorityBacklogForwarding() {
// if this feature has been disabled, just wait
_ = util.InterruptibleSleep(ctx, time.Minute)
continue
}
res := tm.data.EnqueueTaskAndWait(ctxs, forwarderTask)
if res.ctxErr != nil {
return // task queue closing
@@ -258,11 +304,35 @@ func (tm *priTaskMatcher) forwardPolls() {
}
poller := res.poller
meta := poller.pollMetadata
if ft == priorityBacklogPollForwarder {
// This is a forwarder for high-priority backlog on another partition. Override the min
// priority so we get only that backlog and not any other tasks.
pmCopy := *meta
pmCopy.conditions = &matchingservice.PollConditions{
MinPriority: int32(targetPriority),
NoWait: true,
}
meta = &pmCopy
}
// We need to use the real source poller context since it has the poller id and
// identity, plus the right deadline.
task, err := tm.fwdr.ForwardPoll(poller.forwardCtx, poller.pollMetadata)
// identity, plus the right deadline. But we also want to cancel this if our ctx
// closes, so use AfterFunc to join them.
callCtx, cancel := context.WithCancel(poller.forwardCtx)
stop := context.AfterFunc(ctx, cancel)
task, err := ForwardPollWithTarget(callCtx, meta, tm.client, tm.partition, target)
cancel()
_ = stop()
if err == nil {
tm.data.FinishMatchAfterPollForward(poller, task)
} else if ft == priorityBacklogPollForwarder && errors.Is(err, errNoTasks) {
// There are no tasks of the priority we're looking for on the target. This goroutine
// will probably get canceled as soon as ephemeral data updates. In the meantime, wait.
// Re-enqueue but don't disable other forwarders for this poll.
tm.data.ReenqueuePollerIfNotMatched(poller)
// 4× to allow for a few rounds plus propagation.
interval := cmp.Or(tm.config.EphemeralDataUpdateInterval(), time.Minute)
_ = util.InterruptibleSleep(ctx, 4*interval)
} else {
// Re-enqueue to let it match again, if it hasn't gotten a context timeout already.
poller.forwardCtx = nil // disable forwarding next time
@@ -468,6 +538,23 @@ func (tm *priTaskMatcher) ReprocessAllTasks() {
}
}
func (tm *priTaskMatcher) UpdateRemotePriorityBacklogs(backlogs remotePriorityBacklogSet) {
if !tm.config.PriorityBacklogForwarding() {
// if this feature is disabled, stop all forwarders
backlogs = nil
}
// note that only sticky queues get here (for now).
// we want to set up poll forwarders for these levels to send polls to normal partitions
// if they have backlog at higher priority than our backlog.
tm.priorityBacklogForwarders.Sync(backlogs, func(ctx context.Context, key remotePriorityBacklog) {
// +1 to make it match only after local backlog tasks at that priority
effectivePriority := effectivePriorityFactor*key.priority + 1
target := tm.partition.TaskQueue().NormalPartition(int(key.partition))
tm.forwardPolls(ctx, key.priority, effectivePriority, priorityBacklogPollForwarder, target)
})
}
func (tm *priTaskMatcher) poll(
ctx context.Context, pollMetadata *pollMetadata, queryOnly bool,
) (*internalTask, error) {

34
service/matching/task.go
View File

@@ -66,19 +66,22 @@ type (
pollerScalingDecision *taskqueuepb.PollerScalingDecision
recycleToken func(*internalTask)
removeFromMatcher atomic.Pointer[func()]
// taskDispatchRevisionNumber represents the revision number used by the task and is
// max(taskDirectiveRevisionNumber, routingConfigRevisionNumber) for the task.
taskDispatchRevisionNumber int64
targetWorkerDeploymentVersion *deploymentspb.WorkerDeploymentVersion
// These fields are for use by matcherData:
// The following fields are for use by priMatcher/matcherData:
waitableMatchResult
forwardCtx context.Context // non-nil for sync match task only
// effectivePriority is initialized from an explicit task priority if present, or the
// default for the task queue. It can also be the special pollForwarderPriority (higher
// than normal priorities) to indicate the poll forwarder. In some other cases (e.g.
// migration) it may be adjusted from the explicit task priority.
// The scale of effectivePriority is 10× the normal scale to allow inserting forwards
// in between priority levels.
effectivePriority priorityKey
// taskDispatchRevisionNumber represents the revision number used by the task and is max(taskDirectiveRevisionNumber, routingConfigRevisionNumber) for the task.
taskDispatchRevisionNumber int64
pollForwarderType pollForwarderType
}
// taskResponse is used to report the result of either a match with a local poller,
@@ -121,20 +124,21 @@ func newInternalTaskForSyncMatch(
redirectInfo = forwardInfo.GetRedirectInfo()
}
return &internalTask{
taskDispatchRevisionNumber: taskDispatchRevisionNumber,
event: &genericTaskInfo{
AllocatedTaskInfo: &persistencespb.AllocatedTaskInfo{
Data: info,
TaskId: syncMatchTaskId,
},
},
forwardInfo: forwardInfo,
source: source,
redirectInfo: redirectInfo,
responseC: make(chan taskResponse, 1),
effectivePriority: priorityKey(info.GetPriority().GetPriorityKey()),
forwardInfo: forwardInfo,
source: source,
redirectInfo: redirectInfo,
responseC: make(chan taskResponse, 1),
taskDispatchRevisionNumber: taskDispatchRevisionNumber,
targetWorkerDeploymentVersion: targetVersion,
effectivePriority: effectivePriorityFactor * priorityKey(info.GetPriority().GetPriorityKey()),
}
}
@@ -148,7 +152,7 @@ func newInternalTaskFromBacklog(
completionFunc: completionFunc,
},
source: enumsspb.TASK_SOURCE_DB_BACKLOG,
effectivePriority: priorityKey(info.GetData().GetPriority().GetPriorityKey()),
effectivePriority: effectivePriorityFactor * priorityKey(info.GetData().GetPriority().GetPriorityKey()),
}
}
@@ -164,7 +168,7 @@ func newInternalQueryTask(
forwardInfo: request.GetForwardInfo(),
responseC: make(chan taskResponse, 1),
source: enumsspb.TASK_SOURCE_HISTORY,
effectivePriority: priorityKey(request.GetPriority().GetPriorityKey()),
effectivePriority: effectivePriorityFactor * priorityKey(request.GetPriority().GetPriorityKey()),
}
}
@@ -191,12 +195,12 @@ func newInternalStartedTask(info *startedTaskInfo) *internalTask {
return &internalTask{started: info}
}
func newPollForwarderTask() *internalTask {
return &internalTask{effectivePriority: pollForwarderPriority}
func newPollForwarderTask(p priorityKey, t pollForwarderType) *internalTask {
return &internalTask{effectivePriority: p, pollForwarderType: t}
}
func (task *internalTask) isPollForwarder() bool {
return task.effectivePriority == pollForwarderPriority
return task.pollForwarderType != notPollForwarder
}
// isQuery returns true if the underlying task is a query task

136
service/matching/task_queue_partition_manager.go
View File

@@ -3,7 +3,9 @@ package matching
import (
"context"
"errors"
"maps"
"math"
"math/bits"
"sync"
"time"
@@ -18,9 +20,11 @@ import (
"go.temporal.io/server/api/matchingservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
"go.temporal.io/server/common/backoff"
"go.temporal.io/server/common/cache"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/future"
"go.temporal.io/server/common/goro"
"go.temporal.io/server/common/headers"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
@@ -30,6 +34,7 @@ import (
serviceerrors "go.temporal.io/server/common/serviceerror"
"go.temporal.io/server/common/softassert"
"go.temporal.io/server/common/tqid"
"go.temporal.io/server/common/util"
"go.temporal.io/server/common/worker_versioning"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
@@ -43,7 +48,6 @@ const (
)
type (
// Represents a single partition of a (user-level) Task Queue in memory state. Under the hood, each Task Queue
// partition is made of one or more DB-level queues. There is always a default DB queue. For
// versioned TQs, there is an additional DB queue for each Build ID.
@@ -70,6 +74,8 @@ type (
// TODO(stephanos): move cache out of partition manager
cache cache.Cache // non-nil for root-partition
goroGroup goro.Group
autoEnableRateLimiter quotas.RateLimiter
fairnessState enumsspb.FairnessState // Set once on initialization and read only after
defaultQueueFuture *future.FutureImpl[physicalTaskQueueManager]
@@ -191,6 +197,7 @@ func (pm *taskQueuePartitionManagerImpl) initialize() (retErr error) {
}
pm.defaultQueueFuture.Set(defaultQ, nil)
defaultQ.Start()
pm.goroGroup.Go(pm.updateEphemeralData)
return nil
}
@@ -210,10 +217,6 @@ func (pm *taskQueuePartitionManagerImpl) Start() {
go pm.initialize()
}
func (pm *taskQueuePartitionManagerImpl) GetRateLimitManager() *rateLimitManager {
return pm.rateLimitManager
}
// Stop does not unload the partition from matching engine. It is intended to be called by matching engine when
// unloading the partition. For stopping and unloading a partition call unloadFromEngine instead.
func (pm *taskQueuePartitionManagerImpl) Stop(unloadCause unloadCause) {
@@ -244,6 +247,12 @@ func (pm *taskQueuePartitionManagerImpl) Stop(unloadCause unloadCause) {
pm.rateLimitManager.Stop()
pm.engine.updateTaskQueuePartitionGauge(pm.Namespace(), pm.partition, -1)
pm.goroGroup.Cancel()
}
func (pm *taskQueuePartitionManagerImpl) GetRateLimitManager() *rateLimitManager {
return pm.rateLimitManager
}
func (pm *taskQueuePartitionManagerImpl) Namespace() *namespace.Namespace {
@@ -921,7 +930,7 @@ func (pm *taskQueuePartitionManagerImpl) Describe(
if dbq == nil {
return nil, errDefaultQueueNotInit
}
unversionedStatsByPriority = dbq.GetStatsByPriority()
unversionedStatsByPriority = dbq.GetStatsByPriority(true)
userData, _, err := pm.GetUserDataManager().GetUserData()
if err != nil {
@@ -955,7 +964,7 @@ func (pm *taskQueuePartitionManagerImpl) Describe(
}
}
versionsInfo := make(map[string]*taskqueuespb.TaskQueueVersionInfoInternal, 0)
versionsInfo := make(map[string]*taskqueuespb.TaskQueueVersionInfoInternal, len(versions))
for v := range versions {
vInfo := &taskqueuespb.TaskQueueVersionInfoInternal{
PhysicalTaskQueueInfo: &taskqueuespb.PhysicalTaskQueueInfo{},
@@ -977,7 +986,7 @@ func (pm *taskQueuePartitionManagerImpl) Describe(
vInfo.PhysicalTaskQueueInfo.Pollers = physicalQueue.GetAllPollerInfo()
}
if reportStats {
physicalStatsByPriority := physicalQueue.GetStatsByPriority()
physicalStatsByPriority := physicalQueue.GetStatsByPriority(true)
// Clone the physical queue's stats by priority so we can adjust (either add, subtract) them based on the
// attribution model defined below.
@@ -1044,6 +1053,117 @@ func (pm *taskQueuePartitionManagerImpl) Describe(
}, nil
}
func (pm *taskQueuePartitionManagerImpl) updateEphemeralData(ctx context.Context) error {
// for now, this only applies to normal workflow task queues, only with new matcher
if pm.partition.Kind() != enumspb.TASK_QUEUE_KIND_NORMAL ||
pm.partition.TaskType() != enumspb.TASK_QUEUE_TYPE_WORKFLOW ||
!pm.config.NewMatcher {
return nil
}
var prevBacklogPriority map[PhysicalTaskQueueVersion]int64
for {
interval := pm.config.EphemeralDataUpdateInterval()
if interval == 0 {
_ = util.InterruptibleSleep(ctx, time.Minute)
continue
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(backoff.Jitter(interval, 0.05)):
prevBacklogPriority = pm.updateEphemeralDataIteration(prevBacklogPriority)
}
}
}
func (pm *taskQueuePartitionManagerImpl) updateEphemeralDataIteration(prevBacklogPriority map[PhysicalTaskQueueVersion]int64) map[PhysicalTaskQueueVersion]int64 {
negligibleAge := pm.config.BacklogNegligibleAge()
backlogPriority := make(map[PhysicalTaskQueueVersion]int64)
setLevels := func(vk PhysicalTaskQueueVersion, vq physicalTaskQueueManager) {
var levels int64
for key, stats := range vq.GetStatsByPriority(false) {
if key < 64 && stats.ApproximateBacklogAge.AsDuration() > negligibleAge {
levels = levels | 1<<key
}
}
if levels != 0 {
backlogPriority[vk] = levels
}
}
dbq := pm.defaultQueue()
if dbq == nil {
return prevBacklogPriority // shouldn't happen, we only start after initialization
}
setLevels(PhysicalTaskQueueVersion{}, dbq)
pm.versionedQueuesLock.RLock()
for vk, vq := range pm.versionedQueues {
if vk.buildId == "" || vk.deploymentSeriesName == "" {
continue // v3 only
}
setLevels(vk, vq)
}
pm.versionedQueuesLock.RUnlock()
if maps.Equal(backlogPriority, prevBacklogPriority) {
return prevBacklogPriority
}
pm.userDataManager.LocalBacklogPriorityChanged(backlogPriority)
return backlogPriority
}
func (pm *taskQueuePartitionManagerImpl) ephemeralDataChanged(data *taskqueuespb.EphemeralData) {
// for now, only sticky partitions act on ephemeral data, normal partitions ignore it.
if pm.partition.Kind() != enumspb.TASK_QUEUE_KIND_STICKY {
return
}
// transpose map to more useful form
updates := make(map[PhysicalTaskQueueVersion]remotePriorityBacklogSet) // version -> {partition id, max level w/backlog}
for _, part := range data.GetPartition() {
for _, verData := range part.GetVersion() {
versionKey := PhysicalTaskQueueVersion{
buildId: verData.Version.GetBuildId(),
deploymentSeriesName: verData.Version.GetDeploymentName(),
}
byVersion := util.GetOrSetMap(updates, versionKey)
levels := uint64(verData.BacklogPriorityLevels)
for pri := bits.TrailingZeros64(levels); pri != 64; pri = bits.TrailingZeros64(levels) {
levels &^= 1 << pri
backlogKey := remotePriorityBacklog{
partition: part.Partition,
priority: priorityKey(pri),
}
byVersion[backlogKey] = struct{}{}
}
}
}
update := func(key PhysicalTaskQueueVersion, pqm physicalTaskQueueManager) {
pqm.UpdateRemotePriorityBacklogs(updates[key])
}
dbq := pm.defaultQueue()
if dbq != nil {
update(PhysicalTaskQueueVersion{}, dbq)
}
pm.versionedQueuesLock.RLock()
defer pm.versionedQueuesLock.RUnlock()
for key, pqm := range pm.versionedQueues {
update(key, pqm)
}
}
func cloneTaskQueueStats(in *taskqueuepb.TaskQueueStats) *taskqueuepb.TaskQueueStats {
if in == nil {
return &taskqueuepb.TaskQueueStats{ApproximateBacklogAge: durationpb.New(0)}

4
service/matching/task_queue_partition_manager_test.go
View File

@@ -1213,6 +1213,10 @@ func (m *mockUserDataManager) CheckTaskQueueUserDataPropagation(ctx context.Cont
panic("unused")
}
func (m *mockUserDataManager) LocalBacklogPriorityChanged(map[PhysicalTaskQueueVersion]int64) {
panic("unused")
}
func (m *mockUserDataManager) updateVersioningData(data *persistencespb.VersioningData) {
m.Lock()
defer m.Unlock()

183
service/matching/user_data_manager.go
View File

@@ -3,6 +3,7 @@ package matching
import (
"context"
"errors"
"slices"
"strings"
"sync"
"sync/atomic"
@@ -13,6 +14,7 @@ import (
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/api/matchingservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/backoff"
"go.temporal.io/server/common/clock/hybrid_logical_clock"
@@ -51,6 +53,7 @@ type (
// Handles the maybe-long-poll GetUserData RPC.
HandleGetUserDataRequest(ctx context.Context, req *matchingservice.GetTaskQueueUserDataRequest) (*matchingservice.GetTaskQueueUserDataResponse, error)
CheckTaskQueueUserDataPropagation(context.Context, int64, int, int) error
LocalBacklogPriorityChanged(map[PhysicalTaskQueueVersion]int64)
}
UserDataUpdateOptions struct {
@@ -64,24 +67,33 @@ type (
// UserDataUpdateFunc accepts the current user data for a task queue and returns the updated user data, a boolean
// indicating whether this data should be replicated, and an error.
// Extra care should be taken to avoid mutating the current user data to avoid keeping uncommitted data in memory.
UserDataUpdateFunc func(*persistencespb.TaskQueueUserData) (*persistencespb.TaskQueueUserData, bool, error)
UserDataOnChangeFunc func(to *persistencespb.VersionedTaskQueueUserData)
UserDataUpdateFunc func(*persistencespb.TaskQueueUserData) (*persistencespb.TaskQueueUserData, bool, error)
UserDataOnChangeFunc func(to *persistencespb.VersionedTaskQueueUserData)
EphemeralDataChangeFunc func(*taskqueuespb.EphemeralData)
// userDataManager is responsible for fetching and keeping user data up-to-date in-memory
// for a given TQ partition.
// userDataManager is responsible for fetching and keeping user data and ephemeral data
// up-to-date in-memory for a given TQ partition.
//
// If a partition is the root of a normal (non-sticky) task queue with workflow type,
// we say the partition "owns" user data for its task queue. All reads and writes
// to/from the persistence layer passes through userDataManager of the owning partition.
// All other partitions long-poll the latest user data from the owning partition.
// If a partition is the root of a normal (non-sticky) task queue with workflow type, we say the
// partition "owns" user data for its task queue. All reads and writes to/from the persistence
// layer passes through userDataManager of the owning partition. All other partitions long-poll
// the latest user data from their parent in the forwarding tree structure, and the root
// partitions of other queue types poll from the root workflow partition.
//
// Ephemeral data is a little different: First, it's separate for different task queue types.
// Second, we're currently more lax about aggregating it, since the purposes it's used for don't
// require full perfect information. Instead of passing writes to the root, we currently just
// let each partition maintain its own data and merge with data coming from the root. Eventually
// we'll switch to passing writes to the root or aggregating by the tree structure.
userDataManagerImpl struct {
lock sync.Mutex
onFatalErr func(unloadCause)
onUserDataChanged UserDataOnChangeFunc // if set, call this in new goroutine when user data changes
partition tqid.Partition
userData *persistencespb.VersionedTaskQueueUserData
userDataChanged chan struct{}
userDataState userDataState
lock sync.Mutex
onFatalErr func(unloadCause)
onUserDataChanged UserDataOnChangeFunc // if set, call this in new goroutine when user data changes
onEphemeralDataChanged EphemeralDataChangeFunc // if set, call this in new goroutine when ephemeral data changes
partition tqid.Partition
userData *persistencespb.VersionedTaskQueueUserData
userDataChanged chan struct{}
userDataState userDataState
// only set if this partition owns user data of its task queue
store persistence.TaskManager
config *taskQueueConfig
@@ -92,6 +104,11 @@ type (
// userDataReady is fulfilled once versioning data is fetched from the root partition. If this TQ is
// the root partition, it is fulfilled as soon as it is fetched from db.
userDataReady *future.FutureImpl[struct{}]
myEphemeralData *taskqueuespb.VersionedEphemeralData
incomingEphemeralData *taskqueuespb.VersionedEphemeralData
mergedEphemeralData *taskqueuespb.VersionedEphemeralData
ephemeralDataChanged chan struct{}
}
userDataState int
@@ -113,21 +130,24 @@ func newUserDataManager(
matchingClient matchingservice.MatchingServiceClient,
onFatalErr func(unloadCause),
onUserDataChanged UserDataOnChangeFunc,
onEphemeralDataChanged EphemeralDataChangeFunc,
partition tqid.Partition,
config *taskQueueConfig,
logger log.Logger,
registry namespace.Registry,
) *userDataManagerImpl {
m := &userDataManagerImpl{
onFatalErr: onFatalErr,
onUserDataChanged: onUserDataChanged,
partition: partition,
userDataChanged: make(chan struct{}),
config: config,
namespaceRegistry: registry,
logger: logger,
matchingClient: matchingClient,
userDataReady: future.NewFuture[struct{}](),
onFatalErr: onFatalErr,
onUserDataChanged: onUserDataChanged,
onEphemeralDataChanged: onEphemeralDataChanged,
partition: partition,
userDataChanged: make(chan struct{}),
config: config,
namespaceRegistry: registry,
logger: logger,
matchingClient: matchingClient,
userDataReady: future.NewFuture[struct{}](),
ephemeralDataChanged: make(chan struct{}),
}
if partition.IsRoot() && partition.TaskType() == enumspb.TASK_QUEUE_TYPE_WORKFLOW {
@@ -287,11 +307,12 @@ func (m *userDataManagerImpl) fetchUserData(ctx context.Context) error {
defer cancel()
res, err := m.matchingClient.GetTaskQueueUserData(callCtx, &matchingservice.GetTaskQueueUserDataRequest{
NamespaceId: m.partition.NamespaceId(),
TaskQueue: fetchSource.RpcName(),
TaskQueueType: fetchSource.TaskType(),
LastKnownUserDataVersion: knownUserData.GetVersion(),
WaitNewData: hasFetchedUserData,
NamespaceId: m.partition.NamespaceId(),
TaskQueue: fetchSource.RpcName(),
TaskQueueType: fetchSource.TaskType(),
LastKnownUserDataVersion: knownUserData.GetVersion(),
LastKnownEphemeralDataVersion: m.getIncomingEphemeralDataVersion(),
WaitNewData: hasFetchedUserData,
})
if err != nil {
// don't log on context canceled, produces too much log spam at shutdown
@@ -319,6 +340,9 @@ func (m *userDataManagerImpl) fetchUserData(ctx context.Context) error {
} else {
m.logger.Debug("fetched user data from parent, no change")
}
if res.GetEphemeralData() != nil {
m.gotIncomingEphemeralData(res.EphemeralData)
}
hasFetchedUserData = true
m.setUserDataState(userDataEnabled, nil)
return nil
@@ -549,6 +573,7 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
if lastVersion < 0 {
return nil, serviceerror.NewInvalidArgument("last_known_user_data_version must not be negative")
}
lastEphVersion := req.GetLastKnownEphemeralDataVersion()
if req.WaitNewData {
var cancel context.CancelFunc
@@ -558,6 +583,7 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
for {
userData, userDataChanged, err := m.GetUserData()
ephData, ephDataChanged := m.getMergedEphemeralData()
if errors.Is(err, errTaskQueueClosed) {
// If we're closing, return a success with no data, as if the request expired. We shouldn't
// close due to idleness (because of the MarkAlive above), so we're probably closing due to a
@@ -567,15 +593,24 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
} else if err != nil {
return nil, err
}
if userData.GetVersion() > lastVersion {
newUserData := userData.GetVersion() > lastVersion
newEphData := ephData.GetVersion() > lastEphVersion
if newUserData || newEphData {
m.logger.Info("returning user data",
tag.NewBoolTag("long-poll", req.WaitNewData),
tag.NewInt64("request-known-version", lastVersion),
tag.UserDataVersion(userData.GetVersion()),
tag.NewInt64("request-eph-data-version", lastEphVersion),
tag.NewInt64("eph-data-version", ephData.GetVersion()),
)
return &matchingservice.GetTaskQueueUserDataResponse{
UserData: userData,
}, nil
var res matchingservice.GetTaskQueueUserDataResponse
if newUserData {
res.UserData = userData
}
if newEphData {
res.EphemeralData = ephData
}
return &res, nil
} else if userData != nil && userData.Version < lastVersion && m.store != nil {
// When m.store == nil it means this is a non-owner partition, so it is possible
// for the requested version to be greater than the known version if there are
@@ -590,6 +625,9 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
)
return nil, errRequestedVersionTooLarge
}
// For ephemeral data: A similar situation could happen when a partition moves, we might
// have older data than the child. Don't return a error in that case, just wait until we
// have newer data. Note that "version" is a timestamp.
if !req.WaitNewData {
m.logger.Debug("returning empty user data (no data or no change)")
@@ -606,6 +644,8 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
return &matchingservice.GetTaskQueueUserDataResponse{}, nil
case <-userDataChanged:
m.logger.Debug("user data changed while blocked in long poll")
case <-ephDataChanged:
m.logger.Debug("ephemeral data changed while blocked in long poll")
}
}
}
@@ -681,6 +721,83 @@ func (m *userDataManagerImpl) CheckTaskQueueUserDataPropagation(
}
}
func (m *userDataManagerImpl) LocalBacklogPriorityChanged(backlogPriority map[PhysicalTaskQueueVersion]int64) {
// TODO: later, we'll send this data to the root to propagate instead of just keeping it
// locally and merging.
normal, ok := m.partition.(*tqid.NormalPartition)
if !ok {
return
}
byVersion := make([]*taskqueuespb.EphemeralData_ByVersion, 0, len(backlogPriority))
for ver, levels := range backlogPriority {
byVersion = append(byVersion, &taskqueuespb.EphemeralData_ByVersion{
Version: ver.WorkerDeploymentVersionS(),
BacklogPriorityLevels: levels,
})
}
newEph := &taskqueuespb.VersionedEphemeralData{
Data: &taskqueuespb.EphemeralData{
Partition: []*taskqueuespb.EphemeralData_ByPartition{
&taskqueuespb.EphemeralData_ByPartition{
Partition: int32(normal.PartitionId()),
Version: byVersion,
},
},
},
Version: time.Now().UnixNano(),
}
m.lock.Lock()
defer m.lock.Unlock()
m.myEphemeralData = newEph
m.mergeEphemeralDataLocked()
}
func (m *userDataManagerImpl) gotIncomingEphemeralData(eph *taskqueuespb.VersionedEphemeralData) {
m.lock.Lock()
defer m.lock.Unlock()
m.incomingEphemeralData = eph
m.mergeEphemeralDataLocked()
}
func (m *userDataManagerImpl) getMergedEphemeralData() (*taskqueuespb.VersionedEphemeralData, chan struct{}) {
m.lock.Lock()
defer m.lock.Unlock()
return m.mergedEphemeralData, m.ephemeralDataChanged
}
func (m *userDataManagerImpl) getIncomingEphemeralDataVersion() int64 {
m.lock.Lock()
defer m.lock.Unlock()
return m.incomingEphemeralData.GetVersion()
}
func (m *userDataManagerImpl) mergeEphemeralDataLocked() {
m.mergedEphemeralData = &taskqueuespb.VersionedEphemeralData{
Data: &taskqueuespb.EphemeralData{
// data is already separated by partition, so we can just concatenate
Partition: slices.Concat(
m.incomingEphemeralData.GetData().GetPartition(),
m.myEphemeralData.GetData().GetPartition(),
),
},
Version: time.Now().UnixNano(),
}
close(m.ephemeralDataChanged)
m.ephemeralDataChanged = make(chan struct{})
if m.onEphemeralDataChanged != nil {
go m.onEphemeralDataChanged(m.mergedEphemeralData.Data)
}
}
func (m *userDataManagerImpl) setUserDataForNonOwningPartition(userData *persistencespb.VersionedTaskQueueUserData) {
m.lock.Lock()
defer m.lock.Unlock()

12
service/matching/user_data_manager_test.go
View File

@@ -69,7 +69,17 @@ func createUserDataManager(
onFatalErr = func(unloadCause) { t.Fatal("user data manager called onFatalErr") }
}
return newUserDataManager(tm, testOpts.matchingClientMock, onFatalErr, nil, testOpts.dbq.Partition(), newTaskQueueConfig(testOpts.dbq.Partition().TaskQueue(), testOpts.config, ns), logger, mockNamespaceCache)
return newUserDataManager(
tm,
testOpts.matchingClientMock,
onFatalErr,
nil,
nil,
testOpts.dbq.Partition(),
newTaskQueueConfig(testOpts.dbq.Partition().TaskQueue(), testOpts.config, ns),
logger,
mockNamespaceCache,
)
}
func TestUserData_LoadOnInit(t *testing.T) {

171
tests/priority_fairness_test.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"math/rand"
"strings"
"testing"
"time"
@@ -49,7 +50,7 @@ func (s *PrioritySuite) SetupSuite() {
s.FunctionalTestBase.SetupSuiteWithCluster(testcore.WithDynamicConfigOverrides(dynamicConfigOverrides))
}
func (s *PrioritySuite) TestPriority_Activity_Basic() {
func (s *PrioritySuite) TestActivity_Basic() {
const N = 100
const Levels = 5
@@ -225,6 +226,164 @@ func (s *PrioritySuite) TestSubqueue_Migration() {
}
}
func (s *PrioritySuite) TestStickyInteraction_SinglePartition() {
const N = 10
tv := testvars.New(s.T())
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
// one partition for now:
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1)
s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1)
// set these to make the mechanism react faster for the test:
shortTime := 20 * time.Millisecond
s.OverrideDynamicConfig(dynamicconfig.MatchingBacklogNegligibleAge, shortTime)
s.OverrideDynamicConfig(dynamicconfig.MatchingEphemeralDataUpdateInterval, shortTime)
describeSticky := func() (*adminservice.DescribeTaskQueuePartitionResponse, error) {
return s.AdminClient().DescribeTaskQueuePartition(ctx, &adminservice.DescribeTaskQueuePartitionRequest{
Namespace: s.Namespace().String(),
TaskQueuePartition: &taskqueuespb.TaskQueuePartition{
TaskQueue: tv.TaskQueue().Name,
TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW,
PartitionId: &taskqueuespb.TaskQueuePartition_StickyName{StickyName: tv.StickyTaskQueue().Name},
},
BuildIds: &taskqueuepb.TaskQueueVersionSelection{Unversioned: true},
})
}
// poll sticky queue once, otherwise it won't be used
s.T().Log("polling sticky to load")
stickyPoller := s.TaskPoller().PollWorkflowTask(&workflowservice.PollWorkflowTaskQueueRequest{
TaskQueue: &taskqueuepb.TaskQueue{
Name: tv.StickyTaskQueue().Name,
Kind: enumspb.TASK_QUEUE_KIND_STICKY,
NormalName: tv.TaskQueue().Name,
},
})
stickyCtx, stickyCancel := context.WithCancel(ctx)
go stickyPoller.HandleTask(tv, taskpoller.DrainWorkflowTask, taskpoller.WithContext(stickyCtx)) // nolint:errcheck
// wait for poll to reach matching service and load the queue
s.EventuallyWithT(func(c *assert.CollectT) {
res, err := describeSticky()
require.NoError(c, err)
require.NotEmpty(c, res.VersionsInfoInternal[""].GetPhysicalTaskQueueInfo().GetPollers())
}, 5*time.Second, 10*time.Millisecond)
// cancel as soon as it's registered
s.T().Log("canceling sticky poll")
stickyCancel()
// wait for cancel to propagate
time.Sleep(100 * time.Millisecond) // nolint:forbidigo // there's no way to wait for this
// create some wfts at default priority
s.T().Log("creating wfts on normal")
for wfidx := range N {
_, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowId: fmt.Sprintf("wf%d", wfidx),
WorkflowType: tv.WorkflowType(),
TaskQueue: tv.TaskQueue(),
})
s.NoError(err)
}
// process initial tasks on normal queue
s.T().Log("processing wfts")
for range N {
_, err := s.TaskPoller().PollAndHandleWorkflowTask(
tv,
func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
return &workflowservice.RespondWorkflowTaskCompletedRequest{
Commands: []*commandpb.Command{&commandpb.Command{
CommandType: enumspb.COMMAND_TYPE_START_TIMER,
Attributes: &commandpb.Command_StartTimerCommandAttributes{
StartTimerCommandAttributes: &commandpb.StartTimerCommandAttributes{
TimerId: uuid.NewString(),
StartToFireTimeout: durationpb.New(time.Millisecond),
},
},
}},
StickyAttributes: &taskqueuepb.StickyExecutionAttributes{
WorkerTaskQueue: tv.StickyTaskQueue(),
ScheduleToStartTimeout: durationpb.New(10 * time.Second),
},
}, nil
},
taskpoller.WithContext(ctx),
)
s.NoError(err)
}
// after 1 millisecond, we should have N tasks queued on the sticky queue at normal priority
s.T().Log("checking sticky backlog")
s.EventuallyWithT(func(c *assert.CollectT) {
res, err := describeSticky()
require.NoError(c, err)
require.EqualValues(c, N, res.VersionsInfoInternal[""].PhysicalTaskQueueInfo.TaskQueueStats.ApproximateBacklogCount)
}, 5*time.Second, 10*time.Millisecond)
// create N more workflows at high priority and N at lower
s.T().Log("creating high/low wfts on normal")
for wfidx := range N {
_, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowId: fmt.Sprintf("highpri%d", wfidx),
WorkflowType: tv.WorkflowType(),
TaskQueue: tv.TaskQueue(),
Priority: &commonpb.Priority{PriorityKey: 1},
})
s.NoError(err)
_, err = s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowId: fmt.Sprintf("lowpri%d", wfidx),
WorkflowType: tv.WorkflowType(),
TaskQueue: tv.TaskQueue(),
Priority: &commonpb.Priority{PriorityKey: 5},
})
s.NoError(err)
}
// the ephemeral data mechanism is asynchronous, so wait a little while for it to kick in.
// there's no way to check if this is done yet without actually polling, which would mess
// up the results, so just sleep.
time.Sleep(3 * shortTime) // nolint:forbidigo
// now poll the sticky queue. we should get the high priority tasks from the normal queue
// then the normal-priority from sticky, then the low priority from normal.
s.T().Log("polling sticky")
var receivedOrder []string
for range 3 * N {
_, _ = stickyPoller.HandleTask(
tv,
func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
wfid := task.WorkflowExecution.WorkflowId
s.T().Log("task for wf", wfid)
receivedOrder = append(receivedOrder, wfid)
return nil, nil
},
taskpoller.WithContext(ctx),
)
}
// validate the order
var priorityOrder []int
for _, wfid := range receivedOrder {
if strings.HasPrefix(wfid, "highpri") {
priorityOrder = append(priorityOrder, 1)
} else if strings.HasPrefix(wfid, "lowpri") {
priorityOrder = append(priorityOrder, 3)
} else {
priorityOrder = append(priorityOrder, 2) // default priority (wf*)
}
}
w := wrongorderness(priorityOrder)
s.T().Log("wrongorderness:", w)
s.Less(w, 0.1, "tasks should mostly arrive in priority order (high, default, low)")
}
func wrongorderness(vs []int) float64 {
l := len(vs)
wrong := 0
@@ -341,7 +500,7 @@ func (s *FairnessSuite) TriggerAutoEnable(tv *testvars.TestVars) {
cancel()
}
func (s *FairnessSuite) TestFairness_Activity_Basic() {
func (s *FairnessSuite) Test_Activity_Basic() {
const Workflows = 15
const Tasks = 15
const Keys = 10
@@ -632,22 +791,22 @@ func (s *FairnessSuite) countTasksByDrainingActive(ctx context.Context, tv *test
return
}
func (s *FairnessSuite) TestFairness_Migration_FromClassic() {
func (s *FairnessSuite) TestMigration_FromClassic() {
// classic->fair, fair->pri. fair metadata will be created on transition.
s.testMigration(false, false)
}
func (s *FairnessSuite) TestFairness_Migration_FromPri() {
func (s *FairnessSuite) TestMigration_FromPri() {
// pri->fair, fair->pri. fair metadata will be created before transition.
s.testMigration(true, false)
}
func (s *FairnessSuite) TestFairness_Migration_FromFair() {
func (s *FairnessSuite) TestMigration_FromFair() {
// fair->pri, pri->fair. fair metadata will be created first.
s.testMigration(true, true)
}
func (s *FairnessSuite) TestFairness_UpdateWorkflowExecutionOptions_InvalidatesPendingTask() {
func (s *FairnessSuite) TestUpdateWorkflowExecutionOptions_InvalidatesPendingTask() {
if s.doAutoEnable {
s.T().Skip("flaky with autoenable")
}