2016-05-13 06:50:33 +03:00
|
|
|
// Copyright 2015 The etcd Authors
|
2015-09-15 23:54:11 +03:00
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
2016-04-25 22:32:58 +03:00
|
|
|
package mvcc
|
2015-05-22 18:11:43 +03:00
|
|
|
|
|
|
|
import (
|
2016-03-30 21:37:55 +03:00
|
|
|
"encoding/binary"
|
2015-05-27 20:35:51 +03:00
|
|
|
"errors"
|
2015-06-08 19:26:56 +03:00
|
|
|
"math"
|
2015-05-22 18:11:43 +03:00
|
|
|
"sync"
|
2017-05-03 03:18:10 +03:00
|
|
|
"sync/atomic"
|
2015-05-22 18:11:43 +03:00
|
|
|
"time"
|
|
|
|
|
2016-01-05 21:16:50 +03:00
|
|
|
"github.com/coreos/etcd/lease"
|
2016-04-25 22:32:58 +03:00
|
|
|
"github.com/coreos/etcd/mvcc/backend"
|
|
|
|
"github.com/coreos/etcd/mvcc/mvccpb"
|
2016-02-07 00:43:56 +03:00
|
|
|
"github.com/coreos/etcd/pkg/schedule"
|
2016-05-21 08:30:50 +03:00
|
|
|
"github.com/coreos/pkg/capnslog"
|
2016-03-23 03:10:28 +03:00
|
|
|
"golang.org/x/net/context"
|
2015-05-22 18:11:43 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
var (
|
2015-06-17 23:52:17 +03:00
|
|
|
keyBucketName = []byte("key")
|
|
|
|
metaBucketName = []byte("meta")
|
2015-05-27 20:35:51 +03:00
|
|
|
|
2016-03-30 21:37:55 +03:00
|
|
|
consistentIndexKeyName = []byte("consistent_index")
|
2015-05-31 18:59:31 +03:00
|
|
|
scheduledCompactKeyName = []byte("scheduledCompactRev")
|
|
|
|
finishedCompactKeyName = []byte("finishedCompactRev")
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
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")
|
|
|
|
ErrClosed = errors.New("mvcc: closed")
|
2016-05-21 08:30:50 +03:00
|
|
|
|
|
|
|
plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "mvcc")
|
2015-05-22 18:11:43 +03:00
|
|
|
)
|
|
|
|
|
2017-05-09 05:29:53 +03:00
|
|
|
const (
|
|
|
|
// markedRevBytesLen is the byte length of marked revision.
|
|
|
|
// The first `revBytesLen` bytes represents a normal revision. The last
|
|
|
|
// one byte is the mark.
|
|
|
|
markedRevBytesLen = revBytesLen + 1
|
|
|
|
markBytePosition = markedRevBytesLen - 1
|
|
|
|
markTombstone byte = 't'
|
|
|
|
|
|
|
|
restoreChunkKeys = 10000
|
|
|
|
)
|
|
|
|
|
2016-03-30 21:37:55 +03:00
|
|
|
// ConsistentIndexGetter is an interface that wraps the Get method.
|
|
|
|
// Consistent index is the offset of an entry in a consistent replicated log.
|
|
|
|
type ConsistentIndexGetter interface {
|
|
|
|
// ConsistentIndex returns the consistent index of current executing entry.
|
|
|
|
ConsistentIndex() uint64
|
|
|
|
}
|
|
|
|
|
2015-05-22 18:11:43 +03:00
|
|
|
type store struct {
|
2017-01-05 04:01:31 +03:00
|
|
|
ReadView
|
|
|
|
WriteView
|
|
|
|
|
2017-05-03 03:18:10 +03:00
|
|
|
// consistentIndex caches the "consistent_index" key's value. Accessed
|
|
|
|
// through atomics so must be 64-bit aligned.
|
|
|
|
consistentIndex uint64
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
// mu read locks for txns and write locks for non-txn store changes.
|
|
|
|
mu sync.RWMutex
|
2015-05-22 18:11:43 +03:00
|
|
|
|
2016-03-30 21:37:55 +03:00
|
|
|
ig ConsistentIndexGetter
|
|
|
|
|
2015-05-22 18:11:43 +03:00
|
|
|
b backend.Backend
|
|
|
|
kvindex index
|
|
|
|
|
2016-01-08 21:06:33 +03:00
|
|
|
le lease.Lessor
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
// revMuLock protects currentRev and compactMainRev.
|
|
|
|
// Locked at end of write txn and released after write txn unlock lock.
|
|
|
|
// Locked before locking read txn and released after locking.
|
|
|
|
revMu sync.RWMutex
|
|
|
|
// currentRev is the revision of the last completed transaction.
|
|
|
|
currentRev int64
|
|
|
|
// compactMainRev is the main revision of the last compaction.
|
2015-05-31 18:59:31 +03:00
|
|
|
compactMainRev int64
|
2015-05-22 23:35:43 +03:00
|
|
|
|
2016-05-18 19:42:37 +03:00
|
|
|
// bytesBuf8 is a byte slice of length 8
|
|
|
|
// to avoid a repetitive allocation in saveIndex.
|
|
|
|
bytesBuf8 []byte
|
|
|
|
|
2016-02-07 00:43:56 +03:00
|
|
|
fifoSched schedule.Scheduler
|
2016-02-01 02:56:31 +03:00
|
|
|
|
2015-06-18 21:05:31 +03:00
|
|
|
stopc chan struct{}
|
2015-05-22 18:11:43 +03:00
|
|
|
}
|
|
|
|
|
2016-01-06 06:45:18 +03:00
|
|
|
// NewStore returns a new store. It is useful to create a store inside
|
2016-04-25 22:32:58 +03:00
|
|
|
// mvcc pkg. It should only be used for testing externally.
|
2016-03-30 21:37:55 +03:00
|
|
|
func NewStore(b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter) *store {
|
2015-05-22 18:11:43 +03:00
|
|
|
s := &store{
|
2016-01-08 21:06:33 +03:00
|
|
|
b: b,
|
2016-03-30 21:37:55 +03:00
|
|
|
ig: ig,
|
2016-01-08 21:06:33 +03:00
|
|
|
kvindex: newTreeIndex(),
|
|
|
|
|
|
|
|
le: le,
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
currentRev: 1,
|
2015-05-31 18:59:31 +03:00
|
|
|
compactMainRev: -1,
|
2016-02-07 00:43:56 +03:00
|
|
|
|
2017-02-09 16:01:28 +03:00
|
|
|
bytesBuf8: make([]byte, 8),
|
2016-02-07 00:43:56 +03:00
|
|
|
fifoSched: schedule.NewFIFOScheduler(),
|
|
|
|
|
|
|
|
stopc: make(chan struct{}),
|
2015-05-22 18:11:43 +03:00
|
|
|
}
|
2017-01-05 04:01:31 +03:00
|
|
|
s.ReadView = &readView{s}
|
|
|
|
s.WriteView = &writeView{s}
|
2016-01-08 21:06:33 +03:00
|
|
|
if s.le != nil {
|
2017-01-05 04:01:31 +03:00
|
|
|
s.le.SetRangeDeleter(func() lease.TxnDelete { return s.Write() })
|
2016-01-08 21:06:33 +03:00
|
|
|
}
|
|
|
|
|
2015-05-22 18:11:43 +03:00
|
|
|
tx := s.b.BatchTx()
|
|
|
|
tx.Lock()
|
|
|
|
tx.UnsafeCreateBucket(keyBucketName)
|
2015-06-17 23:52:17 +03:00
|
|
|
tx.UnsafeCreateBucket(metaBucketName)
|
2015-05-22 18:11:43 +03:00
|
|
|
tx.Unlock()
|
|
|
|
s.b.ForceCommit()
|
|
|
|
|
2016-01-07 03:25:25 +03:00
|
|
|
if err := s.restore(); err != nil {
|
|
|
|
// TODO: return the error instead of panic here?
|
|
|
|
panic("failed to recover store from backend")
|
|
|
|
}
|
|
|
|
|
2015-05-22 18:11:43 +03:00
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
2016-04-01 20:59:28 +03:00
|
|
|
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)
|
|
|
|
}
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
func (s *store) Hash() (hash uint32, revision int64, err error) {
|
|
|
|
s.b.ForceCommit()
|
|
|
|
h, err := s.b.Hash(DefaultIgnores)
|
|
|
|
return h, s.currentRev, err
|
|
|
|
}
|
|
|
|
|
2016-03-29 02:07:56 +03:00
|
|
|
func (s *store) Compact(rev int64) (<-chan struct{}, error) {
|
2015-05-31 18:59:31 +03:00
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
2017-01-05 04:01:31 +03:00
|
|
|
s.revMu.Lock()
|
|
|
|
defer s.revMu.Unlock()
|
|
|
|
|
2015-05-31 18:59:31 +03:00
|
|
|
if rev <= s.compactMainRev {
|
2016-03-30 23:58:10 +03:00
|
|
|
ch := make(chan struct{})
|
2016-04-01 20:59:28 +03:00
|
|
|
f := func(ctx context.Context) { s.compactBarrier(ctx, ch) }
|
|
|
|
s.fifoSched.Schedule(f)
|
2016-03-30 23:58:10 +03:00
|
|
|
return ch, ErrCompacted
|
2015-05-31 18:59:31 +03:00
|
|
|
}
|
2017-01-05 04:01:31 +03:00
|
|
|
if rev > s.currentRev {
|
2016-03-29 02:07:56 +03:00
|
|
|
return nil, ErrFutureRev
|
2015-08-23 02:44:39 +03:00
|
|
|
}
|
2015-05-31 18:59:31 +03:00
|
|
|
|
2015-08-28 20:35:30 +03:00
|
|
|
start := time.Now()
|
|
|
|
|
2015-05-31 18:59:31 +03:00
|
|
|
s.compactMainRev = rev
|
|
|
|
|
2015-06-08 19:26:56 +03:00
|
|
|
rbytes := newRevBytes()
|
2015-08-20 18:39:07 +03:00
|
|
|
revToBytes(revision{main: rev}, rbytes)
|
2015-05-31 18:59:31 +03:00
|
|
|
|
|
|
|
tx := s.b.BatchTx()
|
|
|
|
tx.Lock()
|
2015-06-17 23:52:17 +03:00
|
|
|
tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
|
2015-05-31 18:59:31 +03:00
|
|
|
tx.Unlock()
|
2015-08-23 02:47:19 +03:00
|
|
|
// ensure that desired compaction is persisted
|
|
|
|
s.b.ForceCommit()
|
2015-05-31 18:59:31 +03:00
|
|
|
|
|
|
|
keep := s.kvindex.Compact(rev)
|
2016-03-29 02:07:56 +03:00
|
|
|
ch := make(chan struct{})
|
2016-02-07 00:43:56 +03:00
|
|
|
var j = func(ctx context.Context) {
|
2016-04-01 20:59:28 +03:00
|
|
|
if ctx.Err() != nil {
|
|
|
|
s.compactBarrier(ctx, ch)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if !s.scheduleCompaction(rev, keep) {
|
|
|
|
s.compactBarrier(nil, ch)
|
2016-02-07 00:43:56 +03:00
|
|
|
return
|
|
|
|
}
|
2016-04-01 20:59:28 +03:00
|
|
|
close(ch)
|
2016-02-07 00:43:56 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
s.fifoSched.Schedule(j)
|
2015-08-28 20:35:30 +03:00
|
|
|
|
2016-04-08 08:09:25 +03:00
|
|
|
indexCompactionPauseDurations.Observe(float64(time.Since(start) / time.Millisecond))
|
2016-03-29 02:07:56 +03:00
|
|
|
return ch, nil
|
2015-05-31 18:59:31 +03:00
|
|
|
}
|
|
|
|
|
2016-10-25 20:07:08 +03:00
|
|
|
// DefaultIgnores is a map of keys to ignore in hash checking.
|
|
|
|
var DefaultIgnores map[backend.IgnoreKey]struct{}
|
|
|
|
|
|
|
|
func init() {
|
|
|
|
DefaultIgnores = map[backend.IgnoreKey]struct{}{
|
|
|
|
// consistent index might be changed due to v2 internal sync, which
|
|
|
|
// is not controllable by the user.
|
|
|
|
{Bucket: string(metaBucketName), Key: string(consistentIndexKeyName)}: {},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-06-27 20:27:27 +03:00
|
|
|
func (s *store) Commit() {
|
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
tx := s.b.BatchTx()
|
|
|
|
tx.Lock()
|
|
|
|
s.saveIndex(tx)
|
|
|
|
tx.Unlock()
|
2016-06-27 20:27:27 +03:00
|
|
|
s.b.ForceCommit()
|
|
|
|
}
|
2015-10-27 07:06:08 +03:00
|
|
|
|
2016-01-07 03:25:25 +03:00
|
|
|
func (s *store) Restore(b backend.Backend) error {
|
2015-06-08 19:26:56 +03:00
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
2016-01-07 03:25:25 +03:00
|
|
|
close(s.stopc)
|
2016-02-07 00:43:56 +03:00
|
|
|
s.fifoSched.Stop()
|
2016-01-07 03:25:25 +03:00
|
|
|
|
2017-05-03 03:18:10 +03:00
|
|
|
atomic.StoreUint64(&s.consistentIndex, 0)
|
2016-01-07 03:25:25 +03:00
|
|
|
s.b = b
|
|
|
|
s.kvindex = newTreeIndex()
|
2017-01-05 04:01:31 +03:00
|
|
|
s.currentRev = 1
|
2016-01-07 03:25:25 +03:00
|
|
|
s.compactMainRev = -1
|
2016-02-07 00:43:56 +03:00
|
|
|
s.fifoSched = schedule.NewFIFOScheduler()
|
2016-01-07 03:25:25 +03:00
|
|
|
s.stopc = make(chan struct{})
|
|
|
|
|
|
|
|
return s.restore()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *store) restore() error {
|
2015-06-08 19:26:56 +03:00
|
|
|
min, max := newRevBytes(), newRevBytes()
|
2016-01-15 02:50:42 +03:00
|
|
|
revToBytes(revision{main: 1}, min)
|
2015-08-20 18:39:07 +03:00
|
|
|
revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)
|
2015-06-08 19:26:56 +03:00
|
|
|
|
2016-08-04 21:17:56 +03:00
|
|
|
keyToLease := make(map[string]lease.LeaseID)
|
|
|
|
|
2015-06-08 19:26:56 +03:00
|
|
|
// restore index
|
|
|
|
tx := s.b.BatchTx()
|
|
|
|
tx.Lock()
|
2015-06-17 23:52:17 +03:00
|
|
|
_, finishedCompactBytes := tx.UnsafeRange(metaBucketName, finishedCompactKeyName, nil, 0)
|
2015-06-08 19:26:56 +03:00
|
|
|
if len(finishedCompactBytes) != 0 {
|
|
|
|
s.compactMainRev = bytesToRev(finishedCompactBytes[0]).main
|
2016-05-21 08:30:50 +03:00
|
|
|
plog.Printf("restore compact to %d", s.compactMainRev)
|
2015-06-08 19:26:56 +03:00
|
|
|
}
|
2017-05-09 05:29:53 +03:00
|
|
|
_, scheduledCompactBytes := tx.UnsafeRange(metaBucketName, scheduledCompactKeyName, nil, 0)
|
|
|
|
scheduledCompact := int64(0)
|
|
|
|
if len(scheduledCompactBytes) != 0 {
|
|
|
|
scheduledCompact = bytesToRev(scheduledCompactBytes[0]).main
|
|
|
|
}
|
2015-06-08 19:26:56 +03:00
|
|
|
|
2017-05-09 05:29:53 +03:00
|
|
|
// index keys concurrently as they're loaded in from tx
|
|
|
|
unorderedc, donec := make(chan map[string]*keyIndex), make(chan struct{})
|
|
|
|
go func() {
|
|
|
|
defer close(donec)
|
|
|
|
for unordered := range unorderedc {
|
|
|
|
// restore the tree index from the unordered index.
|
|
|
|
for _, v := range unordered {
|
|
|
|
s.kvindex.Insert(v)
|
2016-01-08 21:06:33 +03:00
|
|
|
}
|
2015-06-08 19:26:56 +03:00
|
|
|
}
|
2017-05-09 05:29:53 +03:00
|
|
|
}()
|
|
|
|
for {
|
|
|
|
keys, vals := tx.UnsafeRange(keyBucketName, min, max, restoreChunkKeys)
|
|
|
|
if len(keys) == 0 {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
// unbuffered so keys don't pile up in memory
|
|
|
|
unorderedc <- s.restoreChunk(keys, vals, keyToLease)
|
|
|
|
if len(keys) < restoreChunkKeys {
|
|
|
|
// partial set implies final set
|
|
|
|
break
|
|
|
|
}
|
|
|
|
// next set begins after where this one ended
|
|
|
|
newMin := bytesToRev(keys[len(keys)-1][:revBytesLen])
|
|
|
|
newMin.sub++
|
|
|
|
revToBytes(newMin, min)
|
2015-06-08 19:26:56 +03:00
|
|
|
}
|
2017-05-09 05:29:53 +03:00
|
|
|
close(unorderedc)
|
|
|
|
<-donec
|
2016-11-16 14:01:20 +03:00
|
|
|
|
2016-10-12 20:42:55 +03:00
|
|
|
// keys in the range [compacted revision -N, compaction] might all be deleted due to compaction.
|
|
|
|
// the correct revision should be set to compaction revision in the case, not the largest revision
|
|
|
|
// we have seen.
|
2017-01-05 04:01:31 +03:00
|
|
|
if s.currentRev < s.compactMainRev {
|
|
|
|
s.currentRev = s.compactMainRev
|
2016-10-12 20:42:55 +03:00
|
|
|
}
|
2017-05-09 05:29:53 +03:00
|
|
|
if scheduledCompact <= s.compactMainRev {
|
|
|
|
scheduledCompact = 0
|
|
|
|
}
|
2016-10-12 20:42:55 +03:00
|
|
|
|
2016-08-04 21:17:56 +03:00
|
|
|
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)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-06-08 19:26:56 +03:00
|
|
|
tx.Unlock()
|
|
|
|
|
2016-03-31 10:03:34 +03:00
|
|
|
if scheduledCompact != 0 {
|
|
|
|
s.Compact(scheduledCompact)
|
2016-05-21 08:30:50 +03:00
|
|
|
plog.Printf("resume scheduled compaction at %d", scheduledCompact)
|
2016-03-31 10:03:34 +03:00
|
|
|
}
|
|
|
|
|
2015-06-08 19:26:56 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2017-05-09 05:29:53 +03:00
|
|
|
func (s *store) restoreChunk(keys, vals [][]byte, keyToLease map[string]lease.LeaseID) map[string]*keyIndex {
|
|
|
|
// assume half of keys are overwrites
|
|
|
|
unordered := make(map[string]*keyIndex, len(keys)/2)
|
|
|
|
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])
|
|
|
|
s.currentRev = rev.main
|
|
|
|
kstr := string(kv.Key)
|
|
|
|
if isTombstone(key) {
|
|
|
|
if ki, ok := unordered[kstr]; ok {
|
|
|
|
ki.tombstone(rev.main, rev.sub)
|
|
|
|
}
|
|
|
|
delete(keyToLease, kstr)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
if ki, ok := unordered[kstr]; ok {
|
|
|
|
ki.put(rev.main, rev.sub)
|
|
|
|
} else {
|
|
|
|
ki = &keyIndex{key: kv.Key}
|
|
|
|
ki.restore(revision{kv.CreateRevision, 0}, rev, kv.Version)
|
|
|
|
unordered[kstr] = ki
|
|
|
|
}
|
|
|
|
if lid := lease.LeaseID(kv.Lease); lid != lease.NoLease {
|
|
|
|
keyToLease[kstr] = lid
|
|
|
|
} else {
|
|
|
|
delete(keyToLease, kstr)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return unordered
|
|
|
|
}
|
|
|
|
|
2015-06-08 19:26:56 +03:00
|
|
|
func (s *store) Close() error {
|
2015-06-18 21:05:31 +03:00
|
|
|
close(s.stopc)
|
2016-02-07 00:43:56 +03:00
|
|
|
s.fifoSched.Stop()
|
2016-01-06 06:45:18 +03:00
|
|
|
return nil
|
2015-06-08 19:26:56 +03:00
|
|
|
}
|
|
|
|
|
2017-01-05 04:01:31 +03:00
|
|
|
func (s *store) saveIndex(tx backend.BatchTx) {
|
2016-03-30 21:37:55 +03:00
|
|
|
if s.ig == nil {
|
|
|
|
return
|
|
|
|
}
|
2016-05-18 19:42:37 +03:00
|
|
|
bs := s.bytesBuf8
|
2017-05-03 03:18:10 +03:00
|
|
|
ci := s.ig.ConsistentIndex()
|
|
|
|
binary.BigEndian.PutUint64(bs, ci)
|
2016-03-30 21:37:55 +03:00
|
|
|
// put the index into the underlying backend
|
|
|
|
// tx has been locked in TxnBegin, so there is no need to lock it again
|
|
|
|
tx.UnsafePut(metaBucketName, consistentIndexKeyName, bs)
|
2017-05-03 03:18:10 +03:00
|
|
|
atomic.StoreUint64(&s.consistentIndex, ci)
|
2016-03-30 21:37:55 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *store) ConsistentIndex() uint64 {
|
2017-05-03 03:18:10 +03:00
|
|
|
if ci := atomic.LoadUint64(&s.consistentIndex); ci > 0 {
|
|
|
|
return ci
|
|
|
|
}
|
2016-03-30 21:37:55 +03:00
|
|
|
tx := s.b.BatchTx()
|
|
|
|
tx.Lock()
|
|
|
|
defer tx.Unlock()
|
|
|
|
_, vs := tx.UnsafeRange(metaBucketName, consistentIndexKeyName, nil, 0)
|
|
|
|
if len(vs) == 0 {
|
|
|
|
return 0
|
|
|
|
}
|
2017-05-03 03:18:10 +03:00
|
|
|
v := binary.BigEndian.Uint64(vs[0])
|
|
|
|
atomic.StoreUint64(&s.consistentIndex, v)
|
|
|
|
return v
|
2016-03-30 21:37:55 +03:00
|
|
|
}
|
|
|
|
|
2015-11-10 04:03:01 +03:00
|
|
|
// appendMarkTombstone appends tombstone mark to normal revision bytes.
|
|
|
|
func appendMarkTombstone(b []byte) []byte {
|
|
|
|
if len(b) != revBytesLen {
|
2016-05-21 08:30:50 +03:00
|
|
|
plog.Panicf("cannot append mark to non normal revision bytes")
|
2015-11-10 04:03:01 +03:00
|
|
|
}
|
|
|
|
return append(b, markTombstone)
|
|
|
|
}
|
|
|
|
|
|
|
|
// isTombstone checks whether the revision bytes is a tombstone.
|
|
|
|
func isTombstone(b []byte) bool {
|
|
|
|
return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
|
|
|
|
}
|
|
|
|
|
|
|
|
// revBytesRange returns the range of revision bytes at
|
|
|
|
// the given revision.
|
|
|
|
func revBytesRange(rev revision) (start, end []byte) {
|
|
|
|
start = newRevBytes()
|
|
|
|
revToBytes(rev, start)
|
|
|
|
|
|
|
|
end = newRevBytes()
|
|
|
|
endRev := revision{main: rev.main, sub: rev.sub + 1}
|
|
|
|
revToBytes(endRev, end)
|
|
|
|
|
|
|
|
return start, end
|
|
|
|
}
|