Method processPipeline

osscluster.go:1562–1626

(ctx context.Context, cmds []Cmder)

}

func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error {
	// Only call time.Now() if pipeline operation duration callback is set to avoid overhead
	var operationStart time.Time
	pipelineOpDurationCallback := otel.GetPipelineOperationDurationCallback()
	if pipelineOpDurationCallback != nil {
		operationStart = time.Now()
	}
	totalAttempts := 0

	cmdsMap := newCmdsMap()

	if err := c.mapCmdsByNode(ctx, cmdsMap, cmds); err != nil {
		setCmdsErr(cmds, err)
		if pipelineOpDurationCallback != nil {
			operationDuration := time.Since(operationStart)
			pipelineOpDurationCallback(ctx, operationDuration, "PIPELINE", len(cmds), 1, err, nil, 0)
		}
		return err
	}

	var lastErr error
	for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
		totalAttempts++
		if attempt > 0 {
			if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
				setCmdsErr(cmds, err)
				if pipelineOpDurationCallback != nil {
					operationDuration := time.Since(operationStart)
					pipelineOpDurationCallback(ctx, operationDuration, "PIPELINE", len(cmds), totalAttempts, err, nil, 0)
				}
				return err
			}
		}

		failedCmds := newCmdsMap()
		var wg sync.WaitGroup

		for node, cmds := range cmdsMap.m {
			wg.Add(1)
			go func(node *clusterNode, cmds []Cmder) {
				defer wg.Done()
				c.processPipelineNode(ctx, node, cmds, failedCmds)
			}(node, cmds)
		}

		wg.Wait()
		if len(failedCmds.m) == 0 {
			break
		}
		cmdsMap = failedCmds
		lastErr = cmdsFirstErr(cmds)
	}

	// Record pipeline operation duration
	if pipelineOpDurationCallback != nil {
		operationDuration := time.Since(operationStart)
		finalErr := cmdsFirstErr(cmds)
		if finalErr == nil {