func (s *store) rangeKeys(key, end []byte, limit, rangeRev int64, countOnly bool) (kvs []mvccpb.KeyValue, count int, curRev int64, err error) {
	curRev = int64(s.currentRev.main)
	if s.currentRev.sub > 0 {
		curRev += 1
	}
	if rangeRev > curRev {
		return nil, -1, s.currentRev.main, ErrFutureRev
	}
	var rev int64
	if rangeRev <= 0 {
		rev = curRev
	} else {
		rev = rangeRev
	}
	if rev < s.compactMainRev {
		return nil, -1, 0, ErrCompacted
	}
	_, revpairs := s.kvindex.Range(key, end, int64(rev))
	if len(revpairs) == 0 {
		return nil, 0, curRev, nil
	}
	if countOnly {
		return nil, len(revpairs), curRev, nil
	}
	for _, revpair := range revpairs {
		start, end := revBytesRange(revpair)
		_, vs := s.tx.UnsafeRange(keyBucketName, start, end, 0)
		if len(vs) != 1 {
			plog.Fatalf("range cannot find rev (%d,%d)", revpair.main, revpair.sub)
		}
		var kv mvccpb.KeyValue
		if err := kv.Unmarshal(vs[0]); err != nil {
			plog.Fatalf("cannot unmarshal event: %v", err)
		}
		kvs = append(kvs, kv)
		if limit > 0 && len(kvs) >= int(limit) {
			break
		}
	}
	return kvs, len(revpairs), curRev, nil
}
range is a keyword in Go, add Keys suffix.
package mvcc
import (
	"encoding/binary"
	"errors"
	"math"
	"math/rand"
	"sync"
	"time"
	"github.com/coreos/etcd/lease"
	"github.com/coreos/etcd/mvcc/backend"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/coreos/etcd/pkg/schedule"
	"github.com/coreos/pkg/capnslog"
	"golang.org/x/net/context"
)
var (
	keyBucketName  = []byte("key")
	metaBucketName = []byte("meta")
	markedRevBytesLen      = revBytesLen + 1
	markBytePosition       = markedRevBytesLen - 1
	markTombstone     byte = 't'
	consistentIndexKeyName  = []byte("consistent_index")
	scheduledCompactKeyName = []byte("scheduledCompactRev")
	finishedCompactKeyName  = []byte("finishedCompactRev")
	ErrTxnIDMismatch = errors.New("mvcc: txn id mismatch")
	ErrCompacted     = errors.New("mvcc: required revision has been compacted")
	ErrFutureRev     = errors.New("mvcc: required revision is a future revision")
	ErrCanceled      = errors.New("mvcc: watcher is canceled")
	plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "mvcc")
)
type ConsistentIndexGetter interface {
	ConsistentIndex() uint64
}
type store struct {
	mu sync.Mutex  
	ig ConsistentIndexGetter
	b       backend.Backend
	kvindex index
	le lease.Lessor
	currentRev revision
	compactMainRev int64
	tx    backend.BatchTx
	txnID int64  
	bytesBuf8 []byte
	changes   []mvccpb.KeyValue
	fifoSched schedule.Scheduler
	stopc chan struct{}
}
func NewStore(b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter) *store {
	s := &store{
		b:       b,
		ig:      ig,
		kvindex: newTreeIndex(),
		le: le,
		currentRev:     revision{main: 1},
		compactMainRev: -1,
		bytesBuf8: make([]byte, 8, 8),
		fifoSched: schedule.NewFIFOScheduler(),
		stopc: make(chan struct{}),
	}
	if s.le != nil {
		s.le.SetRangeDeleter(s)
	}
	tx := s.b.BatchTx()
	tx.Lock()
	tx.UnsafeCreateBucket(keyBucketName)
	tx.UnsafeCreateBucket(metaBucketName)
	tx.Unlock()
	s.b.ForceCommit()
	if err := s.restore(); err != nil {
		panic("failed to recover store from backend")
	}
	return s
}
func (s *store) Rev() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.currentRev.main
}
func (s *store) FirstRev() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.compactMainRev
}
func (s *store) Put(key, value []byte, lease lease.LeaseID) int64 {
	id := s.TxnBegin()
	s.put(key, value, lease)
	s.txnEnd(id)
	putCounter.Inc()
	return int64(s.currentRev.main)
}
func (s *store) Range(key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
	id := s.TxnBegin()
	kvs, count, rev, err := s.rangeKeys(key, end, ro.Limit, ro.Rev, ro.Count)
	s.txnEnd(id)
	rangeCounter.Inc()
	r = &RangeResult{
		KVs:   kvs,
		Count: count,
		Rev:   rev,
	}
	return r, err
}
func (s *store) DeleteRange(key, end []byte) (n, rev int64) {
	id := s.TxnBegin()
	n = s.deleteRange(key, end)
	s.txnEnd(id)
	deleteCounter.Inc()
	return n, int64(s.currentRev.main)
}
func (s *store) TxnBegin() int64 {
	s.mu.Lock()
	s.currentRev.sub = 0
	s.tx = s.b.BatchTx()
	s.tx.Lock()
	s.saveIndex()
	s.txnID = rand.Int63()
	return s.txnID
}
func (s *store) TxnEnd(txnID int64) error {
	err := s.txnEnd(txnID)
	if err != nil {
		return err
	}
	txnCounter.Inc()
	return nil
}
func (s *store) txnEnd(txnID int64) error {
	if txnID != s.txnID {
		return ErrTxnIDMismatch
	}
	s.tx.Unlock()
	if s.currentRev.sub != 0 {
		s.currentRev.main += 1
	}
	s.currentRev.sub = 0
	dbTotalSize.Set(float64(s.b.Size()))
	s.mu.Unlock()
	return nil
}
func (s *store) TxnRange(txnID int64, key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
	if txnID != s.txnID {
		return nil, ErrTxnIDMismatch
	}
	kvs, count, rev, err := s.rangeKeys(key, end, ro.Limit, ro.Rev, ro.Count)
	r = &RangeResult{
		KVs:   kvs,
		Count: count,
		Rev:   rev,
	}
	return r, err
}
func (s *store) TxnPut(txnID int64, key, value []byte, lease lease.LeaseID) (rev int64, err error) {
	if txnID != s.txnID {
		return 0, ErrTxnIDMismatch
	}
	s.put(key, value, lease)
	return int64(s.currentRev.main + 1), nil
}
func (s *store) TxnDeleteRange(txnID int64, key, end []byte) (n, rev int64, err error) {
	if txnID != s.txnID {
		return 0, 0, ErrTxnIDMismatch
	}
	n = s.deleteRange(key, end)
	if n != 0 || s.currentRev.sub != 0 {
		rev = int64(s.currentRev.main + 1)
	} else {
		rev = int64(s.currentRev.main)
	}
	return n, rev, nil
}
func (s *store) compactBarrier(ctx context.Context, ch chan struct{}) {
	if ctx == nil || ctx.Err() != nil {
		s.mu.Lock()
		select {
		case <-s.stopc:
		default:
			f := func(ctx context.Context) { s.compactBarrier(ctx, ch) }
			s.fifoSched.Schedule(f)
		}
		s.mu.Unlock()
		return
	}
	close(ch)
}
func (s *store) Compact(rev int64) (<-chan struct{}, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if rev <= s.compactMainRev {
		ch := make(chan struct{})
		f := func(ctx context.Context) { s.compactBarrier(ctx, ch) }
		s.fifoSched.Schedule(f)
		return ch, ErrCompacted
	}
	if rev > s.currentRev.main {
		return nil, ErrFutureRev
	}
	start := time.Now()
	s.compactMainRev = rev
	rbytes := newRevBytes()
	revToBytes(revision{main: rev}, rbytes)
	tx := s.b.BatchTx()
	tx.Lock()
	tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
	tx.Unlock()
	s.b.ForceCommit()
	keep := s.kvindex.Compact(rev)
	ch := make(chan struct{})
	var j = func(ctx context.Context) {
		if ctx.Err() != nil {
			s.compactBarrier(ctx, ch)
			return
		}
		if !s.scheduleCompaction(rev, keep) {
			s.compactBarrier(nil, ch)
			return
		}
		close(ch)
	}
	s.fifoSched.Schedule(j)
	indexCompactionPauseDurations.Observe(float64(time.Since(start) / time.Millisecond))
	return ch, nil
}
func (s *store) Hash() (uint32, int64, error) {
	s.b.ForceCommit()
	s.mu.Lock()
	defer s.mu.Unlock()
	ignores := make(map[backend.IgnoreKey]struct{})
	bk := backend.IgnoreKey{Bucket: string(metaBucketName), Key: string(consistentIndexKeyName)}
	ignores[bk] = struct{}{}
	h, err := s.b.Hash(ignores)
	rev := s.currentRev.main
	return h, rev, err
}
func (s *store) Commit() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.tx = s.b.BatchTx()
	s.tx.Lock()
	s.saveIndex()
	s.tx.Unlock()
	s.b.ForceCommit()
}
func (s *store) Restore(b backend.Backend) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	close(s.stopc)
	s.fifoSched.Stop()
	s.b = b
	s.kvindex = newTreeIndex()
	s.currentRev = revision{main: 1}
	s.compactMainRev = -1
	s.tx = b.BatchTx()
	s.txnID = -1
	s.fifoSched = schedule.NewFIFOScheduler()
	s.stopc = make(chan struct{})
	return s.restore()
}
func (s *store) restore() error {
	min, max := newRevBytes(), newRevBytes()
	revToBytes(revision{main: 1}, min)
	revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)
	keyToLease := make(map[string]lease.LeaseID)
	tx := s.b.BatchTx()
	tx.Lock()
	_, finishedCompactBytes := tx.UnsafeRange(metaBucketName, finishedCompactKeyName, nil, 0)
	if len(finishedCompactBytes) != 0 {
		s.compactMainRev = bytesToRev(finishedCompactBytes[0]).main
		plog.Printf("restore compact to %d", s.compactMainRev)
	}
	keys, vals := tx.UnsafeRange(keyBucketName, min, max, 0)
	for i, key := range keys {
		var kv mvccpb.KeyValue
		if err := kv.Unmarshal(vals[i]); err != nil {
			plog.Fatalf("cannot unmarshal event: %v", err)
		}
		rev := bytesToRev(key[:revBytesLen])
		switch {
		case isTombstone(key):
			s.kvindex.Tombstone(kv.Key, rev)
			delete(keyToLease, string(kv.Key))
		default:
			s.kvindex.Restore(kv.Key, revision{kv.CreateRevision, 0}, rev, kv.Version)
			if lid := lease.LeaseID(kv.Lease); lid != lease.NoLease {
				keyToLease[string(kv.Key)] = lid
			} else {
				delete(keyToLease, string(kv.Key))
			}
		}
		s.currentRev = rev
	}
	for key, lid := range keyToLease {
		if s.le == nil {
			panic("no lessor to attach lease")
		}
		err := s.le.Attach(lid, []lease.LeaseItem{{Key: key}})
		if err != nil {
			plog.Errorf("unexpected Attach error: %v", err)
		}
	}
	_, scheduledCompactBytes := tx.UnsafeRange(metaBucketName, scheduledCompactKeyName, nil, 0)
	scheduledCompact := int64(0)
	if len(scheduledCompactBytes) != 0 {
		scheduledCompact = bytesToRev(scheduledCompactBytes[0]).main
		if scheduledCompact <= s.compactMainRev {
			scheduledCompact = 0
		}
	}
	tx.Unlock()
	if scheduledCompact != 0 {
		s.Compact(scheduledCompact)
		plog.Printf("resume scheduled compaction at %d", scheduledCompact)
	}
	return nil
}
func (s *store) Close() error {
	close(s.stopc)
	s.fifoSched.Stop()
	return nil
}
func (a *store) Equal(b *store) bool {
	if a.currentRev != b.currentRev {
		return false
	}
	if a.compactMainRev != b.compactMainRev {
		return false
	}
	return a.kvindex.Equal(b.kvindex)
}