Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

storage: reversion -> revision

Browse Source
This commit is contained in:
Tyler Neely 2015-08-20 08:39:07 -07:00
parent e1dfcec0ab
commit acd7a92f03
9 changed files with 188 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
storage/index.go
View File

@ -8,12 +8,12 @@ import (
) )
type index interface { type index interface {
Get(key []byte, atRev int64) (rev, created reversion, ver int64, err error) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error)
Range(key, end []byte, atRev int64) ([][]byte, []reversion) Range(key, end []byte, atRev int64) ([][]byte, []revision)
Put(key []byte, rev reversion) Put(key []byte, rev revision)
Restore(key []byte, created, modified reversion, ver int64) Restore(key []byte, created, modified revision, ver int64)
Tombstone(key []byte, rev reversion) error Tombstone(key []byte, rev revision) error
Compact(rev int64) map[reversion]struct{} Compact(rev int64) map[revision]struct{}
Equal(b index) bool Equal(b index) bool
} }
@ -28,7 +28,7 @@ func newTreeIndex() index {
} }
} }
func (ti *treeIndex) Put(key []byte, rev reversion) { func (ti *treeIndex) Put(key []byte, rev revision) {
keyi := &keyIndex{key: key} keyi := &keyIndex{key: key}
ti.Lock() ti.Lock()
@ -43,7 +43,7 @@ func (ti *treeIndex) Put(key []byte, rev reversion) {
okeyi.put(rev.main, rev.sub) okeyi.put(rev.main, rev.sub)
} }
func (ti *treeIndex) Restore(key []byte, created, modified reversion, ver int64) { func (ti *treeIndex) Restore(key []byte, created, modified revision, ver int64) {
keyi := &keyIndex{key: key} keyi := &keyIndex{key: key}
ti.Lock() ti.Lock()
@ -58,27 +58,27 @@ func (ti *treeIndex) Restore(key []byte, created, modified reversion, ver int64)
okeyi.put(modified.main, modified.sub) okeyi.put(modified.main, modified.sub)
} }
func (ti *treeIndex) Get(key []byte, atRev int64) (modified, created reversion, ver int64, err error) { func (ti *treeIndex) Get(key []byte, atRev int64) (modified, created revision, ver int64, err error) {
keyi := &keyIndex{key: key} keyi := &keyIndex{key: key}
ti.RLock() ti.RLock()
defer ti.RUnlock() defer ti.RUnlock()
item := ti.tree.Get(keyi) item := ti.tree.Get(keyi)
if item == nil { if item == nil {
return reversion{}, reversion{}, 0, ErrReversionNotFound return revision{}, revision{}, 0, ErrRevisionNotFound
} }
keyi = item.(*keyIndex) keyi = item.(*keyIndex)
return keyi.get(atRev) return keyi.get(atRev)
} }
func (ti *treeIndex) Range(key, end []byte, atRev int64) (keys [][]byte, revs []reversion) { func (ti *treeIndex) Range(key, end []byte, atRev int64) (keys [][]byte, revs []revision) {
if end == nil { if end == nil {
rev, _, _, err := ti.Get(key, atRev) rev, _, _, err := ti.Get(key, atRev)
if err != nil { if err != nil {
return nil, nil return nil, nil
} }
return [][]byte{key}, []reversion{rev} return [][]byte{key}, []revision{rev}
} }
keyi := &keyIndex{key: key} keyi := &keyIndex{key: key}
@ -104,14 +104,14 @@ func (ti *treeIndex) Range(key, end []byte, atRev int64) (keys [][]byte, revs []
return keys, revs return keys, revs
} }
func (ti *treeIndex) Tombstone(key []byte, rev reversion) error { func (ti *treeIndex) Tombstone(key []byte, rev revision) error {
keyi := &keyIndex{key: key} keyi := &keyIndex{key: key}
ti.Lock() ti.Lock()
defer ti.Unlock() defer ti.Unlock()
item := ti.tree.Get(keyi) item := ti.tree.Get(keyi)
if item == nil { if item == nil {
return ErrReversionNotFound return ErrRevisionNotFound
} }
ki := item.(*keyIndex) ki := item.(*keyIndex)
@ -119,8 +119,8 @@ func (ti *treeIndex) Tombstone(key []byte, rev reversion) error {
return nil return nil
} }
func (ti *treeIndex) Compact(rev int64) map[reversion]struct{} { func (ti *treeIndex) Compact(rev int64) map[revision]struct{} {
available := make(map[reversion]struct{}) available := make(map[revision]struct{})
emptyki := make([]*keyIndex, 0) emptyki := make([]*keyIndex, 0)
log.Printf("store.index: compact %d", rev) log.Printf("store.index: compact %d", rev)
// TODO: do not hold the lock for long time? // TODO: do not hold the lock for long time?
@ -137,7 +137,7 @@ func (ti *treeIndex) Compact(rev int64) map[reversion]struct{} {
return available return available
} }
func compactIndex(rev int64, available map[reversion]struct{}, emptyki *[]*keyIndex) func(i btree.Item) bool { func compactIndex(rev int64, available map[revision]struct{}, emptyki *[]*keyIndex) func(i btree.Item) bool {
return func(i btree.Item) bool { return func(i btree.Item) bool {
keyi := i.(*keyIndex) keyi := i.(*keyIndex)
keyi.compact(rev, available) keyi.compact(rev, available)

100
storage/index_test.go
View File

@ -9,20 +9,20 @@ func TestIndexPutAndGet(t *testing.T) {
index := newTestTreeIndex() index := newTestTreeIndex()
tests := []T{ tests := []T{
{[]byte("foo"), 0, ErrReversionNotFound, 0}, {[]byte("foo"), 0, ErrRevisionNotFound, 0},
{[]byte("foo"), 1, nil, 1}, {[]byte("foo"), 1, nil, 1},
{[]byte("foo"), 3, nil, 1}, {[]byte("foo"), 3, nil, 1},
{[]byte("foo"), 5, nil, 5}, {[]byte("foo"), 5, nil, 5},
{[]byte("foo"), 6, nil, 5}, {[]byte("foo"), 6, nil, 5},
{[]byte("foo1"), 0, ErrReversionNotFound, 0}, {[]byte("foo1"), 0, ErrRevisionNotFound, 0},
{[]byte("foo1"), 1, ErrReversionNotFound, 0}, {[]byte("foo1"), 1, ErrRevisionNotFound, 0},
{[]byte("foo1"), 2, nil, 2}, {[]byte("foo1"), 2, nil, 2},
{[]byte("foo1"), 5, nil, 2}, {[]byte("foo1"), 5, nil, 2},
{[]byte("foo1"), 6, nil, 6}, {[]byte("foo1"), 6, nil, 6},
{[]byte("foo2"), 0, ErrReversionNotFound, 0}, {[]byte("foo2"), 0, ErrRevisionNotFound, 0},
{[]byte("foo2"), 1, ErrReversionNotFound, 0}, {[]byte("foo2"), 1, ErrRevisionNotFound, 0},
{[]byte("foo2"), 3, nil, 3}, {[]byte("foo2"), 3, nil, 3},
{[]byte("foo2"), 4, nil, 4}, {[]byte("foo2"), 4, nil, 4},
{[]byte("foo2"), 6, nil, 4}, {[]byte("foo2"), 6, nil, 4},
@ -33,11 +33,11 @@ func TestIndexPutAndGet(t *testing.T) {
func TestIndexRange(t *testing.T) { func TestIndexRange(t *testing.T) {
atRev := int64(3) atRev := int64(3)
allKeys := [][]byte{[]byte("foo"), []byte("foo1"), []byte("foo2")} allKeys := [][]byte{[]byte("foo"), []byte("foo1"), []byte("foo2")}
allRevs := []reversion{{main: 1}, {main: 2}, {main: 3}} allRevs := []revision{{main: 1}, {main: 2}, {main: 3}}
tests := []struct { tests := []struct {
key, end []byte key, end []byte
wkeys [][]byte wkeys [][]byte
wrevs []reversion wrevs []revision
}{ }{
// single key that not found // single key that not found
{ {
@ -87,7 +87,7 @@ func TestIndexRange(t *testing.T) {
func TestIndexTombstone(t *testing.T) { func TestIndexTombstone(t *testing.T) {
index := newTestTreeIndex() index := newTestTreeIndex()
err := index.Tombstone([]byte("foo"), reversion{main: 7}) err := index.Tombstone([]byte("foo"), revision{main: 7})
if err != nil { if err != nil {
t.Errorf("tombstone error = %v, want nil", err) t.Errorf("tombstone error = %v, want nil", err)
} }
@ -95,9 +95,9 @@ func TestIndexTombstone(t *testing.T) {
if err != nil { if err != nil {
t.Errorf("get error = %v, want nil", err) t.Errorf("get error = %v, want nil", err)
} }
w := reversion{main: 7} w := revision{main: 7}
if !reflect.DeepEqual(rev, w) { if !reflect.DeepEqual(rev, w) {
t.Errorf("get reversion = %+v, want %+v", rev, w) t.Errorf("get revision = %+v, want %+v", rev, w)
} }
} }
@ -105,26 +105,26 @@ func TestContinuousCompact(t *testing.T) {
index := newTestTreeIndex() index := newTestTreeIndex()
tests := []T{ tests := []T{
{[]byte("foo"), 0, ErrReversionNotFound, 0}, {[]byte("foo"), 0, ErrRevisionNotFound, 0},
{[]byte("foo"), 1, nil, 1}, {[]byte("foo"), 1, nil, 1},
{[]byte("foo"), 3, nil, 1}, {[]byte("foo"), 3, nil, 1},
{[]byte("foo"), 5, nil, 5}, {[]byte("foo"), 5, nil, 5},
{[]byte("foo"), 6, nil, 5}, {[]byte("foo"), 6, nil, 5},
{[]byte("foo1"), 0, ErrReversionNotFound, 0}, {[]byte("foo1"), 0, ErrRevisionNotFound, 0},
{[]byte("foo1"), 1, ErrReversionNotFound, 0}, {[]byte("foo1"), 1, ErrRevisionNotFound, 0},
{[]byte("foo1"), 2, nil, 2}, {[]byte("foo1"), 2, nil, 2},
{[]byte("foo1"), 5, nil, 2}, {[]byte("foo1"), 5, nil, 2},
{[]byte("foo1"), 6, nil, 6}, {[]byte("foo1"), 6, nil, 6},
{[]byte("foo2"), 0, ErrReversionNotFound, 0}, {[]byte("foo2"), 0, ErrRevisionNotFound, 0},
{[]byte("foo2"), 1, ErrReversionNotFound, 0}, {[]byte("foo2"), 1, ErrRevisionNotFound, 0},
{[]byte("foo2"), 3, nil, 3}, {[]byte("foo2"), 3, nil, 3},
{[]byte("foo2"), 4, nil, 4}, {[]byte("foo2"), 4, nil, 4},
{[]byte("foo2"), 6, nil, 4}, {[]byte("foo2"), 6, nil, 4},
} }
wa := map[reversion]struct{}{ wa := map[revision]struct{}{
reversion{main: 1}: struct{}{}, revision{main: 1}: struct{}{},
} }
ga := index.Compact(1) ga := index.Compact(1)
if !reflect.DeepEqual(ga, wa) { if !reflect.DeepEqual(ga, wa) {
@ -132,9 +132,9 @@ func TestContinuousCompact(t *testing.T) {
} }
verify(t, index, tests) verify(t, index, tests)
wa = map[reversion]struct{}{ wa = map[revision]struct{}{
reversion{main: 1}: struct{}{}, revision{main: 1}: struct{}{},
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
} }
ga = index.Compact(2) ga = index.Compact(2)
if !reflect.DeepEqual(ga, wa) { if !reflect.DeepEqual(ga, wa) {
@ -142,10 +142,10 @@ func TestContinuousCompact(t *testing.T) {
} }
verify(t, index, tests) verify(t, index, tests)
wa = map[reversion]struct{}{ wa = map[revision]struct{}{
reversion{main: 1}: struct{}{}, revision{main: 1}: struct{}{},
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
reversion{main: 3}: struct{}{}, revision{main: 3}: struct{}{},
} }
ga = index.Compact(3) ga = index.Compact(3)
if !reflect.DeepEqual(ga, wa) { if !reflect.DeepEqual(ga, wa) {
@ -153,45 +153,45 @@ func TestContinuousCompact(t *testing.T) {
} }
verify(t, index, tests) verify(t, index, tests)
wa = map[reversion]struct{}{ wa = map[revision]struct{}{
reversion{main: 1}: struct{}{}, revision{main: 1}: struct{}{},
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
reversion{main: 4}: struct{}{}, revision{main: 4}: struct{}{},
} }
ga = index.Compact(4) ga = index.Compact(4)
delete(wa, reversion{main: 3}) delete(wa, revision{main: 3})
tests[12] = T{[]byte("foo2"), 3, ErrReversionNotFound, 0} tests[12] = T{[]byte("foo2"), 3, ErrRevisionNotFound, 0}
if !reflect.DeepEqual(wa, ga) { if !reflect.DeepEqual(wa, ga) {
t.Errorf("a = %v, want %v", ga, wa) t.Errorf("a = %v, want %v", ga, wa)
} }
verify(t, index, tests) verify(t, index, tests)
wa = map[reversion]struct{}{ wa = map[revision]struct{}{
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
reversion{main: 4}: struct{}{}, revision{main: 4}: struct{}{},
reversion{main: 5}: struct{}{}, revision{main: 5}: struct{}{},
} }
ga = index.Compact(5) ga = index.Compact(5)
delete(wa, reversion{main: 1}) delete(wa, revision{main: 1})
if !reflect.DeepEqual(ga, wa) { if !reflect.DeepEqual(ga, wa) {
t.Errorf("a = %v, want %v", ga, wa) t.Errorf("a = %v, want %v", ga, wa)
} }
tests[1] = T{[]byte("foo"), 1, ErrReversionNotFound, 0} tests[1] = T{[]byte("foo"), 1, ErrRevisionNotFound, 0}
tests[2] = T{[]byte("foo"), 3, ErrReversionNotFound, 0} tests[2] = T{[]byte("foo"), 3, ErrRevisionNotFound, 0}
verify(t, index, tests) verify(t, index, tests)
wa = map[reversion]struct{}{ wa = map[revision]struct{}{
reversion{main: 4}: struct{}{}, revision{main: 4}: struct{}{},
reversion{main: 5}: struct{}{}, revision{main: 5}: struct{}{},
reversion{main: 6}: struct{}{}, revision{main: 6}: struct{}{},
} }
ga = index.Compact(6) ga = index.Compact(6)
delete(wa, reversion{main: 2}) delete(wa, revision{main: 2})
if !reflect.DeepEqual(ga, wa) { if !reflect.DeepEqual(ga, wa) {
t.Errorf("a = %v, want %v", ga, wa) t.Errorf("a = %v, want %v", ga, wa)
} }
tests[7] = T{[]byte("foo1"), 2, ErrReversionNotFound, 0} tests[7] = T{[]byte("foo1"), 2, ErrRevisionNotFound, 0}
tests[8] = T{[]byte("foo1"), 5, ErrReversionNotFound, 0} tests[8] = T{[]byte("foo1"), 5, ErrRevisionNotFound, 0}
verify(t, index, tests) verify(t, index, tests)
} }
@ -217,11 +217,11 @@ type T struct {
func newTestTreeIndex() index { func newTestTreeIndex() index {
index := newTreeIndex() index := newTreeIndex()
index.Put([]byte("foo"), reversion{main: 1}) index.Put([]byte("foo"), revision{main: 1})
index.Put([]byte("foo1"), reversion{main: 2}) index.Put([]byte("foo1"), revision{main: 2})
index.Put([]byte("foo2"), reversion{main: 3}) index.Put([]byte("foo2"), revision{main: 3})
index.Put([]byte("foo2"), reversion{main: 4}) index.Put([]byte("foo2"), revision{main: 4})
index.Put([]byte("foo"), reversion{main: 5}) index.Put([]byte("foo"), revision{main: 5})
index.Put([]byte("foo1"), reversion{main: 6}) index.Put([]byte("foo1"), revision{main: 6})
return index return index
} }

46
storage/key_index.go
View File

@ -10,10 +10,10 @@ import (
) )
var ( var (
ErrReversionNotFound = errors.New("stroage: reversion not found") ErrRevisionNotFound = errors.New("stroage: revision not found")
) )
// keyIndex stores the reversion of an key in the backend. // keyIndex stores the revision of an key in the backend.
// Each keyIndex has at least one key generation. // Each keyIndex has at least one key generation.
// Each generation might have several key versions. // Each generation might have several key versions.
// Tombstone on a key appends an tombstone version at the end // Tombstone on a key appends an tombstone version at the end
@ -55,16 +55,16 @@ var (
// {empty} -> key SHOULD be removed. // {empty} -> key SHOULD be removed.
type keyIndex struct { type keyIndex struct {
key []byte key []byte
modified reversion // the main rev of the last modification modified revision // the main rev of the last modification
generations []generation generations []generation
} }
// put puts a reversion to the keyIndex. // put puts a revision to the keyIndex.
func (ki *keyIndex) put(main int64, sub int64) { func (ki *keyIndex) put(main int64, sub int64) {
rev := reversion{main: main, sub: sub} rev := revision{main: main, sub: sub}
if !rev.GreaterThan(ki.modified) { if !rev.GreaterThan(ki.modified) {
log.Panicf("store.keyindex: put with unexpected smaller reversion [%v / %v]", rev, ki.modified) log.Panicf("store.keyindex: put with unexpected smaller revision [%v / %v]", rev, ki.modified)
} }
if len(ki.generations) == 0 { if len(ki.generations) == 0 {
ki.generations = append(ki.generations, generation{}) ki.generations = append(ki.generations, generation{})
@ -78,17 +78,17 @@ func (ki *keyIndex) put(main int64, sub int64) {
ki.modified = rev ki.modified = rev
} }
func (ki *keyIndex) restore(created, modified reversion, ver int64) { func (ki *keyIndex) restore(created, modified revision, ver int64) {
if len(ki.generations) != 0 { if len(ki.generations) != 0 {
log.Panicf("store.keyindex: cannot restore non-empty keyIndex") log.Panicf("store.keyindex: cannot restore non-empty keyIndex")
} }
ki.modified = modified ki.modified = modified
g := generation{created: created, ver: ver, revs: []reversion{modified}} g := generation{created: created, ver: ver, revs: []revision{modified}}
ki.generations = append(ki.generations, g) ki.generations = append(ki.generations, g)
} }
// tombstone puts a reversion, pointing to a tombstone, to the keyIndex. // tombstone puts a revision, pointing to a tombstone, to the keyIndex.
// It also creates a new empty generation in the keyIndex. // It also creates a new empty generation in the keyIndex.
func (ki *keyIndex) tombstone(main int64, sub int64) { func (ki *keyIndex) tombstone(main int64, sub int64) {
if ki.isEmpty() { if ki.isEmpty() {
@ -98,18 +98,18 @@ func (ki *keyIndex) tombstone(main int64, sub int64) {
ki.generations = append(ki.generations, generation{}) ki.generations = append(ki.generations, generation{})
} }
// get gets the modified, created reversion and version of the key that satisfies the given atRev. // get gets the modified, created revision and version of the key that satisfies the given atRev.
// Rev must be higher than or equal to the given atRev. // Rev must be higher than or equal to the given atRev.
func (ki *keyIndex) get(atRev int64) (modified, created reversion, ver int64, err error) { func (ki *keyIndex) get(atRev int64) (modified, created revision, ver int64, err error) {
if ki.isEmpty() { if ki.isEmpty() {
log.Panicf("store.keyindex: unexpected get on empty keyIndex %s", string(ki.key)) log.Panicf("store.keyindex: unexpected get on empty keyIndex %s", string(ki.key))
} }
g := ki.findGeneration(atRev) g := ki.findGeneration(atRev)
if g.isEmpty() { if g.isEmpty() {
return reversion{}, reversion{}, 0, ErrReversionNotFound return revision{}, revision{}, 0, ErrRevisionNotFound
} }
f := func(rev reversion) bool { f := func(rev revision) bool {
if rev.main <= atRev { if rev.main <= atRev {
return false return false
} }
@ -121,22 +121,22 @@ func (ki *keyIndex) get(atRev int64) (modified, created reversion, ver int64, er
return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil
} }
return reversion{}, reversion{}, 0, ErrReversionNotFound return revision{}, revision{}, 0, ErrRevisionNotFound
} }
// compact compacts a keyIndex by removing the versions with smaller or equal // compact compacts a keyIndex by removing the versions with smaller or equal
// reversion than the given atRev except the largest one (If the largest one is // revision than the given atRev except the largest one (If the largest one is
// a tombstone, it will not be kept). // a tombstone, it will not be kept).
// If a generation becomes empty during compaction, it will be removed. // If a generation becomes empty during compaction, it will be removed.
func (ki *keyIndex) compact(atRev int64, available map[reversion]struct{}) { func (ki *keyIndex) compact(atRev int64, available map[revision]struct{}) {
if ki.isEmpty() { if ki.isEmpty() {
log.Panic("store.keyindex: unexpected compact on empty keyIndex %s", string(ki.key)) log.Panic("store.keyindex: unexpected compact on empty keyIndex %s", string(ki.key))
} }
// walk until reaching the first reversion that has an reversion smaller or equal to // walk until reaching the first revision that has an revision smaller or equal to
// the atReversion. // the atRevision.
// add it to the available map // add it to the available map
f := func(rev reversion) bool { f := func(rev revision) bool {
if rev.main <= atRev { if rev.main <= atRev {
available[rev] = struct{}{} available[rev] = struct{}{}
return false return false
@ -231,18 +231,18 @@ func (ki *keyIndex) String() string {
type generation struct { type generation struct {
ver int64 ver int64
created reversion // when the generation is created (put in first reversion). created revision // when the generation is created (put in first revision).
revs []reversion revs []revision
} }
func (g *generation) isEmpty() bool { return g == nil || len(g.revs) == 0 } func (g *generation) isEmpty() bool { return g == nil || len(g.revs) == 0 }
// walk walks through the reversions in the generation in ascending order. // walk walks through the revisions in the generation in ascending order.
// It passes the revision to the given function. // It passes the revision to the given function.
// walk returns until: 1. it finishs walking all pairs 2. the function returns false. // walk returns until: 1. it finishs walking all pairs 2. the function returns false.
// walk returns the position at where it stopped. If it stopped after // walk returns the position at where it stopped. If it stopped after
// finishing walking, -1 will be returned. // finishing walking, -1 will be returned.
func (g *generation) walk(f func(rev reversion) bool) int { func (g *generation) walk(f func(rev revision) bool) int {
l := len(g.revs) l := len(g.revs)
for i := range g.revs { for i := range g.revs {
ok := f(g.revs[l-i-1]) ok := f(g.revs[l-i-1])

124
storage/key_index_test.go
View File

@ -13,7 +13,7 @@ func TestKeyIndexGet(t *testing.T) {
// {8[1], 10[2], 12(t)[3]} // {8[1], 10[2], 12(t)[3]}
// {4[2], 6(t)[3]} // {4[2], 6(t)[3]}
ki := newTestKeyIndex() ki := newTestKeyIndex()
ki.compact(4, make(map[reversion]struct{})) ki.compact(4, make(map[revision]struct{}))
tests := []struct { tests := []struct {
rev int64 rev int64
@ -55,8 +55,8 @@ func TestKeyIndexPut(t *testing.T) {
wki := &keyIndex{ wki := &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{5, 0}, modified: revision{5, 0},
generations: []generation{{created: reversion{5, 0}, ver: 1, revs: []reversion{{main: 5}}}}, generations: []generation{{created: revision{5, 0}, ver: 1, revs: []revision{{main: 5}}}},
} }
if !reflect.DeepEqual(ki, wki) { if !reflect.DeepEqual(ki, wki) {
t.Errorf("ki = %+v, want %+v", ki, wki) t.Errorf("ki = %+v, want %+v", ki, wki)
@ -66,8 +66,8 @@ func TestKeyIndexPut(t *testing.T) {
wki = &keyIndex{ wki = &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{7, 0}, modified: revision{7, 0},
generations: []generation{{created: reversion{5, 0}, ver: 2, revs: []reversion{{main: 5}, {main: 7}}}}, generations: []generation{{created: revision{5, 0}, ver: 2, revs: []revision{{main: 5}, {main: 7}}}},
} }
if !reflect.DeepEqual(ki, wki) { if !reflect.DeepEqual(ki, wki) {
t.Errorf("ki = %+v, want %+v", ki, wki) t.Errorf("ki = %+v, want %+v", ki, wki)
@ -82,8 +82,8 @@ func TestKeyIndexTombstone(t *testing.T) {
wki := &keyIndex{ wki := &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{7, 0}, modified: revision{7, 0},
generations: []generation{{created: reversion{5, 0}, ver: 2, revs: []reversion{{main: 5}, {main: 7}}}, {}}, generations: []generation{{created: revision{5, 0}, ver: 2, revs: []revision{{main: 5}, {main: 7}}}, {}},
} }
if !reflect.DeepEqual(ki, wki) { if !reflect.DeepEqual(ki, wki) {
t.Errorf("ki = %+v, want %+v", ki, wki) t.Errorf("ki = %+v, want %+v", ki, wki)
@ -95,10 +95,10 @@ func TestKeyIndexTombstone(t *testing.T) {
wki = &keyIndex{ wki = &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{15, 0}, modified: revision{15, 0},
generations: []generation{ generations: []generation{
{created: reversion{5, 0}, ver: 2, revs: []reversion{{main: 5}, {main: 7}}}, {created: revision{5, 0}, ver: 2, revs: []revision{{main: 5}, {main: 7}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 9}, {main: 15}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 9}, {main: 15}}},
{}, {},
}, },
} }
@ -112,176 +112,176 @@ func TestKeyIndexCompact(t *testing.T) {
compact int64 compact int64
wki *keyIndex wki *keyIndex
wam map[reversion]struct{} wam map[revision]struct{}
}{ }{
{ {
1, 1,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{2, 0}, ver: 3, revs: []reversion{{main: 2}, {main: 4}, {main: 6}}}, {created: revision{2, 0}, ver: 3, revs: []revision{{main: 2}, {main: 4}, {main: 6}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{}, map[revision]struct{}{},
}, },
{ {
2, 2,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{2, 0}, ver: 3, revs: []reversion{{main: 2}, {main: 4}, {main: 6}}}, {created: revision{2, 0}, ver: 3, revs: []revision{{main: 2}, {main: 4}, {main: 6}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
}, },
}, },
{ {
3, 3,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{2, 0}, ver: 3, revs: []reversion{{main: 2}, {main: 4}, {main: 6}}}, {created: revision{2, 0}, ver: 3, revs: []revision{{main: 2}, {main: 4}, {main: 6}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 2}: struct{}{}, revision{main: 2}: struct{}{},
}, },
}, },
{ {
4, 4,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{2, 0}, ver: 3, revs: []reversion{{main: 4}, {main: 6}}}, {created: revision{2, 0}, ver: 3, revs: []revision{{main: 4}, {main: 6}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 4}: struct{}{}, revision{main: 4}: struct{}{},
}, },
}, },
{ {
5, 5,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{2, 0}, ver: 3, revs: []reversion{{main: 4}, {main: 6}}}, {created: revision{2, 0}, ver: 3, revs: []revision{{main: 4}, {main: 6}}},
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 4}: struct{}{}, revision{main: 4}: struct{}{},
}, },
}, },
{ {
6, 6,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{}, map[revision]struct{}{},
}, },
{ {
7, 7,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{}, map[revision]struct{}{},
}, },
{ {
8, 8,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 8}: struct{}{}, revision{main: 8}: struct{}{},
}, },
}, },
{ {
9, 9,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 8}, {main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 8}, {main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 8}: struct{}{}, revision{main: 8}: struct{}{},
}, },
}, },
{ {
10, 10,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 10}: struct{}{}, revision{main: 10}: struct{}{},
}, },
}, },
{ {
11, 11,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{ generations: []generation{
{created: reversion{8, 0}, ver: 3, revs: []reversion{{main: 10}, {main: 12}}}, {created: revision{8, 0}, ver: 3, revs: []revision{{main: 10}, {main: 12}}},
{}, {},
}, },
}, },
map[reversion]struct{}{ map[revision]struct{}{
reversion{main: 10}: struct{}{}, revision{main: 10}: struct{}{},
}, },
}, },
{ {
12, 12,
&keyIndex{ &keyIndex{
key: []byte("foo"), key: []byte("foo"),
modified: reversion{12, 0}, modified: revision{12, 0},
generations: []generation{{}}, generations: []generation{{}},
}, },
map[reversion]struct{}{}, map[revision]struct{}{},
}, },
} }
// Continous Compaction // Continous Compaction
ki := newTestKeyIndex() ki := newTestKeyIndex()
for i, tt := range tests { for i, tt := range tests {
am := make(map[reversion]struct{}) am := make(map[revision]struct{})
ki.compact(tt.compact, am) ki.compact(tt.compact, am)
if !reflect.DeepEqual(ki, tt.wki) { if !reflect.DeepEqual(ki, tt.wki) {
t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki) t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki)
@ -294,7 +294,7 @@ func TestKeyIndexCompact(t *testing.T) {
// Jump Compaction // Jump Compaction
for i, tt := range tests { for i, tt := range tests {
if (i%2 == 0 && i < 6) && (i%2 == 1 && i > 6) { if (i%2 == 0 && i < 6) && (i%2 == 1 && i > 6) {
am := make(map[reversion]struct{}) am := make(map[revision]struct{})
ki.compact(tt.compact, am) ki.compact(tt.compact, am)
if !reflect.DeepEqual(ki, tt.wki) { if !reflect.DeepEqual(ki, tt.wki) {
t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki) t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki)
@ -308,7 +308,7 @@ func TestKeyIndexCompact(t *testing.T) {
// OnceCompaction // OnceCompaction
for i, tt := range tests { for i, tt := range tests {
ki := newTestKeyIndex() ki := newTestKeyIndex()
am := make(map[reversion]struct{}) am := make(map[revision]struct{})
ki.compact(tt.compact, am) ki.compact(tt.compact, am)
if !reflect.DeepEqual(ki, tt.wki) { if !reflect.DeepEqual(ki, tt.wki) {
t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki) t.Errorf("#%d: ki = %+v, want %+v", i, ki, tt.wki)

28
storage/kvstore.go
View File

@ -23,8 +23,8 @@ var (
finishedCompactKeyName = []byte("finishedCompactRev") finishedCompactKeyName = []byte("finishedCompactRev")
ErrTxnIDMismatch = errors.New("storage: txn id mismatch") ErrTxnIDMismatch = errors.New("storage: txn id mismatch")
ErrCompacted = errors.New("storage: required reversion has been compacted") ErrCompacted = errors.New("storage: required revision has been compacted")
ErrFutureRev = errors.New("storage: required reversion is a future reversion") ErrFutureRev = errors.New("storage: required revision is a future revision")
) )
type store struct { type store struct {
@ -33,8 +33,8 @@ type store struct {
b backend.Backend b backend.Backend
kvindex index kvindex index
currentRev reversion currentRev revision
// the main reversion of the last compaction // the main revision of the last compaction
compactMainRev int64 compactMainRev int64
tmu sync.Mutex // protect the txnID field tmu sync.Mutex // protect the txnID field
@ -52,7 +52,7 @@ func newStore(path string) *store {
s := &store{ s := &store{
b: backend.New(path, batchInterval, batchLimit), b: backend.New(path, batchInterval, batchLimit),
kvindex: newTreeIndex(), kvindex: newTreeIndex(),
currentRev: reversion{}, currentRev: revision{},
compactMainRev: -1, compactMainRev: -1,
stopc: make(chan struct{}), stopc: make(chan struct{}),
} }
@ -160,7 +160,7 @@ func (s *store) Compact(rev int64) error {
s.compactMainRev = rev s.compactMainRev = rev
rbytes := newRevBytes() rbytes := newRevBytes()
revToBytes(reversion{main: rev}, rbytes) revToBytes(revision{main: rev}, rbytes)
tx := s.b.BatchTx() tx := s.b.BatchTx()
tx.Lock() tx.Lock()
@ -184,8 +184,8 @@ func (s *store) Restore() error {
defer s.mu.Unlock() defer s.mu.Unlock()
min, max := newRevBytes(), newRevBytes() min, max := newRevBytes(), newRevBytes()
revToBytes(reversion{}, min) revToBytes(revision{}, min)
revToBytes(reversion{main: math.MaxInt64, sub: math.MaxInt64}, max) revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)
// restore index // restore index
tx := s.b.BatchTx() tx := s.b.BatchTx()
@ -209,14 +209,14 @@ func (s *store) Restore() error {
// restore index // restore index
switch e.Type { switch e.Type {
case storagepb.PUT: case storagepb.PUT:
s.kvindex.Restore(e.Kv.Key, reversion{e.Kv.CreateIndex, 0}, rev, e.Kv.Version) s.kvindex.Restore(e.Kv.Key, revision{e.Kv.CreateIndex, 0}, rev, e.Kv.Version)
case storagepb.DELETE: case storagepb.DELETE:
s.kvindex.Tombstone(e.Kv.Key, rev) s.kvindex.Tombstone(e.Kv.Key, rev)
default: default:
log.Panicf("storage: unexpected event type %s", e.Type) log.Panicf("storage: unexpected event type %s", e.Type)
} }
// update reversion // update revision
s.currentRev = rev s.currentRev = rev
} }
@ -308,7 +308,7 @@ func (s *store) put(key, value []byte, rev int64) {
} }
ibytes := newRevBytes() ibytes := newRevBytes()
revToBytes(reversion{main: rev, sub: s.currentRev.sub}, ibytes) revToBytes(revision{main: rev, sub: s.currentRev.sub}, ibytes)
ver = ver + 1 ver = ver + 1
event := storagepb.Event{ event := storagepb.Event{
@ -331,7 +331,7 @@ func (s *store) put(key, value []byte, rev int64) {
tx.Lock() tx.Lock()
defer tx.Unlock() defer tx.Unlock()
tx.UnsafePut(keyBucketName, ibytes, d) tx.UnsafePut(keyBucketName, ibytes, d)
s.kvindex.Put(key, reversion{main: rev, sub: s.currentRev.sub}) s.kvindex.Put(key, revision{main: rev, sub: s.currentRev.sub})
s.currentRev.sub += 1 s.currentRev.sub += 1
} }
@ -388,7 +388,7 @@ func (s *store) delete(key []byte, mainrev int64) bool {
} }
ibytes := newRevBytes() ibytes := newRevBytes()
revToBytes(reversion{main: mainrev, sub: s.currentRev.sub}, ibytes) revToBytes(revision{main: mainrev, sub: s.currentRev.sub}, ibytes)
event := storagepb.Event{ event := storagepb.Event{
Type: storagepb.DELETE, Type: storagepb.DELETE,
@ -403,7 +403,7 @@ func (s *store) delete(key []byte, mainrev int64) bool {
} }
tx.UnsafePut(keyBucketName, ibytes, d) tx.UnsafePut(keyBucketName, ibytes, d)
err = s.kvindex.Tombstone(key, reversion{main: mainrev, sub: s.currentRev.sub}) err = s.kvindex.Tombstone(key, revision{main: mainrev, sub: s.currentRev.sub})
if err != nil { if err != nil {
log.Fatalf("storage: cannot tombstone an existing key (%s): %v", string(key), err) log.Fatalf("storage: cannot tombstone an existing key (%s): %v", string(key), err)
} }

8
storage/kvstore_compaction.go
View File

@ -5,7 +5,7 @@ import (
"time" "time"
) )
func (s *store) scheduleCompaction(compactMainRev int64, keep map[reversion]struct{}) { func (s *store) scheduleCompaction(compactMainRev int64, keep map[revision]struct{}) {
defer s.wg.Done() defer s.wg.Done()
end := make([]byte, 8) end := make([]byte, 8)
binary.BigEndian.PutUint64(end, uint64(compactMainRev+1)) binary.BigEndian.PutUint64(end, uint64(compactMainRev+1))
@ -13,7 +13,7 @@ func (s *store) scheduleCompaction(compactMainRev int64, keep map[reversion]stru
batchsize := int64(10000) batchsize := int64(10000)
last := make([]byte, 8+1+8) last := make([]byte, 8+1+8)
for { for {
var rev reversion var rev revision
tx := s.b.BatchTx() tx := s.b.BatchTx()
tx.Lock() tx.Lock()
@ -28,14 +28,14 @@ func (s *store) scheduleCompaction(compactMainRev int64, keep map[reversion]stru
if len(keys) == 0 { if len(keys) == 0 {
rbytes := make([]byte, 8+1+8) rbytes := make([]byte, 8+1+8)
revToBytes(reversion{main: compactMainRev}, rbytes) revToBytes(revision{main: compactMainRev}, rbytes)
tx.UnsafePut(metaBucketName, finishedCompactKeyName, rbytes) tx.UnsafePut(metaBucketName, finishedCompactKeyName, rbytes)
tx.Unlock() tx.Unlock()
return return
} }
// update last // update last
revToBytes(reversion{main: rev.main, sub: rev.sub + 1}, last) revToBytes(revision{main: rev.main, sub: rev.sub + 1}, last)
tx.Unlock() tx.Unlock()
select { select {

8
storage/kvstore_test.go
View File

@ -407,9 +407,9 @@ func TestRestore(t *testing.T) {
s0.DeleteRange([]byte("foo3"), nil) s0.DeleteRange([]byte("foo3"), nil)
mink := newRevBytes() mink := newRevBytes()
revToBytes(reversion{main: 0, sub: 0}, mink) revToBytes(revision{main: 0, sub: 0}, mink)
maxk := newRevBytes() maxk := newRevBytes()
revToBytes(reversion{main: math.MaxInt64, sub: math.MaxInt64}, maxk) revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, maxk)
s0kvs, _, err := s0.rangeKeys(mink, maxk, 0, 0) s0kvs, _, err := s0.rangeKeys(mink, maxk, 0, 0)
if err != nil { if err != nil {
t.Fatalf("rangeKeys on s0 error (%v)", err) t.Fatalf("rangeKeys on s0 error (%v)", err)
@ -442,7 +442,7 @@ func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
// write scheduled compaction, but not do compaction // write scheduled compaction, but not do compaction
rbytes := newRevBytes() rbytes := newRevBytes()
revToBytes(reversion{main: 2}, rbytes) revToBytes(revision{main: 2}, rbytes)
tx := s0.b.BatchTx() tx := s0.b.BatchTx()
tx.Lock() tx.Lock()
tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes) tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
@ -462,7 +462,7 @@ func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
// check the key in backend is deleted // check the key in backend is deleted
revbytes := newRevBytes() revbytes := newRevBytes()
// TODO: compact should delete main=2 key too // TODO: compact should delete main=2 key too
revToBytes(reversion{main: 1}, revbytes) revToBytes(revision{main: 1}, revbytes)
tx = s1.b.BatchTx() tx = s1.b.BatchTx()
tx.Lock() tx.Lock()
ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0) ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)

10
storage/reversion.go → storage/revision.go
View File

@ -2,12 +2,12 @@ package storage
import "encoding/binary" import "encoding/binary"
type reversion struct { type revision struct {
main int64 main int64
sub int64 sub int64
} }
func (a reversion) GreaterThan(b reversion) bool { func (a revision) GreaterThan(b revision) bool {
if a.main > b.main { if a.main > b.main {
return true return true
} }
@ -21,14 +21,14 @@ func newRevBytes() []byte {
return make([]byte, 8+1+8) return make([]byte, 8+1+8)
} }
func revToBytes(rev reversion, bytes []byte) { func revToBytes(rev revision, bytes []byte) {
binary.BigEndian.PutUint64(bytes, uint64(rev.main)) binary.BigEndian.PutUint64(bytes, uint64(rev.main))
bytes[8] = '_' bytes[8] = '_'
binary.BigEndian.PutUint64(bytes[9:], uint64(rev.sub)) binary.BigEndian.PutUint64(bytes[9:], uint64(rev.sub))
} }
func bytesToRev(bytes []byte) reversion { func bytesToRev(bytes []byte) revision {
return reversion{ return revision{
main: int64(binary.BigEndian.Uint64(bytes[0:8])), main: int64(binary.BigEndian.Uint64(bytes[0:8])),
sub: int64(binary.BigEndian.Uint64(bytes[9:])), sub: int64(binary.BigEndian.Uint64(bytes[9:])),
} }

18
storage/reversion_test.go → storage/revision_test.go
View File

@ -7,17 +7,17 @@ import (
"testing" "testing"
) )
// TestReversion tests that reversion could be encoded to and decoded from // TestRevision tests that revision could be encoded to and decoded from
// bytes slice. Moreover, the lexicograph order of its byte slice representation // bytes slice. Moreover, the lexicograph order of its byte slice representation
// follows the order of (main, sub). // follows the order of (main, sub).
func TestReversion(t *testing.T) { func TestRevision(t *testing.T) {
tests := []reversion{ tests := []revision{
// order in (main, sub) // order in (main, sub)
reversion{}, revision{},
reversion{main: 1, sub: 0}, revision{main: 1, sub: 0},
reversion{main: 1, sub: 1}, revision{main: 1, sub: 1},
reversion{main: 2, sub: 0}, revision{main: 2, sub: 0},
reversion{main: math.MaxInt64, sub: math.MaxInt64}, revision{main: math.MaxInt64, sub: math.MaxInt64},
} }
bs := make([][]byte, len(tests)) bs := make([][]byte, len(tests))
@ -27,7 +27,7 @@ func TestReversion(t *testing.T) {
bs[i] = b bs[i] = b
if grev := bytesToRev(b); !reflect.DeepEqual(grev, tt) { if grev := bytesToRev(b); !reflect.DeepEqual(grev, tt) {
t.Errorf("#%d: reversion = %+v, want %+v", i, grev, tt) t.Errorf("#%d: revision = %+v, want %+v", i, grev, tt)
} }
} }