mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
Unify shared code (and constants) with cindex package.
This commit is contained in:
parent
f53b70facb
commit
e90504fe62
@ -36,6 +36,7 @@ import (
|
|||||||
"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
|
"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
|
||||||
"go.etcd.io/etcd/server/v3/etcdserver/api/v2error"
|
"go.etcd.io/etcd/server/v3/etcdserver/api/v2error"
|
||||||
"go.etcd.io/etcd/server/v3/etcdserver/api/v2store"
|
"go.etcd.io/etcd/server/v3/etcdserver/api/v2store"
|
||||||
|
"go.etcd.io/etcd/server/v3/etcdserver/cindex"
|
||||||
"go.etcd.io/etcd/server/v3/mvcc"
|
"go.etcd.io/etcd/server/v3/mvcc"
|
||||||
"go.etcd.io/etcd/server/v3/mvcc/backend"
|
"go.etcd.io/etcd/server/v3/mvcc/backend"
|
||||||
"go.etcd.io/etcd/server/v3/wal"
|
"go.etcd.io/etcd/server/v3/wal"
|
||||||
@ -91,7 +92,7 @@ func migrateCommandFunc(cmd *cobra.Command, args []string) {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
readKeys(reader, be)
|
readKeys(reader, be)
|
||||||
mvcc.UpdateConsistentIndex(be, index)
|
cindex.UpdateConsistentIndex(be.BatchTx(), index)
|
||||||
err := <-errc
|
err := <-errc
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println("failed to transform keys")
|
fmt.Println("failed to transform keys")
|
||||||
|
@ -80,11 +80,7 @@ func (ci *consistentIndex) UnsafeSave(tx backend.BatchTx) {
|
|||||||
// Never save 0 as it means that we didn't loaded the real index yet.
|
// Never save 0 as it means that we didn't loaded the real index yet.
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
bs := make([]byte, 8) // this is kept on stack (not heap) so its quick.
|
unsafeUpdateConsistentIndex(tx, index)
|
||||||
binary.BigEndian.PutUint64(bs, index)
|
|
||||||
// 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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ci *consistentIndex) SetBatchTx(tx backend.BatchTx) {
|
func (ci *consistentIndex) SetBatchTx(tx backend.BatchTx) {
|
||||||
@ -130,3 +126,22 @@ func ReadConsistentIndex(tx backend.ReadTx) uint64 {
|
|||||||
defer tx.Unlock()
|
defer tx.Unlock()
|
||||||
return unsafeReadConsistentIndex(tx)
|
return unsafeReadConsistentIndex(tx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func unsafeUpdateConsistentIndex(tx backend.BatchTx, index uint64) {
|
||||||
|
bs := make([]byte, 8) // this is kept on stack (not heap) so its quick.
|
||||||
|
binary.BigEndian.PutUint64(bs, index)
|
||||||
|
// 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)
|
||||||
|
}
|
||||||
|
|
||||||
|
func UpdateConsistentIndex(tx backend.BatchTx, index uint64) {
|
||||||
|
tx.Lock()
|
||||||
|
defer tx.Unlock()
|
||||||
|
|
||||||
|
oldi := unsafeReadConsistentIndex(tx)
|
||||||
|
if index <= oldi {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
unsafeUpdateConsistentIndex(tx, index)
|
||||||
|
}
|
||||||
|
@ -35,9 +35,8 @@ import (
|
|||||||
|
|
||||||
var (
|
var (
|
||||||
keyBucketName = []byte("key")
|
keyBucketName = []byte("key")
|
||||||
metaBucketName = []byte("meta")
|
metaBucketName = cindex.MetaBucketName
|
||||||
|
|
||||||
consistentIndexKeyName = []byte("consistent_index")
|
|
||||||
scheduledCompactKeyName = []byte("scheduledCompactRev")
|
scheduledCompactKeyName = []byte("scheduledCompactRev")
|
||||||
finishedCompactKeyName = []byte("finishedCompactRev")
|
finishedCompactKeyName = []byte("finishedCompactRev")
|
||||||
|
|
||||||
@ -128,7 +127,7 @@ func NewStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, ci cindex.Cons
|
|||||||
tx := s.b.BatchTx()
|
tx := s.b.BatchTx()
|
||||||
tx.Lock()
|
tx.Lock()
|
||||||
tx.UnsafeCreateBucket(keyBucketName)
|
tx.UnsafeCreateBucket(keyBucketName)
|
||||||
tx.UnsafeCreateBucket(metaBucketName)
|
cindex.UnsafeCreateMetaBucket(tx)
|
||||||
tx.Unlock()
|
tx.Unlock()
|
||||||
s.b.ForceCommit()
|
s.b.ForceCommit()
|
||||||
|
|
||||||
@ -308,7 +307,7 @@ func init() {
|
|||||||
DefaultIgnores = map[backend.IgnoreKey]struct{}{
|
DefaultIgnores = map[backend.IgnoreKey]struct{}{
|
||||||
// consistent index might be changed due to v2 internal sync, which
|
// consistent index might be changed due to v2 internal sync, which
|
||||||
// is not controllable by the user.
|
// is not controllable by the user.
|
||||||
{Bucket: string(metaBucketName), Key: string(consistentIndexKeyName)}: {},
|
{Bucket: string(metaBucketName), Key: string(cindex.ConsistentIndexKeyName)}: {},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -15,33 +15,12 @@
|
|||||||
package mvcc
|
package mvcc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/binary"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
|
|
||||||
"go.etcd.io/etcd/api/v3/mvccpb"
|
"go.etcd.io/etcd/api/v3/mvccpb"
|
||||||
"go.etcd.io/etcd/server/v3/mvcc/backend"
|
"go.etcd.io/etcd/server/v3/mvcc/backend"
|
||||||
)
|
)
|
||||||
|
|
||||||
func UpdateConsistentIndex(be backend.Backend, index uint64) {
|
|
||||||
tx := be.BatchTx()
|
|
||||||
tx.Lock()
|
|
||||||
defer tx.Unlock()
|
|
||||||
|
|
||||||
var oldi uint64
|
|
||||||
_, vs := tx.UnsafeRange(metaBucketName, consistentIndexKeyName, nil, 0)
|
|
||||||
if len(vs) != 0 {
|
|
||||||
oldi = binary.BigEndian.Uint64(vs[0])
|
|
||||||
}
|
|
||||||
|
|
||||||
if index <= oldi {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
bs := make([]byte, 8)
|
|
||||||
binary.BigEndian.PutUint64(bs, index)
|
|
||||||
tx.UnsafePut(metaBucketName, consistentIndexKeyName, bs)
|
|
||||||
}
|
|
||||||
|
|
||||||
func WriteKV(be backend.Backend, kv mvccpb.KeyValue) {
|
func WriteKV(be backend.Backend, kv mvccpb.KeyValue) {
|
||||||
ibytes := newRevBytes()
|
ibytes := newRevBytes()
|
||||||
revToBytes(revision{main: kv.ModRevision}, ibytes)
|
revToBytes(revision{main: kv.ModRevision}, ibytes)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user