Method executeMultiSlot

osscluster_router.go:148–177

executeMultiSlot executes commands across multiple slots concurrently

(ctx context.Context, cmd Cmder, slotMap map[int][]string, keyOrder []string, policy *routing.CommandPolicy, firstKeyPos int)

// executeMultiSlot executes commands across multiple slots concurrently
func (c *ClusterClient) executeMultiSlot(ctx context.Context, cmd Cmder, slotMap map[int][]string, keyOrder []string, policy *routing.CommandPolicy, firstKeyPos int) error {
	results := make(chan slotResult, len(slotMap))
	var wg sync.WaitGroup

	// Execute on each slot concurrently
	for slot, keys := range slotMap {
		wg.Add(1)
		go func(slot int, keys []string) {
			defer wg.Done()

			node, err := c.cmdNodeWithShardPicker(ctx, cmd.Name(), slot, c.opt.ShardPicker)
			if err != nil {
				results <- slotResult{nil, keys, err}
				return
			}

			// Create a command for this specific slot's keys
			subCmd := c.createSlotSpecificCommand(ctx, cmd, keys, firstKeyPos)
			err = node.Client.Process(ctx, subCmd)
			results <- slotResult{subCmd, keys, err}
		}(slot, keys)
	}

	go func() {
		wg.Wait()
		close(results)
	}()

	return c.aggregateMultiSlotResults(ctx, cmd, results, keyOrder, policy)
}

// createSlotSpecificCommand creates a new command for a specific slot's keys.
// firstKeyPos is passed in from the caller (computed once in executeMultiShard)