Method processTxPipeline

osscluster.go:1835–1935

(ctx context.Context, cmds []Cmder)

}

func (c *ClusterClient) processTxPipeline(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

	// Trim multi .. exec.
	cmds = cmds[1 : len(cmds)-1]

	if len(cmds) == 0 {
		return nil
	}

	state, err := c.state.Get(ctx)
	if err != nil {
		setCmdsErr(cmds, err)
		if pipelineOpDurationCallback != nil {
			operationDuration := time.Since(operationStart)
			pipelineOpDurationCallback(ctx, operationDuration, "MULTI", len(cmds), 1, err, nil, 0)
		}
		return err
	}

	keyedCmdsBySlot := c.slottedKeyedCommands(ctx, cmds)
	slot := -1
	switch len(keyedCmdsBySlot) {
	case 0:
		slot = hashtag.RandomSlot()
	case 1:
		for sl := range keyedCmdsBySlot {
			slot = sl
			break
		}
	default:
		// TxPipeline does not support cross slot transaction.
		setCmdsErr(cmds, ErrCrossSlot)
		if pipelineOpDurationCallback != nil {
			operationDuration := time.Since(operationStart)
			pipelineOpDurationCallback(ctx, operationDuration, "MULTI", len(cmds), 1, ErrCrossSlot, nil, 0)
		}
		return ErrCrossSlot
	}

	node, err := state.slotMasterNode(slot)
	if err != nil {
		setCmdsErr(cmds, err)
		if pipelineOpDurationCallback != nil {
			operationDuration := time.Since(operationStart)
			pipelineOpDurationCallback(ctx, operationDuration, "MULTI", len(cmds), 1, err, nil, 0)
		}
		return err
	}

	var lastErr error
	cmdsMap := map[*clusterNode][]Cmder{node: cmds}