README
¶
Matching Client
This package provides the client for communicating with the Matching service. It handles two key concerns: partition selection and routing.
Partition Selection (Load Balancing)
The LoadBalancer distributes add/poll API calls across available task queue partitions.
Write path: Partitions are selected uniformly at random from the configured write partition count.
Read path: Partitions are selected to balance poller counts across partitions. The load balancer tracks outstanding polls per partition and sends new polls to the partition with the fewest active polls.
The number of partitions is controlled by one of two mechanisms:
- Managed partition scaling
- Dynamic config (
matching.numTaskqueueWritePartitions,matching.numTaskqueueReadPartitions)
There are test hooks to force specific partition selection for testing.
Routing
Routing determines which matching node owns a given task queue partition. All clients (frontend, history) independently perform this computation using consistent hashing via ringpop.
Basic Routing
Each partition has a routing key of the form:
namespace_id:queue_name:task_type
This key is hashed with the consistent hashing algorithm to find the owning node.
Spread Routing
With basic routing, partitions of the same queue are placed independently, which can cause multiple partitions to land on the same node, creating hot spots.
Spread routing groups partitions into batches and uses LookupN to ensure
partitions within a batch are assigned to different nodes if possible.
The batch size is controlled by dynamic config matching.spreadRoutingBatchSize,
default zero (i.e. use basic routing).
Algorithm:
- Compute batch number:
batch = partition_id / batch_size - Compute index within batch:
index = partition_id % batch_size - Generate routing key with batch number (batch 0 omits the batch number for backward compatibility):
namespace_id:queue_name:batch_number:task_type - Call
LookupN(key, index+1)and take the host at positionindex
For example, with batch size 8 and partition 25:
- Batch 3 (floor(25/8)), index 1 (25%8)
- Key:
namespace_id:queue_name:3:task_type - Call
LookupN(key, 2), take host at index 1
If fewer hosts are available than the batch size, wrap around to spread among available hosts.
Tradeoff: Larger batch sizes provide better spread but cause more partition movement when membership changes.
Changes
Changing partition count dynamically is generally safe and doesn't cause partitions to move. The caveat is that when reducing, write partitions has to be reduced first, and then the extra partitions have to be empty before reducing read partitions.
Changing batch size will cause most partitions to move between nodes.
To avoid moving lots of partitions simultaneously on a live cluster,
spread routing can be rolled out gradually (partition by partition)
using wall-clock-synchronized changes. See the the GradualChange mechanism.
Interface
The Route(partition) method on the client computes the owning node address for any partition.
This is used internally by the grpc client, and can be used by other code to
determine the owner for other purposes (e.g. matching engine knowing when to
unload non-owned partitions).
Managed partition scaling
If managed partition scaling is in use, the server communicates the current read/write partition counts to the client (in a grpc trailer). The client also communicates it back to the server (in a grpc header) to ensure the client isn't using stale counts. The client maintains a cache of the current counts for each task queue.
Documentation
¶
Index ¶
- Constants
- func NewClient(timeout time.Duration, longPollTimeout time.Duration, ...) matchingservice.MatchingServiceClient
- func NewMetricClient(client matchingservice.MatchingServiceClient, metricsHandler metrics.Handler, ...) matchingservice.MatchingServiceClient
- func NewRetryableClient(client matchingservice.MatchingServiceClient, ...) matchingservice.MatchingServiceClient
- func ParseEstimatedTasksAllPartitions(ctx context.Context) int
- type LoadBalancer
- type PartitionCounts
- type RoutingClient
Constants ¶
const ( // DefaultTimeout is the max timeout for regular calls DefaultTimeout = time.Minute * debug.TimeoutMultiplier // DefaultLongPollTimeout is the max timeout for long poll calls DefaultLongPollTimeout = time.Minute * 5 * debug.TimeoutMultiplier )
Variables ¶
This section is empty.
Functions ¶
func NewClient ¶
func NewClient( timeout time.Duration, longPollTimeout time.Duration, clients common.ClientCache, metricsHandler metrics.Handler, logger log.Logger, lb LoadBalancer, spreadRouting dynamicconfig.TypedPropertyFn[dynamicconfig.GradualChange[int]], resolver membership.ServiceResolver, connectionCloseDelay dynamicconfig.DurationPropertyFn, ) matchingservice.MatchingServiceClient
NewClient creates a new matching service gRPC client
func NewMetricClient ¶
func NewMetricClient( client matchingservice.MatchingServiceClient, metricsHandler metrics.Handler, logger log.Logger, throttledLogger log.Logger, ) matchingservice.MatchingServiceClient
NewMetricClient creates a new instance of matchingservice.MatchingServiceClient that emits metrics
func NewRetryableClient ¶
func NewRetryableClient( client matchingservice.MatchingServiceClient, policy, pollPolicy backoff.RetryPolicy, isRetryable backoff.IsRetryable, ) matchingservice.MatchingServiceClient
NewRetryableClient creates a new instance of matchingservice.MatchingServiceClient with retry policy
Types ¶
type LoadBalancer ¶
type LoadBalancer interface {
// PickWritePartition returns the task queue partition for adding an
// activity or workflow task and the estimated number of tasks added
// across all partitions per root task. The input is the name of the
// original task queue (with no partition info).
PickWritePartition(
taskQueue *tqid.TaskQueue,
pc PartitionCounts,
) (*tqid.NormalPartition, int)
// PickReadPartition returns the task queue partition to send a poller to.
// Input is name of the original task queue as specified by caller.
PickReadPartition(
taskQueue *tqid.TaskQueue,
pc PartitionCounts,
) *pollToken
}
LoadBalancer is the interface for implementers of component that distributes add/poll api calls across available task queue partitions when possible
func NewLoadBalancer ¶
func NewLoadBalancer( namespaceIDToName func(id namespace.ID) (namespace.Name, error), dc *dynamicconfig.Collection, testHooks testhooks.TestHooks, ) LoadBalancer
NewLoadBalancer returns an instance of matching load balancer that can help distribute api calls across task queue partitions
type PartitionCounts ¶
type PartitionCounts struct {
Read, Write int32
BacklogCap number.Compact8
BacklogCount []number.Compact8
}
PartitionCounts is a smaller version of taskqueuespb.ClientPartitionCounts that we can more easily pass around and put in a map.
func ParsePartitionCountsFromIncomingContext ¶
func ParsePartitionCountsFromIncomingContext(ctx context.Context) (PartitionCounts, error)
func (PartitionCounts) Equal ¶
func (pc PartitionCounts) Equal(other PartitionCounts) bool
func (PartitionCounts) SetTrailer ¶
func (pc PartitionCounts) SetTrailer(ctx context.Context) error
func (PartitionCounts) Valid ¶
func (pc PartitionCounts) Valid() bool
Source Files