Method write

xenomint/state.go:358–478

(
	ctx context.Context, req *types.Request, isLeader bool)

}

func (s *State) write(
	ctx context.Context, req *types.Request, isLeader bool) (ref *QueryTracker, resp *types.Response, err error,
) {
	var (
		lastSeq           uint64
		query             = &QueryTracker{Req: req}
		totalAffectedRows int64
		curAffectedRows   int64
		lastInsertID      int64
		start             = time.Now()

		lockAcquired, writeDone, enqueued, lockReleased, respBuilt time.Duration
	)

	defer func() {
		var fields = log.Fields{}
		fields["lastSeq"] = lastSeq
		fields["1#lockAcquired"] = float64(lockAcquired.Nanoseconds()) / 1000
		if writeDone > 0 {
			fields["2#writeDone"] = float64((writeDone - lockAcquired).Nanoseconds()) / 1000
		}
		if enqueued > 0 {
			fields["3#enqueued"] = float64((enqueued - writeDone).Nanoseconds()) / 1000
		}
		if lockReleased > 0 {
			fields["4#lockReleased"] = float64((lockReleased - enqueued).Nanoseconds()) / 1000
		}
		if respBuilt > 0 {
			fields["5#respBuilt"] = float64((respBuilt - lockReleased).Nanoseconds()) / 1000
		}
		log.WithFields(fields).Debug("Write duration stat (us)")
		if ctx.Err() != nil {
			log.WithError(err).WithField("req", req).Warning("write query canceled")
		}
	}()

	if err = func() (err error) {
		var (
			ierr error
			qcnt = len(req.Payload.Queries)
		)
		s.Lock()
		lockAcquired = time.Since(start)
		defer func() {
			s.Unlock()
			lockReleased = time.Since(start)
		}()
		lastSeq = s.getSeq()
		if qcnt > 1 && s.level == sql.LevelReadUncommitted {
			// Set savepoint
			if _, ierr = s.handler.Exec(`SAVEPOINT "?"`, lastSeq); ierr != nil {
				err = errors.Wrapf(ierr, "failed to create savepoint %d", lastSeq)
				return
			}
			defer func() {
				_, _ = s.handler.Exec(`ROLLBACK TO "?"`, lastSeq)
			}()
		}