// newTwoPhaseCommitter creates a twoPhaseCommitter.
358654
func newTwoPhaseCommitter(txn *tikvTxn, connID uint64) (*twoPhaseCommitter, error) {
	var (
		keys    [][]byte
		size    int
		putCnt  int
		delCnt  int
		lockCnt int
	)
	mutations := make(map[string]*mutationEx)
	err := txn.us.WalkBuffer(func(k kv.Key, v []byte) error {
		if len(v) > 0 {
			op := pb.Op_Put
			if c := txn.us.LookupConditionPair(k); c != nil && c.ShouldNotExist() {
				op = pb.Op_Insert
			}
			mutations[string(k)] = &mutationEx{
				Mutation: pb.Mutation{
					Op:    op,
					Key:   k,
					Value: v,
				},
			}
			putCnt++
		} else {
			mutations[string(k)] = &mutationEx{
				Mutation: pb.Mutation{
					Op:  pb.Op_Del,
					Key: k,
				},
			}
			delCnt++
		}
		keys = append(keys, k)
		entrySize := len(k) + len(v)
		if entrySize > kv.TxnEntrySizeLimit {
			return kv.ErrEntryTooLarge.GenWithStackByArgs(kv.TxnEntrySizeLimit, entrySize)
		}
		size += entrySize
		return nil
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	for _, lockKey := range txn.lockKeys {
		if _, ok := mutations[string(lockKey)]; !ok {
			mutations[string(lockKey)] = &mutationEx{
				Mutation: pb.Mutation{
					Op:  pb.Op_Lock,
					Key: lockKey,
				},
			}
			lockCnt++
			keys = append(keys, lockKey)
			size += len(lockKey)
		}
	}
	if len(keys) == 0 {
		return nil, nil
	}

	for _, pair := range txn.assertions {
		mutation, ok := mutations[string(pair.key)]
		if !ok {
			logutil.Logger(context.Background()).Error("ASSERTION FAIL!!! assertion exists but no mutation?",
				zap.Stringer("assertion", pair))
			continue
		}
		// Only apply the first assertion!
		if mutation.asserted {
			continue
		}
		switch pair.assertion {
		case kv.Exist:
			mutation.Assertion = pb.Assertion_Exist
		case kv.NotExist:
			mutation.Assertion = pb.Assertion_NotExist
		default:
			mutation.Assertion = pb.Assertion_None
		}
		mutation.asserted = true
	}

	entrylimit := atomic.LoadUint64(&kv.TxnEntryCountLimit)
	if len(keys) > int(entrylimit) || size > kv.TxnTotalSizeLimit {
		return nil, kv.ErrTxnTooLarge
	}
	const logEntryCount = 10000
	const logSize = 4 * 1024 * 1024 // 4MB
	if len(keys) > logEntryCount || size > logSize {
		tableID := tablecodec.DecodeTableID(keys[0])
		logutil.Logger(context.Background()).Info("[BIG_TXN]",
			zap.Uint64("con", connID),
			zap.Int64("table ID", tableID),
			zap.Int("size", size),
			zap.Int("keys", len(keys)),
			zap.Int("puts", putCnt),
			zap.Int("dels", delCnt),
			zap.Int("locks", lockCnt),
			zap.Uint64("txnStartTS", txn.startTS))
	}

	// Convert from sec to ms
	maxTxnTimeUse := uint64(config.GetGlobalConfig().TiKVClient.MaxTxnTimeUse) * 1000

	// Sanity check for startTS.
	if txn.StartTS() == math.MaxUint64 {
		err = errors.Errorf("try to commit with invalid txnStartTS: %d", txn.StartTS())
		logutil.Logger(context.Background()).Error("commit failed",
			zap.Uint64("conn", connID),
			zap.Error(err))
		return nil, errors.Trace(err)
	}

	commitDetail := &execdetails.CommitDetails{WriteSize: size, WriteKeys: len(keys)}
	metrics.TiKVTxnWriteKVCountHistogram.Observe(float64(commitDetail.WriteKeys))
	metrics.TiKVTxnWriteSizeHistogram.Observe(float64(commitDetail.WriteSize))
	return &twoPhaseCommitter{
		store:         txn.store,
		txn:           txn,
		startTS:       txn.StartTS(),
		keys:          keys,
		mutations:     mutations,
		lockTTL:       txnLockTTL(txn.startTime, size),
		priority:      getTxnPriority(txn),
		syncLog:       getTxnSyncLog(txn),
		connID:        connID,
		maxTxnTimeUse: maxTxnTimeUse,
		detail:        commitDetail,
	}, nil
}
200088
func (s *Build) SetCurrentPhase(v string) *Build {
	s.CurrentPhase = &v
	return s
}
444324
func NewTopic(name string, config TopicConfig, client *Client) *Topic {
	return &Topic{
		handle: C.rd_kafka_topic_new(client.handle, C.CString(name), config.handle),
		client: client,
		name:   name,
	}
}
846169
func NewTimer(sec time.Duration, worker *worker.Worker) (*Timer, error) {
	return &Timer{
		ticker: time.NewTicker(sec * time.Second),
		worker: worker,
	}, nil
}
412048
func (c *Client) SetPhase(phase migration.Phase) error {
	args := params.SetMigrationPhaseArgs{
		Phase: phase.String(),
	}
	return c.caller.FacadeCall("SetPhase", args, nil)
}
862367
func (RealTime) NewTimer(d time.Duration) Timer {
	return &builtinTimer{
		timer: time.NewTimer(d),
	}
}
1097052
func NewTransactionalLatencyInjector(b Backend, latency time.Duration, jitter int, logger log.Logger) *TransactionalLatencyInjector {
	return &TransactionalLatencyInjector{
		LatencyInjector: NewLatencyInjector(b, latency, jitter, logger),
		Transactional:   b.(Transactional),
	}
}
174880
func (a *Astilectron) NewTray(o *TrayOptions) *Tray {
	return newTray(o, a.canceller, a.dispatcher, a.identifier, a.writer)
}
774639
func (t *Transferer) NewTransfer(conf *TransferConfig) (*storagetransfer.TransferJob, error) {
	job, err := conf.Job()
	if err != nil {
		return nil, err
	}

	return t.svc.Create(job).Do()
}
1059807
func NewTruncateLogger(f string) (*FileLogger, error) {
	return NewFileLogger(f, INFO, TRUNC, 0, 0, BUFSIZE)
}
619618
func (f *FakeClock) NewTimer(d time.Duration) Timer {
	f.lock.Lock()
	defer f.lock.Unlock()
	stopTime := f.time.Add(d)
	ch := make(chan time.Time, 1) // Don't block!
	timer := &fakeTimer{
		fakeClock: f,
		waiter: fakeClockWaiter{
			targetTime: stopTime,
			destChan:   ch,
		},
	}
	f.waiters = append(f.waiters, timer.waiter)
	return timer
}
1190630
func (s *Statsd) NewTiming(name string, sampleRate float64) *Timing {
	s.rates.Set(s.prefix+name, sampleRate)
	return &Timing{
		name: s.prefix + name,
		obs:  s.timings.Observe,
	}
}
127865
func (facade *Facade) onePhase(tagString string) (string, error) {
	if err := facade.auth(tagString); err != nil {
		return "", errors.Trace(err)
	}
	phase, err := facade.backend.MigrationPhase()
	if err != nil {
		return "", errors.Trace(err)
	}
	return phase.String(), nil
}
737638
func NewTeeFromJSON(js json.RawMessage) (*WriteOperations, error) {
	tmp := struct {
		Tee json.RawMessage `json:"tee"`
	}{}
	if err := json.Unmarshal(js, &tmp); err != nil {
		return nil, err
	} else if step, err := NewWriteOperationsFromJSON(tmp.Tee); err != nil {
		return nil, err
	} else {
		return step, nil
	}
}
456282
func (tp *DocumentsWriterPerThreadPool) newThreadState() (ts *ThreadState) {
	tp.Lock()
	defer tp.Unlock()

	// Create a new empty thread state if possible
	if len(tp.threadStates) < cap(tp.threadStates) {
		ts = newThreadState(len(tp.threadStates))
		tp.threadStates = append(tp.threadStates, ts)
		tp.lockedList.PushBack(ts.id)
	}
	return
}
1308786
func NewTransferManager(concurrencyLimit int) TransferManager {
	return &transferManager{
		concurrencyLimit: concurrencyLimit,
		transfers:        make(map[string]Transfer),
	}
}
785614
func (woc *wfOperationCtx) markWorkflowPhase(phase wfv1.NodePhase, markCompleted bool, message ...string) {
	if woc.wf.Status.Phase != phase {
		woc.log.Infof("Updated phase %s -> %s", woc.wf.Status.Phase, phase)
		woc.updated = true
		woc.wf.Status.Phase = phase
		if woc.wf.ObjectMeta.Labels == nil {
			woc.wf.ObjectMeta.Labels = make(map[string]string)
		}
		woc.wf.ObjectMeta.Labels[common.LabelKeyPhase] = string(phase)
	}
	if woc.wf.Status.StartedAt.IsZero() {
		woc.updated = true
		woc.wf.Status.StartedAt = metav1.Time{Time: time.Now().UTC()}
	}
	if len(message) > 0 && woc.wf.Status.Message != message[0] {
		woc.log.Infof("Updated message %s -> %s", woc.wf.Status.Message, message[0])
		woc.updated = true
		woc.wf.Status.Message = message[0]
	}

	switch phase {
	case wfv1.NodeSucceeded, wfv1.NodeFailed, wfv1.NodeError:
		// wait for all daemon nodes to get terminated before marking workflow completed
		if markCompleted && !woc.hasDaemonNodes() {
			woc.log.Infof("Marking workflow completed")
			woc.wf.Status.FinishedAt = metav1.Time{Time: time.Now().UTC()}
			if woc.wf.ObjectMeta.Labels == nil {
				woc.wf.ObjectMeta.Labels = make(map[string]string)
			}
			woc.wf.ObjectMeta.Labels[common.LabelKeyCompleted] = "true"
			woc.updated = true
		}
	}
}