speedup fdb (#3860)

pkg/meta/tkv.go

@@ -46,7 +46,7 @@ type kvtxn interface {
  scan(begin, end []byte, keysOnly bool, handler func(k, v []byte) bool)
  exist(prefix []byte) bool
  set(key, value []byte)
- append(key []byte, value []byte) []byte
+ append(key []byte, value []byte)
  incrBy(key []byte, value int64) int64
  delete(key []byte)
 }
@@ -3232,9 +3232,11 @@ func (m *kvMeta) LoadMeta(r io.Reader) error {
  go func() {
  defer wg.Done()
  var buffer []*pair
+ var total int
  for p := range kv {
  buffer = append(buffer, p)
- if len(buffer) >= batch {
+ total += len(p.key) + len(p.value)
+ if len(buffer) >= batch || total > 5<<20 {
  err := m.txn(func(tx *kvTxn) error {
  for _, p := range buffer {
  tx.set(p.key, p.value)
@@ -3245,6 +3247,7 @@ func (m *kvMeta) LoadMeta(r io.Reader) error {
  logger.Fatalf("write %d pairs: %s", len(buffer), err)
  }
  buffer = buffer[:0]
+ total = 0
  }
  }
  if len(buffer) > 0 {

pkg/meta/tkv_badger.go

@@ -112,10 +112,9 @@ func (tx *badgerTxn) set(key, value []byte) {
  }
 }
 
-func (tx *badgerTxn) append(key []byte, value []byte) []byte {
+func (tx *badgerTxn) append(key []byte, value []byte) {
  list := append(tx.get(key), value...)
  tx.set(key, list)
- return list
 }
 
 func (tx *badgerTxn) incrBy(key []byte, value int64) int64 {

pkg/meta/tkv_etcd.go

@@ -151,10 +151,9 @@ func (tx *etcdTxn) set(key, value []byte) {
  }
 }
 
-func (tx *etcdTxn) append(key []byte, value []byte) []byte {
+func (tx *etcdTxn) append(key []byte, value []byte) {
  new := append(tx.get(key), value...)
  tx.set(key, new)
- return new
 }
 
 func (tx *etcdTxn) incrBy(key []byte, value int64) int64 {

pkg/meta/tkv_fdb.go

@@ -52,6 +52,7 @@ func newFdbClient(addr string) (tkvClient, error) {
  if err != nil {
  return nil, fmt.Errorf("open database: %s", err)
  }
+ // TODO: database options
  return withPrefix(&fdbClient{db}, append([]byte(u.Query().Get("prefix")), 0xFD)), nil
 }
 
@@ -74,6 +75,7 @@ func (c *fdbClient) scan(prefix []byte, handler func(key, value []byte)) error {
  var done bool
  for {
  if _, err := c.client.ReadTransact(func(t fdb.ReadTransaction) (interface{}, error) {
+ // TODO: t.Options().SetPriorityBatch()
  snapshot := t.Snapshot()
  iter := snapshot.GetRange(
  fdb.KeyRange{Begin: begin, End: end},
@@ -113,7 +115,6 @@ func (c *fdbClient) reset(prefix []byte) error {
 }
 
 func (c *fdbClient) close() error {
- // c = &fdbClient{}
  return nil
 }
 
@@ -128,10 +129,13 @@ func (tx *fdbTxn) get(key []byte) []byte {
 }
 
 func (tx *fdbTxn) gets(keys ...[]byte) [][]byte {
- ret := make([][]byte, len(keys))
+ fut := make([]fdb.FutureByteSlice, len(keys))
  for i, key := range keys {
- val := tx.Get(fdb.Key(key)).MustGet()
- ret[i] = val
+ fut[i] = tx.Get(fdb.Key(key))
+ }
+ ret := make([][]byte, len(keys))
+ for i, f := range fut {
+ ret[i] = f.MustGet()
  }
  return ret
 }
@@ -158,13 +162,13 @@ func (tx *fdbTxn) set(key, value []byte) {
  tx.Set(fdb.Key(key), value)
 }
 
-func (tx *fdbTxn) append(key []byte, value []byte) []byte {
+func (tx *fdbTxn) append(key []byte, value []byte) {
  tx.AppendIfFits(fdb.Key(key), fdb.Key(value))
- return tx.Get(fdb.Key(key)).MustGet()
 }
 
 func (tx *fdbTxn) incrBy(key []byte, value int64) int64 {
  tx.Add(fdb.Key(key), packCounter(value))
+ // TODO: don't return new value if not needed
  return parseCounter(tx.Get(fdb.Key(key)).MustGet())
 }
 

pkg/meta/tkv_mem.go

@@ -131,10 +131,9 @@ func (tx *memTxn) set(key, value []byte) {
  tx.buffer[string(key)] = value
 }
 
-func (tx *memTxn) append(key []byte, value []byte) []byte {
+func (tx *memTxn) append(key []byte, value []byte) {
  new := append(tx.get(key), value...)
  tx.set(key, new)
- return new
 }
 
 func (tx *memTxn) incrBy(key []byte, value int64) int64 {

pkg/meta/tkv_prefix.go

@@ -60,8 +60,8 @@ func (tx *prefixTxn) set(key, value []byte) {
  tx.kvTxn.set(tx.realKey(key), value)
 }
 
-func (tx *prefixTxn) append(key []byte, value []byte) []byte {
- return tx.kvTxn.append(tx.realKey(key), value)
+func (tx *prefixTxn) append(key []byte, value []byte) {
+ tx.kvTxn.append(tx.realKey(key), value)
 }
 
 func (tx *prefixTxn) incrBy(key []byte, value int64) int64 {

pkg/meta/tkv_tikv.go

@@ -146,10 +146,9 @@ func (tx *tikvTxn) set(key, value []byte) {
  }
 }
 
-func (tx *tikvTxn) append(key []byte, value []byte) []byte {
+func (tx *tikvTxn) append(key []byte, value []byte) {
  new := append(tx.get(key), value...)
  tx.set(key, new)
- return new
 }
 
 func (tx *tikvTxn) incrBy(key []byte, value int64) int64 {