*: allow fully concurrent large read

etcdserver/v3_server.go

@@ -86,8 +86,6 @@ type Authenticator interface {
 }
 
 func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
- defer warnOfExpensiveReadOnlyRangeRequest(s.getLogger(), time.Now(), r)
-
  if !r.Serializable {
  err := s.linearizableReadNotify(ctx)
  if err != nil {

mvcc/backend/backend.go

@@ -49,6 +49,11 @@ var (
 )
 
 type Backend interface {
+ // CommittedReadTx returns a non-blocking read tx that is suitable for large reads.
+ // CommittedReadTx call itself will not return until the current BatchTx gets committed to
+ // ensure consistency.
+ CommittedReadTx() ReadTx
+
  ReadTx() ReadTx
  BatchTx() BatchTx
 
@@ -97,6 +102,8 @@ type backend struct {
 
  readTx *readTx
 
+ concurrentReadTxCh chan chan ReadTx
+
  stopc chan struct{}
  donec chan struct{}
 
@@ -165,6 +172,8 @@ func newBackend(bcfg BackendConfig) *backend {
  buckets: make(map[string]*bolt.Bucket),
  },
 
+ concurrentReadTxCh: make(chan chan ReadTx),
+
  stopc: make(chan struct{}),
  donec: make(chan struct{}),
 
@@ -184,6 +193,12 @@ func (b *backend) BatchTx() BatchTx {
 
 func (b *backend) ReadTx() ReadTx { return b.readTx }
 
+func (b *backend) CommittedReadTx() ReadTx {
+ rch := make(chan ReadTx)
+ b.concurrentReadTxCh <- rch
+ return <-rch
+}
+
 // ForceCommit forces the current batching tx to commit.
 func (b *backend) ForceCommit() {
  b.batchTx.Commit()
@@ -301,6 +316,25 @@ func (b *backend) run() {
  b.batchTx.Commit()
  }
  t.Reset(b.batchInterval)
+ b.createConcurrentReadTxs()
+ }
+}
+
+func (b *backend) createConcurrentReadTxs() {
+ // do not allow too many concurrent read txs.
+ // TODO: improve this by having a global pending counter?
+ for i := 0; i < 100; i++ {
+ select {
+ case rch := <-b.concurrentReadTxCh:
+ rtx, err := b.db.Begin(false)
+ if err != nil {
+ plog.Fatalf("cannot begin read tx (%s)", err)
+ }
+ rch <- &concurrentReadTx{tx: rtx}
+ default:
+ // no more to create.
+ return
+ }
  }
 }
 

mvcc/backend/backend_test.go

@@ -300,6 +300,81 @@ func TestBackendWritebackForEach(t *testing.T) {
  }
 }
 
+// TestBackendConcurrentReadTx checks if the concurrent tx is created correctly.
+func TestBackendConcurrentReadTx(t *testing.T) {
+ b, tmpPath := NewTmpBackend(2*time.Second, 10000)
+ defer cleanup(b, tmpPath)
+
+ var rtx0 ReadTx
+ done := make(chan struct{})
+ go func() {
+ rtx0 = b.ConcurrentReadTx()
+ close(done)
+ }()
+
+ tx := b.BatchTx()
+ tx.Lock()
+ tx.UnsafeCreateBucket([]byte("key"))
+ for i := 0; i < 5; i++ {
+ k := []byte(fmt.Sprintf("%04d", i))
+ tx.UnsafePut([]byte("key"), k, []byte("bar"))
+ }
+ tx.Unlock()
+
+ select {
+ case <-done:
+ t.Fatal("concurrent read tx should block on the last batch tx!")
+ case <-time.After(time.Second):
+ }
+
+ select {
+ case <-done:
+ case <-time.After(4 * time.Second):
+ t.Fatal("commit the last batched tx should unblock concurrent tx!")
+ }
+
+ rtx0.Lock()
+ defer rtx0.Unlock()
+ ks, _ := rtx0.UnsafeRange([]byte("key"), []byte(fmt.Sprintf("%04d", 0)), []byte(fmt.Sprintf("%04d", 5)), 0)
+ if len(ks) != 5 {
+ t.Errorf("got %d keys, expect %d", len(ks), 5)
+ }
+
+ // test if we can create concurrent read while the previous read tx is still open
+ var rtx1 ReadTx
+ done = make(chan struct{})
+ go func() {
+ rtx1 = b.ConcurrentReadTx()
+ rtx1.Lock()
+ rtx1.UnsafeForEach([]byte(""), nil)
+ rtx1.Unlock()
+ close(done)
+ }()
+ select {
+ case <-done:
+ case <-time.After(4 * time.Second):
+ t.Fatal("cannot create concurrent read")
+ }
+
+ done = make(chan struct{})
+ // test if we can create concurrent write while the previous read tx is still open
+ go func() {
+ tx := b.BatchTx()
+ tx.Lock()
+ for i := 0; i < 5; i++ {
+ k := []byte(fmt.Sprintf("%04d", i))
+ tx.UnsafePut([]byte("key"), k, []byte("bar"))
+ }
+ tx.Unlock()
+ close(done)
+ }()
+ select {
+ case <-done:
+ case <-time.After(4 * time.Second):
+ t.Fatal("cannot create concurrent write")
+ }
+}
+
 func cleanup(b Backend, path string) {
  b.Close()
  os.Remove(path)

mvcc/backend/concurrent_read_tx.go

@@ -0,0 +1,38 @@
+// Copyright 2018 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 backend
+
+import (
+ bolt "github.com/coreos/bbolt"
+)
+
+type concurrentReadTx struct {
+ tx *bolt.Tx
+}
+
+func (rt *concurrentReadTx) Lock() {}
+func (rt *concurrentReadTx) Unlock() { rt.tx.Rollback() }
+
+func (rt *concurrentReadTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
+ bucket := rt.tx.Bucket(bucketName)
+ if bucket == nil {
+ plog.Fatalf("bucket %s does not exist", bucketName)
+ }
+ return unsafeRange(bucket.Cursor(), key, endKey, limit)
+}
+
+func (rt *concurrentReadTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
+ return unsafeForEach(rt.tx, bucketName, visitor)
+}

mvcc/index.go

@@ -91,10 +91,11 @@ func (ti *treeIndex) keyIndex(keyi *keyIndex) *keyIndex {
 func (ti *treeIndex) visit(key, end []byte, f func(ki *keyIndex)) {
  keyi, endi := &keyIndex{key: key}, &keyIndex{key: end}
 
- ti.RLock()
- defer ti.RUnlock()
+ ti.Lock()
+ clone := ti.tree.Clone()
+ ti.Unlock()
 
- ti.tree.AscendGreaterOrEqual(keyi, func(item btree.Item) bool {
+ clone.AscendGreaterOrEqual(keyi, func(item btree.Item) bool {
  if len(endi.key) > 0 && !item.Less(endi) {
  return false
  }

mvcc/kvstore_test.go

@@ -640,7 +640,7 @@ func TestTxnBlockBackendForceCommit(t *testing.T) {
  s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  defer os.Remove(tmpPath)
 
- txn := s.Read()
+ txn := s.Write()
 
  done := make(chan struct{})
  go func() {
@@ -742,6 +742,7 @@ type fakeBackend struct {
 
 func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
 func (b *fakeBackend) ReadTx() backend.ReadTx { return b.tx }
+func (b *fakeBackend) ConcurrentReadTx() 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 }

mvcc/kvstore_txn.go

@@ -21,9 +21,21 @@ import (
  "go.uber.org/zap"
 )
 
+const (
+ expensiveReadLimit = 1000
+ readonly = true
+ readwrite = false
+)
+
 type storeTxnRead struct {
- s *store
- tx backend.ReadTx
+ s *store
+ tx backend.ReadTx
+ txlocked bool
+
+ // for creating concurrent read tx when the read is expensive.
+ b backend.Backend
+ // is the transcation readonly?
+ ro bool
 
  firstRev int64
  rev int64
@@ -33,10 +45,15 @@ func (s *store) Read() TxnRead {
  s.mu.RLock()
  tx := s.b.ReadTx()
  s.revMu.RLock()
- tx.Lock()
  firstRev, rev := s.compactMainRev, s.currentRev
  s.revMu.RUnlock()
- return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev})
+ return newMetricsTxnRead(&storeTxnRead{
+ s: s,
+ tx: tx,
+ b: s.b,
+ ro: readonly,
+ firstRev: firstRev,
+ rev: rev})
 }
 
 func (tr *storeTxnRead) FirstRev() int64 { return tr.firstRev }
@@ -47,7 +64,9 @@ func (tr *storeTxnRead) Range(key, end []byte, ro RangeOptions) (r *RangeResult,
 }
 
 func (tr *storeTxnRead) End() {
- tr.tx.Unlock()
+ if tr.txlocked {
+ tr.tx.Unlock()
+ }
  tr.s.mu.RUnlock()
 }
 
@@ -64,10 +83,15 @@ func (s *store) Write() TxnWrite {
  tx := s.b.BatchTx()
  tx.Lock()
  tw := &storeTxnWrite{
- storeTxnRead: storeTxnRead{s, tx, 0, 0},
- tx: tx,
- beginRev: s.currentRev,
- changes: make([]mvccpb.KeyValue, 0, 4),
+ storeTxnRead: storeTxnRead{
+ s: s,
+ txlocked: true,
+ tx: tx,
+ ro: readwrite,
+ },
+ tx: tx,
+ beginRev: s.currentRev,
+ changes: make([]mvccpb.KeyValue, 0, 4),
  }
  return newMetricsTxnWrite(tw)
 }
@@ -134,6 +158,15 @@ func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions
  limit = len(revpairs)
  }
 
+ if limit > expensiveReadLimit && !tr.txlocked && tr.ro { // first expensive read in a read only transcation
+ // too many keys to range. upgrade the read transcation to concurrent read tx.
+ tr.tx = tr.b.CommittedReadTx()
+ }
+ if !tr.txlocked {
+ tr.tx.Lock()
+ tr.txlocked = true
+ }
+
  kvs := make([]mvccpb.KeyValue, limit)
  revBytes := newRevBytes()
  for i, revpair := range revpairs[:len(kvs)] {