123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830 |
- // Copyright 2015 The etcd Authors
- //
- // 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.
- package mvcc
- import (
- "crypto/rand"
- "encoding/binary"
- "fmt"
- "math"
- mrand "math/rand"
- "os"
- "reflect"
- "sync"
- "testing"
- "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/etcd/pkg/testutil"
- )
- func TestStoreRev(t *testing.T) {
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer s.Close()
- defer os.Remove(tmpPath)
- for i := 1; i <= 3; i++ {
- s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
- if r := s.Rev(); r != int64(i+1) {
- t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
- }
- }
- }
- func TestStorePut(t *testing.T) {
- kv := mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 1,
- ModRevision: 2,
- Version: 1,
- }
- kvb, err := kv.Marshal()
- if err != nil {
- t.Fatal(err)
- }
- tests := []struct {
- rev revision
- r indexGetResp
- rr *rangeResp
- wrev revision
- wkey []byte
- wkv mvccpb.KeyValue
- wputrev revision
- }{
- {
- revision{1, 0},
- indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
- nil,
- revision{2, 0},
- newTestKeyBytes(revision{2, 0}, false),
- mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 2,
- ModRevision: 2,
- Version: 1,
- Lease: 1,
- },
- revision{2, 0},
- },
- {
- revision{1, 1},
- indexGetResp{revision{2, 0}, revision{2, 0}, 1, nil},
- &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
- revision{2, 0},
- newTestKeyBytes(revision{2, 0}, false),
- mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 2,
- ModRevision: 2,
- Version: 2,
- Lease: 2,
- },
- revision{2, 0},
- },
- {
- revision{2, 0},
- indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
- &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
- revision{3, 0},
- newTestKeyBytes(revision{3, 0}, false),
- mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 2,
- ModRevision: 3,
- Version: 3,
- Lease: 3,
- },
- revision{3, 0},
- },
- }
- for i, tt := range tests {
- s := newFakeStore()
- b := s.b.(*fakeBackend)
- fi := s.kvindex.(*fakeIndex)
- s.currentRev = tt.rev.main
- fi.indexGetRespc <- tt.r
- if tt.rr != nil {
- b.tx.rangeRespc <- *tt.rr
- }
- s.Put([]byte("foo"), []byte("bar"), lease.LeaseID(i+1))
- data, err := tt.wkv.Marshal()
- if err != nil {
- t.Errorf("#%d: marshal err = %v, want nil", i, err)
- }
- wact := []testutil.Action{
- {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
- }
- if tt.rr != nil {
- wact = []testutil.Action{
- {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
- }
- }
- if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
- }
- wact = []testutil.Action{
- {"get", []interface{}{[]byte("foo"), tt.wputrev.main}},
- {"put", []interface{}{[]byte("foo"), tt.wputrev}},
- }
- if g := fi.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
- }
- if s.currentRev != tt.wrev.main {
- t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
- }
- s.Close()
- }
- }
- func TestStoreRange(t *testing.T) {
- key := newTestKeyBytes(revision{2, 0}, false)
- kv := mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 1,
- ModRevision: 2,
- Version: 1,
- }
- kvb, err := kv.Marshal()
- if err != nil {
- t.Fatal(err)
- }
- wrev := int64(2)
- tests := []struct {
- idxr indexRangeResp
- r rangeResp
- }{
- {
- indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
- rangeResp{[][]byte{key}, [][]byte{kvb}},
- },
- {
- indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
- rangeResp{[][]byte{key}, [][]byte{kvb}},
- },
- }
- ro := RangeOptions{Limit: 1, Rev: 0, Count: false}
- for i, tt := range tests {
- s := newFakeStore()
- b := s.b.(*fakeBackend)
- fi := s.kvindex.(*fakeIndex)
- s.currentRev = 2
- b.tx.rangeRespc <- tt.r
- fi.indexRangeRespc <- tt.idxr
- ret, err := s.Range([]byte("foo"), []byte("goo"), ro)
- if err != nil {
- t.Errorf("#%d: err = %v, want nil", i, err)
- }
- if w := []mvccpb.KeyValue{kv}; !reflect.DeepEqual(ret.KVs, w) {
- t.Errorf("#%d: kvs = %+v, want %+v", i, ret.KVs, w)
- }
- if ret.Rev != wrev {
- t.Errorf("#%d: rev = %d, want %d", i, ret.Rev, wrev)
- }
- wstart := newRevBytes()
- revToBytes(tt.idxr.revs[0], wstart)
- wact := []testutil.Action{
- {"range", []interface{}{keyBucketName, wstart, []byte(nil), int64(0)}},
- }
- if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
- }
- wact = []testutil.Action{
- {"range", []interface{}{[]byte("foo"), []byte("goo"), wrev}},
- }
- if g := fi.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
- }
- if s.currentRev != 2 {
- t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, 2)
- }
- s.Close()
- }
- }
- func TestStoreDeleteRange(t *testing.T) {
- key := newTestKeyBytes(revision{2, 0}, false)
- kv := mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 1,
- ModRevision: 2,
- Version: 1,
- }
- kvb, err := kv.Marshal()
- if err != nil {
- t.Fatal(err)
- }
- tests := []struct {
- rev revision
- r indexRangeResp
- rr rangeResp
- wkey []byte
- wrev revision
- wrrev int64
- wdelrev revision
- }{
- {
- revision{2, 0},
- indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
- rangeResp{[][]byte{key}, [][]byte{kvb}},
- newTestKeyBytes(revision{3, 0}, true),
- revision{3, 0},
- 2,
- revision{3, 0},
- },
- }
- for i, tt := range tests {
- s := newFakeStore()
- b := s.b.(*fakeBackend)
- fi := s.kvindex.(*fakeIndex)
- s.currentRev = tt.rev.main
- fi.indexRangeRespc <- tt.r
- b.tx.rangeRespc <- tt.rr
- n, _ := s.DeleteRange([]byte("foo"), []byte("goo"))
- if n != 1 {
- t.Errorf("#%d: n = %d, want 1", i, n)
- }
- data, err := (&mvccpb.KeyValue{
- Key: []byte("foo"),
- }).Marshal()
- if err != nil {
- t.Errorf("#%d: marshal err = %v, want nil", i, err)
- }
- wact := []testutil.Action{
- {"seqput", []interface{}{keyBucketName, tt.wkey, data}},
- }
- if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
- }
- wact = []testutil.Action{
- {"range", []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
- {"tombstone", []interface{}{[]byte("foo"), tt.wdelrev}},
- }
- if g := fi.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
- }
- if s.currentRev != tt.wrev.main {
- t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
- }
- }
- }
- func TestStoreCompact(t *testing.T) {
- s := newFakeStore()
- defer s.Close()
- b := s.b.(*fakeBackend)
- fi := s.kvindex.(*fakeIndex)
- s.currentRev = 3
- fi.indexCompactRespc <- map[revision]struct{}{{1, 0}: {}}
- key1 := newTestKeyBytes(revision{1, 0}, false)
- key2 := newTestKeyBytes(revision{2, 0}, false)
- b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
- s.Compact(3)
- s.fifoSched.WaitFinish(1)
- if s.compactMainRev != 3 {
- t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
- }
- end := make([]byte, 8)
- binary.BigEndian.PutUint64(end, uint64(4))
- wact := []testutil.Action{
- {"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
- {"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
- {"delete", []interface{}{keyBucketName, key2}},
- {"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
- }
- if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("tx actions = %+v, want %+v", g, wact)
- }
- wact = []testutil.Action{
- {"compact", []interface{}{int64(3)}},
- }
- if g := fi.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("index action = %+v, want %+v", g, wact)
- }
- }
- func TestStoreRestore(t *testing.T) {
- s := newFakeStore()
- b := s.b.(*fakeBackend)
- fi := s.kvindex.(*fakeIndex)
- putkey := newTestKeyBytes(revision{3, 0}, false)
- putkv := mvccpb.KeyValue{
- Key: []byte("foo"),
- Value: []byte("bar"),
- CreateRevision: 4,
- ModRevision: 4,
- Version: 1,
- }
- putkvb, err := putkv.Marshal()
- if err != nil {
- t.Fatal(err)
- }
- delkey := newTestKeyBytes(revision{5, 0}, true)
- delkv := mvccpb.KeyValue{
- Key: []byte("foo"),
- }
- delkvb, err := delkv.Marshal()
- if err != nil {
- t.Fatal(err)
- }
- b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
- b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
- b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
- b.tx.rangeRespc <- rangeResp{nil, nil}
- s.restore()
- if s.compactMainRev != 3 {
- t.Errorf("compact rev = %d, want 5", s.compactMainRev)
- }
- if s.currentRev != 5 {
- t.Errorf("current rev = %v, want 5", s.currentRev)
- }
- wact := []testutil.Action{
- {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
- {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
- {"range", []interface{}{keyBucketName, newTestRevBytes(revision{1, 0}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(restoreChunkKeys)}},
- }
- if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("tx actions = %+v, want %+v", g, wact)
- }
- gens := []generation{
- {created: revision{4, 0}, ver: 2, revs: []revision{{3, 0}, {5, 0}}},
- {created: revision{0, 0}, ver: 0, revs: nil},
- }
- ki := &keyIndex{key: []byte("foo"), modified: revision{5, 0}, generations: gens}
- wact = []testutil.Action{
- {"keyIndex", []interface{}{ki}},
- {"insert", []interface{}{ki}},
- }
- if g := fi.Action(); !reflect.DeepEqual(g, wact) {
- t.Errorf("index action = %+v, want %+v", g, wact)
- }
- }
- func TestRestoreDelete(t *testing.T) {
- oldChunk := restoreChunkKeys
- restoreChunkKeys = mrand.Intn(3) + 2
- defer func() { restoreChunkKeys = oldChunk }()
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer os.Remove(tmpPath)
- keys := make(map[string]struct{})
- for i := 0; i < 20; i++ {
- ks := fmt.Sprintf("foo-%d", i)
- k := []byte(ks)
- s.Put(k, []byte("bar"), lease.NoLease)
- keys[ks] = struct{}{}
- switch mrand.Intn(3) {
- case 0:
- // put random key from past via random range on map
- ks = fmt.Sprintf("foo-%d", mrand.Intn(i+1))
- s.Put([]byte(ks), []byte("baz"), lease.NoLease)
- keys[ks] = struct{}{}
- case 1:
- // delete random key via random range on map
- for k := range keys {
- s.DeleteRange([]byte(k), nil)
- delete(keys, k)
- break
- }
- }
- }
- s.Close()
- s = NewStore(b, &lease.FakeLessor{}, nil)
- defer s.Close()
- for i := 0; i < 20; i++ {
- ks := fmt.Sprintf("foo-%d", i)
- r, err := s.Range([]byte(ks), nil, RangeOptions{})
- if err != nil {
- t.Fatal(err)
- }
- if _, ok := keys[ks]; ok {
- if len(r.KVs) == 0 {
- t.Errorf("#%d: expected %q, got deleted", i, ks)
- }
- } else if len(r.KVs) != 0 {
- t.Errorf("#%d: expected deleted, got %q", i, ks)
- }
- }
- }
- func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
- b, tmpPath := backend.NewDefaultTmpBackend()
- s0 := NewStore(b, &lease.FakeLessor{}, nil)
- defer os.Remove(tmpPath)
- s0.Put([]byte("foo"), []byte("bar"), lease.NoLease)
- s0.Put([]byte("foo"), []byte("bar1"), lease.NoLease)
- s0.Put([]byte("foo"), []byte("bar2"), lease.NoLease)
- // write scheduled compaction, but not do compaction
- rbytes := newRevBytes()
- revToBytes(revision{main: 2}, rbytes)
- tx := s0.b.BatchTx()
- tx.Lock()
- tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
- tx.Unlock()
- s0.Close()
- s1 := NewStore(b, &lease.FakeLessor{}, nil)
- // wait for scheduled compaction to be finished
- time.Sleep(100 * time.Millisecond)
- if _, err := s1.Range([]byte("foo"), nil, RangeOptions{Rev: 1}); err != ErrCompacted {
- t.Errorf("range on compacted rev error = %v, want %v", err, ErrCompacted)
- }
- // check the key in backend is deleted
- revbytes := newRevBytes()
- revToBytes(revision{main: 1}, revbytes)
- // The disk compaction is done asynchronously and requires more time on slow disk.
- // try 5 times for CI with slow IO.
- for i := 0; i < 5; i++ {
- tx = s1.b.BatchTx()
- tx.Lock()
- ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
- tx.Unlock()
- if len(ks) != 0 {
- time.Sleep(100 * time.Millisecond)
- continue
- }
- return
- }
- t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
- }
- type hashKVResult struct {
- hash uint32
- compactRev int64
- }
- // TestHashKVWhenCompacting ensures that HashKV returns correct hash when compacting.
- func TestHashKVWhenCompacting(t *testing.T) {
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer os.Remove(tmpPath)
- rev := 10000
- for i := 2; i <= rev; i++ {
- s.Put([]byte("foo"), []byte(fmt.Sprintf("bar%d", i)), lease.NoLease)
- }
- hashCompactc := make(chan hashKVResult, 1)
- donec := make(chan struct{})
- var wg sync.WaitGroup
- for i := 0; i < 10; i++ {
- wg.Add(1)
- go func() {
- defer wg.Done()
- for {
- hash, _, compactRev, err := s.HashByRev(int64(rev))
- if err != nil {
- t.Fatal(err)
- }
- select {
- case <-donec:
- return
- case hashCompactc <- hashKVResult{hash, compactRev}:
- }
- }
- }()
- }
- go func() {
- defer close(donec)
- revHash := make(map[int64]uint32)
- for round := 0; round < 1000; round++ {
- r := <-hashCompactc
- if revHash[r.compactRev] == 0 {
- revHash[r.compactRev] = r.hash
- }
- if r.hash != revHash[r.compactRev] {
- t.Fatalf("Hashes differ (current %v) != (saved %v)", r.hash, revHash[r.compactRev])
- }
- }
- }()
- wg.Add(1)
- go func() {
- defer wg.Done()
- for i := 100; i >= 0; i-- {
- _, err := s.Compact(int64(rev - 1 - i))
- if err != nil {
- t.Fatal(err)
- }
- time.Sleep(10 * time.Millisecond)
- }
- }()
- select {
- case <-donec:
- wg.Wait()
- case <-time.After(10 * time.Second):
- testutil.FatalStack(t, "timeout")
- }
- }
- // TestHashKVZeroRevision ensures that "HashByRev(0)" computes
- // correct hash value with latest revision.
- func TestHashKVZeroRevision(t *testing.T) {
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer os.Remove(tmpPath)
- rev := 1000
- for i := 2; i <= rev; i++ {
- s.Put([]byte("foo"), []byte(fmt.Sprintf("bar%d", i)), lease.NoLease)
- }
- if _, err := s.Compact(int64(rev / 2)); err != nil {
- t.Fatal(err)
- }
- hash1, _, _, err := s.HashByRev(int64(rev))
- if err != nil {
- t.Fatal(err)
- }
- var hash2 uint32
- hash2, _, _, err = s.HashByRev(0)
- if err != nil {
- t.Fatal(err)
- }
- if hash1 != hash2 {
- t.Errorf("hash %d (rev %d) != hash %d (rev 0)", hash1, rev, hash2)
- }
- }
- func TestTxnPut(t *testing.T) {
- // assign arbitrary size
- bytesN := 30
- sliceN := 100
- keys := createBytesSlice(bytesN, sliceN)
- vals := createBytesSlice(bytesN, sliceN)
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer cleanup(s, b, tmpPath)
- for i := 0; i < sliceN; i++ {
- txn := s.Write()
- base := int64(i + 2)
- if rev := txn.Put(keys[i], vals[i], lease.NoLease); rev != base {
- t.Errorf("#%d: rev = %d, want %d", i, rev, base)
- }
- txn.End()
- }
- }
- func TestTxnBlockBackendForceCommit(t *testing.T) {
- b, tmpPath := backend.NewDefaultTmpBackend()
- s := NewStore(b, &lease.FakeLessor{}, nil)
- defer os.Remove(tmpPath)
- txn := s.Read()
- done := make(chan struct{})
- go func() {
- s.b.ForceCommit()
- done <- struct{}{}
- }()
- select {
- case <-done:
- t.Fatalf("failed to block ForceCommit")
- case <-time.After(100 * time.Millisecond):
- }
- txn.End()
- select {
- case <-done:
- case <-time.After(5 * time.Second): // wait 5 seconds for CI with slow IO
- testutil.FatalStack(t, "failed to execute ForceCommit")
- }
- }
- // TODO: test attach key to lessor
- func newTestRevBytes(rev revision) []byte {
- bytes := newRevBytes()
- revToBytes(rev, bytes)
- return bytes
- }
- func newTestKeyBytes(rev revision, tombstone bool) []byte {
- bytes := newRevBytes()
- revToBytes(rev, bytes)
- if tombstone {
- bytes = appendMarkTombstone(bytes)
- }
- return bytes
- }
- func newFakeStore() *store {
- b := &fakeBackend{&fakeBatchTx{
- Recorder: &testutil.RecorderBuffered{},
- rangeRespc: make(chan rangeResp, 5)}}
- fi := &fakeIndex{
- Recorder: &testutil.RecorderBuffered{},
- indexGetRespc: make(chan indexGetResp, 1),
- indexRangeRespc: make(chan indexRangeResp, 1),
- indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
- indexCompactRespc: make(chan map[revision]struct{}, 1),
- }
- s := &store{
- b: b,
- le: &lease.FakeLessor{},
- kvindex: fi,
- currentRev: 0,
- compactMainRev: -1,
- fifoSched: schedule.NewFIFOScheduler(),
- stopc: make(chan struct{}),
- }
- s.ReadView, s.WriteView = &readView{s}, &writeView{s}
- return s
- }
- type rangeResp struct {
- keys [][]byte
- vals [][]byte
- }
- type fakeBatchTx struct {
- testutil.Recorder
- rangeRespc chan rangeResp
- }
- func (b *fakeBatchTx) Lock() {}
- func (b *fakeBatchTx) Unlock() {}
- func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
- func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
- b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
- }
- func (b *fakeBatchTx) UnsafeSeqPut(bucketName []byte, key []byte, value []byte) {
- b.Recorder.Record(testutil.Action{Name: "seqput", Params: []interface{}{bucketName, key, value}})
- }
- func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
- b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
- r := <-b.rangeRespc
- return r.keys, r.vals
- }
- func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
- b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
- }
- func (b *fakeBatchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
- return nil
- }
- func (b *fakeBatchTx) Commit() {}
- func (b *fakeBatchTx) CommitAndStop() {}
- type fakeBackend struct {
- tx *fakeBatchTx
- }
- func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
- func (b *fakeBackend) ReadTx() backend.ReadTx { return b.tx }
- func (b *fakeBackend) Hash(ignores map[backend.IgnoreKey]struct{}) (uint32, error) { return 0, nil }
- func (b *fakeBackend) Size() int64 { return 0 }
- func (b *fakeBackend) SizeInUse() int64 { return 0 }
- func (b *fakeBackend) Snapshot() backend.Snapshot { return nil }
- func (b *fakeBackend) ForceCommit() {}
- func (b *fakeBackend) Defrag() error { return nil }
- func (b *fakeBackend) Close() error { return nil }
- type indexGetResp struct {
- rev revision
- created revision
- ver int64
- err error
- }
- type indexRangeResp struct {
- keys [][]byte
- revs []revision
- }
- type indexRangeEventsResp struct {
- revs []revision
- }
- type fakeIndex struct {
- testutil.Recorder
- indexGetRespc chan indexGetResp
- indexRangeRespc chan indexRangeResp
- indexRangeEventsRespc chan indexRangeEventsResp
- indexCompactRespc chan map[revision]struct{}
- }
- func (i *fakeIndex) Revisions(key, end []byte, atRev int64) []revision {
- _, rev := i.Range(key, end, atRev)
- return rev
- }
- func (i *fakeIndex) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error) {
- i.Recorder.Record(testutil.Action{Name: "get", Params: []interface{}{key, atRev}})
- r := <-i.indexGetRespc
- return r.rev, r.created, r.ver, r.err
- }
- func (i *fakeIndex) Range(key, end []byte, atRev int64) ([][]byte, []revision) {
- i.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{key, end, atRev}})
- r := <-i.indexRangeRespc
- return r.keys, r.revs
- }
- func (i *fakeIndex) Put(key []byte, rev revision) {
- i.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{key, rev}})
- }
- func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
- i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
- return nil
- }
- func (i *fakeIndex) RangeSince(key, end []byte, rev int64) []revision {
- i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
- r := <-i.indexRangeEventsRespc
- return r.revs
- }
- func (i *fakeIndex) Compact(rev int64) map[revision]struct{} {
- i.Recorder.Record(testutil.Action{Name: "compact", Params: []interface{}{rev}})
- return <-i.indexCompactRespc
- }
- func (i *fakeIndex) Keep(rev int64) map[revision]struct{} {
- i.Recorder.Record(testutil.Action{Name: "keep", Params: []interface{}{rev}})
- return <-i.indexCompactRespc
- }
- func (i *fakeIndex) Equal(b index) bool { return false }
- func (i *fakeIndex) Insert(ki *keyIndex) {
- i.Recorder.Record(testutil.Action{Name: "insert", Params: []interface{}{ki}})
- }
- func (i *fakeIndex) KeyIndex(ki *keyIndex) *keyIndex {
- i.Recorder.Record(testutil.Action{Name: "keyIndex", Params: []interface{}{ki}})
- return nil
- }
- func createBytesSlice(bytesN, sliceN int) [][]byte {
- rs := [][]byte{}
- for len(rs) != sliceN {
- v := make([]byte, bytesN)
- if _, err := rand.Read(v); err != nil {
- panic(err)
- }
- rs = append(rs, v)
- }
- return rs
- }
|