Skip to content

Commit e3d2d67

Browse files
serathiusserathius
committed
Use maxRequestSizeBytes for batch limiting
Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>
1 parent 803c62b commit e3d2d67

File tree

14 files changed

+80
-71
lines changed

14 files changed

+80
-71
lines changed

‎server/config/config.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -363,6 +363,6 @@ func (c *ServerConfig) BootstrapTimeoutEffective() time.Duration {
363363

364364
func (c *ServerConfig) BackendPath() string { return datadir.ToBackendFileName(c.DataDir) }
365365

366-
func (c *ServerConfig) MaxRequestBytesWithOverhead() uint {
367-
return c.MaxRequestBytes + grpcOverheadBytes
366+
func (c *ServerConfig) MaxRequestBytesWithOverhead() int {
367+
return int(c.MaxRequestBytes) + grpcOverheadBytes
368368
}

‎server/etcdserver/api/v3rpc/grpc.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ func Server(s *etcdserver.EtcdServer, tls *tls.Config, interceptor grpc.UnarySer
6161
opts = append(opts, grpc.ChainUnaryInterceptor(chainUnaryInterceptors...))
6262
opts = append(opts, grpc.ChainStreamInterceptor(chainStreamInterceptors...))
6363

64-
opts = append(opts, grpc.MaxRecvMsgSize(int(s.Cfg.MaxRequestBytesWithOverhead())))
64+
opts = append(opts, grpc.MaxRecvMsgSize(s.Cfg.MaxRequestBytesWithOverhead()))
6565
opts = append(opts, grpc.MaxSendMsgSize(maxSendBytes))
6666
opts = append(opts, grpc.MaxConcurrentStreams(s.Cfg.MaxConcurrentStreams))
6767

‎server/etcdserver/api/v3rpc/watch.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ type watchServer struct {
4343
clusterID int64
4444
memberID int64
4545

46-
maxRequestBytes uint
46+
maxRequestBytes int
4747

4848
sg apply.RaftStatusGetter
4949
watchable mvcc.WatchableKV
@@ -126,7 +126,7 @@ type serverWatchStream struct {
126126
clusterID int64
127127
memberID int64
128128

129-
maxRequestBytes uint
129+
maxRequestBytes int
130130

131131
sg apply.RaftStatusGetter
132132
watchable mvcc.WatchableKV
@@ -544,12 +544,12 @@ func IsCreateEvent(e mvccpb.Event) bool {
544544

545545
func sendFragments(
546546
wr *pb.WatchResponse,
547-
maxRequestBytes uint,
547+
maxRequestBytes int,
548548
sendFunc func(*pb.WatchResponse) error,
549549
) error {
550550
// no need to fragment if total request size is smaller
551551
// than max request limit or response contains only one event
552-
if uint(wr.Size()) < maxRequestBytes || len(wr.Events) < 2 {
552+
if wr.Size() < maxRequestBytes || len(wr.Events) < 2 {
553553
return sendFunc(wr)
554554
}
555555

@@ -562,7 +562,7 @@ func sendFragments(
562562
cur := ow
563563
for _, ev := range wr.Events[idx:] {
564564
cur.Events = append(cur.Events, ev)
565-
if len(cur.Events) > 1 && uint(cur.Size()) >= maxRequestBytes {
565+
if len(cur.Events) > 1 && cur.Size() >= maxRequestBytes {
566566
cur.Events = cur.Events[:len(cur.Events)-1]
567567
break
568568
}

‎server/etcdserver/api/v3rpc/watch_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import (
2727
func TestSendFragment(t *testing.T) {
2828
tt := []struct {
2929
wr *pb.WatchResponse
30-
maxRequestBytes uint
30+
maxRequestBytes int
3131
fragments int
3232
werr error
3333
}{

‎server/etcdserver/corrupt_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -516,7 +516,7 @@ func TestHashKVHandler(t *testing.T) {
516516
etcdSrv.cluster.SetID(types.ID(localClusterID), types.ID(localClusterID))
517517
be, _ := betesting.NewDefaultTmpBackend(t)
518518
defer betesting.Close(t, be)
519-
etcdSrv.kv = mvcc.New(zap.NewNop(), be, &lease.FakeLessor{}, mvcc.StoreConfig{})
519+
etcdSrv.kv = mvcc.New(zap.NewNop(), be, &lease.FakeLessor{}, mvcc.WatchableStoreConfig{})
520520
defer func() {
521521
assert.NoError(t, etcdSrv.kv.Close())
522522
}()

‎server/etcdserver/server.go‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -368,9 +368,12 @@ func NewServer(cfg config.ServerConfig) (srv *EtcdServer, err error) {
368368
return nil, err
369369
}
370370

371-
mvccStoreConfig := mvcc.StoreConfig{
372-
CompactionBatchLimit: cfg.CompactionBatchLimit,
373-
CompactionSleepInterval: cfg.CompactionSleepInterval,
371+
mvccStoreConfig := mvcc.WatchableStoreConfig{
372+
StoreConfig: mvcc.StoreConfig{
373+
CompactionBatchLimit: cfg.CompactionBatchLimit,
374+
CompactionSleepInterval: cfg.CompactionSleepInterval,
375+
},
376+
WatchBatchMaxSize: cfg.MaxRequestBytesWithOverhead(),
374377
}
375378
srv.kv = mvcc.New(srv.Logger(), srv.be, srv.lessor, mvccStoreConfig)
376379
srv.corruptionChecker = newCorruptionChecker(cfg.Logger, srv, srv.kv.HashStorage())

‎server/etcdserver/server_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -652,7 +652,7 @@ func TestSnapshotDisk(t *testing.T) {
652652
v2store: st,
653653
consistIndex: cindex.NewConsistentIndex(be),
654654
}
655-
srv.kv = mvcc.New(zaptest.NewLogger(t), be, &lease.FakeLessor{}, mvcc.StoreConfig{})
655+
srv.kv = mvcc.New(zaptest.NewLogger(t), be, &lease.FakeLessor{}, mvcc.WatchableStoreConfig{})
656656
defer func() {
657657
assert.NoError(t, srv.kv.Close())
658658
}()
@@ -703,7 +703,7 @@ func TestSnapshotMemory(t *testing.T) {
703703
v2store: st,
704704
consistIndex: cindex.NewConsistentIndex(be),
705705
}
706-
srv.kv = mvcc.New(zaptest.NewLogger(t), be, &lease.FakeLessor{}, mvcc.StoreConfig{})
706+
srv.kv = mvcc.New(zaptest.NewLogger(t), be, &lease.FakeLessor{}, mvcc.WatchableStoreConfig{})
707707
defer func() {
708708
assert.NoError(t, srv.kv.Close())
709709
}()
@@ -775,7 +775,7 @@ func TestSnapshotOrdering(t *testing.T) {
775775
beHooks: serverstorage.NewBackendHooks(lg, ci),
776776
}
777777

778-
s.kv = mvcc.New(lg, be, &lease.FakeLessor{}, mvcc.StoreConfig{})
778+
s.kv = mvcc.New(lg, be, &lease.FakeLessor{}, mvcc.WatchableStoreConfig{})
779779
s.be = be
780780

781781
s.start()
@@ -869,7 +869,7 @@ func TestConcurrentApplyAndSnapshotV3(t *testing.T) {
869869
authStore: auth.NewAuthStore(lg, schema.NewAuthBackend(lg, be), nil, 1),
870870
}
871871

872-
s.kv = mvcc.New(lg, be, &lease.FakeLessor{}, mvcc.StoreConfig{})
872+
s.kv = mvcc.New(lg, be, &lease.FakeLessor{}, mvcc.WatchableStoreConfig{})
873873
s.be = be
874874

875875
s.start()

‎server/storage/mvcc/kv_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -757,7 +757,7 @@ func TestKVSnapshot(t *testing.T) {
757757

758758
func TestWatchableKVWatch(t *testing.T) {
759759
b, _ := betesting.NewDefaultTmpBackend(t)
760-
s := New(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
760+
s := New(zaptest.NewLogger(t), b, &lease.FakeLessor{}, WatchableStoreConfig{})
761761
defer cleanup(s, b)
762762

763763
w := s.NewWatchStream()

‎server/storage/mvcc/watchable_store.go‎

Lines changed: 16 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ type watchable interface {
5050

5151
type watchableStore struct {
5252
*store
53+
watchBatchMaxSize int
5354

5455
// mu protects watcher groups and batches. It should never be locked
5556
// before locking store.mu to avoid deadlock.
@@ -76,24 +77,30 @@ var _ WatchableKV = (*watchableStore)(nil)
7677
// cancel operations.
7778
type cancelFunc func()
7879

79-
func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) *watchableStore {
80+
func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg WatchableStoreConfig) *watchableStore {
8081
s := newWatchableStore(lg, b, le, cfg)
8182
s.wg.Add(2)
8283
go s.syncWatchersLoop()
8384
go s.syncVictimsLoop()
8485
return s
8586
}
8687

87-
func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) *watchableStore {
88+
type WatchableStoreConfig struct {
89+
StoreConfig
90+
WatchBatchMaxSize int
91+
}
92+
93+
func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg WatchableStoreConfig) *watchableStore {
8894
if lg == nil {
8995
lg = zap.NewNop()
9096
}
9197
s := &watchableStore{
92-
store: NewStore(lg, b, le, cfg),
93-
victimc: make(chan struct{}, 1),
94-
unsynced: newWatcherGroup(),
95-
synced: newWatcherGroup(),
96-
stopc: make(chan struct{}),
98+
store: NewStore(lg, b, le, cfg.StoreConfig),
99+
victimc: make(chan struct{}, 1),
100+
unsynced: newWatcherGroup(),
101+
synced: newWatcherGroup(),
102+
stopc: make(chan struct{}),
103+
watchBatchMaxSize: cfg.WatchBatchMaxSize,
97104
}
98105
s.store.ReadView = &readView{s}
99106
s.store.WriteView = &writeView{s}
@@ -373,7 +380,7 @@ func (s *watchableStore) syncWatchers() int {
373380
tx.RUnlock()
374381

375382
victims := make(watcherBatch)
376-
wb := newWatcherBatch(wg, evs)
383+
wb := newWatcherBatch(wg, evs, s.watchBatchMaxSize)
377384
for w := range wg.watchers {
378385
if w.minRev < compactionRev {
379386
// Skip the watcher that failed to send compacted watch response due to w.ch is full.
@@ -449,7 +456,7 @@ func kvsToEvents(lg *zap.Logger, wg *watcherGroup, revs, vals [][]byte) (evs []m
449456
// watchers that watch on the key of the event.
450457
func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) {
451458
victim := make(watcherBatch)
452-
for w, eb := range newWatcherBatch(&s.synced, evs) {
459+
for w, eb := range newWatcherBatch(&s.synced, evs, s.watchBatchMaxSize) {
453460
if w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: rev}) {
454461
pendingEventsGauge.Add(float64(len(eb.evs)))
455462
} else {

‎server/storage/mvcc/watchable_store_bench_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import (
2727

2828
func BenchmarkWatchableStorePut(b *testing.B) {
2929
be, _ := betesting.NewDefaultTmpBackend(b)
30-
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, StoreConfig{})
30+
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, WatchableStoreConfig{})
3131
defer cleanup(s, be)
3232

3333
// arbitrary number of bytes
@@ -47,7 +47,7 @@ func BenchmarkWatchableStorePut(b *testing.B) {
4747
// some synchronization operations, such as mutex locking.
4848
func BenchmarkWatchableStoreTxnPut(b *testing.B) {
4949
be, _ := betesting.NewDefaultTmpBackend(b)
50-
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, StoreConfig{})
50+
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, WatchableStoreConfig{})
5151
defer cleanup(s, be)
5252

5353
// arbitrary number of bytes
@@ -78,7 +78,7 @@ func BenchmarkWatchableStoreWatchPutUnsync(b *testing.B) {
7878

7979
func benchmarkWatchableStoreWatchPut(b *testing.B, synced bool) {
8080
be, _ := betesting.NewDefaultTmpBackend(b)
81-
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, StoreConfig{})
81+
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, WatchableStoreConfig{})
8282
defer cleanup(s, be)
8383

8484
k := []byte("testkey")
@@ -122,7 +122,7 @@ func benchmarkWatchableStoreWatchPut(b *testing.B, synced bool) {
122122
// we should put to simulate the real-world use cases.
123123
func BenchmarkWatchableStoreUnsyncedCancel(b *testing.B) {
124124
be, _ := betesting.NewDefaultTmpBackend(b)
125-
ws := newWatchableStore(zaptest.NewLogger(b), be, &lease.FakeLessor{}, StoreConfig{})
125+
ws := newWatchableStore(zaptest.NewLogger(b), be, &lease.FakeLessor{}, WatchableStoreConfig{})
126126

127127
defer cleanup(ws, be)
128128

@@ -164,7 +164,7 @@ func BenchmarkWatchableStoreUnsyncedCancel(b *testing.B) {
164164

165165
func BenchmarkWatchableStoreSyncedCancel(b *testing.B) {
166166
be, _ := betesting.NewDefaultTmpBackend(b)
167-
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, StoreConfig{})
167+
s := New(zaptest.NewLogger(b), be, &lease.FakeLessor{}, WatchableStoreConfig{})
168168

169169
defer cleanup(s, be)
170170

0 commit comments

Comments
 (0)