etcd/mvcc/watchable_store.go

387 lines
9.2 KiB
Go
Raw Normal View History

2015-09-15 23:54:11 +03:00
// Copyright 2015 CoreOS, Inc.
//
// 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.
2016-04-25 22:32:58 +03:00
package mvcc
import (
"log"
"sync"
"time"
2016-01-05 21:16:50 +03:00
"github.com/coreos/etcd/lease"
2016-04-25 22:32:58 +03:00
"github.com/coreos/etcd/mvcc/backend"
"github.com/coreos/etcd/mvcc/mvccpb"
)
const (
// chanBufLen is the length of the buffered chan
// for sending out watched events.
// TODO: find a good buf value. 1024 is just a random one that
// seems to be reasonable.
chanBufLen = 1024
)
type watchable interface {
2016-02-26 19:55:28 +03:00
watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse) (*watcher, cancelFunc)
2016-03-03 09:34:54 +03:00
progress(w *watcher)
rev() int64
}
type watchableStore struct {
mu sync.Mutex
*store
2016-01-03 07:20:22 +03:00
// contains all unsynced watchers that needs to sync with events that have happened
2016-02-26 19:55:28 +03:00
unsynced watcherGroup
storage: use map for watchable store unsynced This is for `TODO: use map to reduce cancel cost`. I switched slice to map, and benchmark results show that map implementation performs better, as follows: ``` [1]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 215212 1307 -99.39% BenchmarkWatchableStoreUnsyncedCancel-2 120453 710 -99.41% BenchmarkWatchableStoreUnsyncedCancel-4 120765 748 -99.38% BenchmarkWatchableStoreUnsyncedCancel-8 121391 719 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [2]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 212550 1117 -99.47% BenchmarkWatchableStoreUnsyncedCancel-2 120927 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120752 699 -99.42% BenchmarkWatchableStoreUnsyncedCancel-8 121012 688 -99.43% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 197 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [3]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 214268 1183 -99.45% BenchmarkWatchableStoreUnsyncedCancel-2 120763 759 -99.37% BenchmarkWatchableStoreUnsyncedCancel-4 120321 708 -99.41% BenchmarkWatchableStoreUnsyncedCancel-8 121628 680 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [4]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 208332 1089 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121011 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120678 681 -99.44% BenchmarkWatchableStoreUnsyncedCancel-8 121303 721 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 194 1 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [5]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 211900 1097 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121795 753 -99.38% BenchmarkWatchableStoreUnsyncedCancel-4 123182 700 -99.43% BenchmarkWatchableStoreUnsyncedCancel-8 122820 688 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 198 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 140 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 141 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 141 0 -100.00% ```
2015-10-21 22:06:27 +03:00
2016-01-03 07:20:22 +03:00
// contains all synced watchers that are in sync with the progress of the store.
// The key of the map is the key that the watcher watches on.
2016-02-26 19:55:28 +03:00
synced watcherGroup
stopc chan struct{}
wg sync.WaitGroup
}
// cancelFunc updates unsynced and synced maps when running
// cancel operations.
type cancelFunc func()
2016-03-30 21:37:55 +03:00
func New(b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter) ConsistentWatchableKV {
return newWatchableStore(b, le, ig)
}
func newWatchableStore(b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter) *watchableStore {
s := &watchableStore{
2016-03-30 21:37:55 +03:00
store: NewStore(b, le, ig),
2016-02-26 19:55:28 +03:00
unsynced: newWatcherGroup(),
synced: newWatcherGroup(),
storage: use map for watchable store unsynced This is for `TODO: use map to reduce cancel cost`. I switched slice to map, and benchmark results show that map implementation performs better, as follows: ``` [1]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 215212 1307 -99.39% BenchmarkWatchableStoreUnsyncedCancel-2 120453 710 -99.41% BenchmarkWatchableStoreUnsyncedCancel-4 120765 748 -99.38% BenchmarkWatchableStoreUnsyncedCancel-8 121391 719 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [2]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 212550 1117 -99.47% BenchmarkWatchableStoreUnsyncedCancel-2 120927 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120752 699 -99.42% BenchmarkWatchableStoreUnsyncedCancel-8 121012 688 -99.43% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 197 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [3]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 214268 1183 -99.45% BenchmarkWatchableStoreUnsyncedCancel-2 120763 759 -99.37% BenchmarkWatchableStoreUnsyncedCancel-4 120321 708 -99.41% BenchmarkWatchableStoreUnsyncedCancel-8 121628 680 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [4]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 208332 1089 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121011 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120678 681 -99.44% BenchmarkWatchableStoreUnsyncedCancel-8 121303 721 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 194 1 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [5]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 211900 1097 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121795 753 -99.38% BenchmarkWatchableStoreUnsyncedCancel-4 123182 700 -99.43% BenchmarkWatchableStoreUnsyncedCancel-8 122820 688 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 198 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 140 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 141 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 141 0 -100.00% ```
2015-10-21 22:06:27 +03:00
stopc: make(chan struct{}),
}
if s.le != nil {
// use this store as the deleter so revokes trigger watch events
s.le.SetRangeDeleter(s)
}
s.wg.Add(1)
2016-01-03 07:20:22 +03:00
go s.syncWatchersLoop()
return s
}
2016-01-05 21:16:50 +03:00
func (s *watchableStore) Put(key, value []byte, lease lease.LeaseID) (rev int64) {
s.mu.Lock()
defer s.mu.Unlock()
2016-01-05 02:12:59 +03:00
rev = s.store.Put(key, value, lease)
changes := s.store.getChanges()
if len(changes) != 1 {
log.Panicf("unexpected len(changes) != 1 after put")
}
2016-04-25 22:32:58 +03:00
ev := mvccpb.Event{
Type: mvccpb.PUT,
Kv: &changes[0],
}
2016-04-25 22:32:58 +03:00
s.notify(rev, []mvccpb.Event{ev})
return rev
}
func (s *watchableStore) DeleteRange(key, end []byte) (n, rev int64) {
s.mu.Lock()
defer s.mu.Unlock()
n, rev = s.store.DeleteRange(key, end)
changes := s.store.getChanges()
if len(changes) != int(n) {
log.Panicf("unexpected len(changes) != n after deleteRange")
}
if n == 0 {
return n, rev
}
2016-04-25 22:32:58 +03:00
evs := make([]mvccpb.Event, n)
for i, change := range changes {
2016-04-25 22:32:58 +03:00
evs[i] = mvccpb.Event{
Type: mvccpb.DELETE,
Kv: &change}
evs[i].Kv.ModRevision = rev
}
2016-02-03 10:02:15 +03:00
s.notify(rev, evs)
return n, rev
}
func (s *watchableStore) TxnBegin() int64 {
s.mu.Lock()
return s.store.TxnBegin()
}
func (s *watchableStore) TxnEnd(txnID int64) error {
err := s.store.TxnEnd(txnID)
if err != nil {
return err
}
changes := s.getChanges()
if len(changes) == 0 {
s.mu.Unlock()
return nil
}
rev := s.store.Rev()
2016-04-25 22:32:58 +03:00
evs := make([]mvccpb.Event, len(changes))
for i, change := range changes {
switch change.CreateRevision {
case 0:
2016-04-25 22:32:58 +03:00
evs[i] = mvccpb.Event{
Type: mvccpb.DELETE,
Kv: &changes[i]}
evs[i].Kv.ModRevision = rev
default:
2016-04-25 22:32:58 +03:00
evs[i] = mvccpb.Event{
Type: mvccpb.PUT,
Kv: &changes[i]}
}
}
s.notify(rev, evs)
s.mu.Unlock()
return nil
}
func (s *watchableStore) Close() error {
close(s.stopc)
s.wg.Wait()
return s.store.Close()
}
func (s *watchableStore) NewWatchStream() WatchStream {
watchStreamGauge.Inc()
return &watchStream{
watchable: s,
ch: make(chan WatchResponse, chanBufLen),
cancels: make(map[WatchID]cancelFunc),
2016-03-03 09:34:54 +03:00
watchers: make(map[WatchID]*watcher),
}
}
2016-02-26 19:55:28 +03:00
func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse) (*watcher, cancelFunc) {
s.mu.Lock()
defer s.mu.Unlock()
2016-01-03 07:20:22 +03:00
wa := &watcher{
2016-02-26 19:55:28 +03:00
key: key,
end: end,
cur: startRev,
id: id,
ch: ch,
}
s.store.mu.Lock()
synced := startRev > s.store.currentRev.main || startRev == 0
if synced {
wa.cur = s.store.currentRev.main + 1
2016-03-07 04:41:30 +03:00
if startRev > wa.cur {
wa.cur = startRev
}
}
s.store.mu.Unlock()
if synced {
s.synced.add(wa)
} else {
2016-01-03 07:20:22 +03:00
slowWatcherGauge.Inc()
2016-02-03 07:06:53 +03:00
s.unsynced.add(wa)
}
2016-01-03 07:20:22 +03:00
watcherGauge.Inc()
cancel := cancelFunc(func() {
s.mu.Lock()
defer s.mu.Unlock()
2016-02-03 07:06:53 +03:00
// remove references of the watcher
if s.unsynced.delete(wa) {
2016-01-03 07:20:22 +03:00
slowWatcherGauge.Dec()
watcherGauge.Dec()
storage: use map for watchable store unsynced This is for `TODO: use map to reduce cancel cost`. I switched slice to map, and benchmark results show that map implementation performs better, as follows: ``` [1]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 215212 1307 -99.39% BenchmarkWatchableStoreUnsyncedCancel-2 120453 710 -99.41% BenchmarkWatchableStoreUnsyncedCancel-4 120765 748 -99.38% BenchmarkWatchableStoreUnsyncedCancel-8 121391 719 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [2]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 212550 1117 -99.47% BenchmarkWatchableStoreUnsyncedCancel-2 120927 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120752 699 -99.42% BenchmarkWatchableStoreUnsyncedCancel-8 121012 688 -99.43% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 197 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [3]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 214268 1183 -99.45% BenchmarkWatchableStoreUnsyncedCancel-2 120763 759 -99.37% BenchmarkWatchableStoreUnsyncedCancel-4 120321 708 -99.41% BenchmarkWatchableStoreUnsyncedCancel-8 121628 680 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 200 1 -99.50% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 138 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [4]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 208332 1089 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121011 691 -99.43% BenchmarkWatchableStoreUnsyncedCancel-4 120678 681 -99.44% BenchmarkWatchableStoreUnsyncedCancel-8 121303 721 -99.41% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 194 1 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 139 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 139 0 -100.00% [5]: benchmark old ns/op new ns/op delta BenchmarkWatchableStoreUnsyncedCancel 211900 1097 -99.48% BenchmarkWatchableStoreUnsyncedCancel-2 121795 753 -99.38% BenchmarkWatchableStoreUnsyncedCancel-4 123182 700 -99.43% BenchmarkWatchableStoreUnsyncedCancel-8 122820 688 -99.44% benchmark old allocs new allocs delta BenchmarkWatchableStoreUnsyncedCancel 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-2 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-4 0 0 +0.00% BenchmarkWatchableStoreUnsyncedCancel-8 0 0 +0.00% benchmark old bytes new bytes delta BenchmarkWatchableStoreUnsyncedCancel 198 1 -99.49% BenchmarkWatchableStoreUnsyncedCancel-2 140 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-4 141 0 -100.00% BenchmarkWatchableStoreUnsyncedCancel-8 141 0 -100.00% ```
2015-10-21 22:06:27 +03:00
return
}
2016-02-03 06:28:42 +03:00
if s.synced.delete(wa) {
watcherGauge.Dec()
}
// If we cannot find it, it should have finished watch.
})
return wa, cancel
}
2016-02-01 08:42:39 +03:00
// syncWatchersLoop syncs the watcher in the unsynced map every 100ms.
2016-01-03 07:20:22 +03:00
func (s *watchableStore) syncWatchersLoop() {
defer s.wg.Done()
for {
s.mu.Lock()
2016-01-03 07:20:22 +03:00
s.syncWatchers()
s.mu.Unlock()
select {
case <-time.After(100 * time.Millisecond):
case <-s.stopc:
return
}
}
}
2016-01-03 07:20:22 +03:00
// syncWatchers periodically syncs unsynced watchers by: Iterate all unsynced
// watchers to get the minimum revision within its range, skipping the
// watcher if its current revision is behind the compact revision of the
// store. And use this minimum revision to get all key-value pairs. Then send
2016-01-03 07:20:22 +03:00
// those events to watchers.
func (s *watchableStore) syncWatchers() {
s.store.mu.Lock()
defer s.store.mu.Unlock()
2016-02-26 19:55:28 +03:00
if s.unsynced.size() == 0 {
return
}
2016-01-03 07:20:22 +03:00
// in order to find key-value pairs from unsynced watchers, we need to
// find min revision index, and these revisions can be used to
// query the backend store of key-value pairs
curRev := s.store.currentRev.main
2016-02-26 19:55:28 +03:00
compactionRev := s.store.compactMainRev
minRev := s.unsynced.scanMinRev(curRev, compactionRev)
minBytes, maxBytes := newRevBytes(), newRevBytes()
revToBytes(revision{main: minRev}, minBytes)
revToBytes(revision{main: curRev + 1}, maxBytes)
// UnsafeRange returns keys and values. And in boltdb, keys are revisions.
// values are actual key-value pairs in backend.
tx := s.store.b.BatchTx()
tx.Lock()
revs, vs := tx.UnsafeRange(keyBucketName, minBytes, maxBytes, 0)
2016-02-26 19:55:28 +03:00
evs := kvsToEvents(&s.unsynced, revs, vs)
tx.Unlock()
2016-03-03 09:34:54 +03:00
wb := newWatcherBatch(&s.unsynced, evs)
for w, eb := range wb {
select {
// s.store.Rev also uses Lock, so just return directly
case w.ch <- WatchResponse{WatchID: w.id, Events: eb.evs, Revision: s.store.currentRev.main}:
pendingEventsGauge.Add(float64(len(eb.evs)))
default:
// TODO: handle the full unsynced watchers.
// continue to process other watchers for now, the full ones
// will be processed next time and hopefully it will not be full.
continue
}
if eb.moreRev != 0 {
w.cur = eb.moreRev
continue
}
w.cur = curRev
s.synced.add(w)
s.unsynced.delete(w)
}
2016-03-03 09:34:54 +03:00
// bring all un-notified watchers to synced.
for w := range s.unsynced.watchers {
if !wb.contains(w) {
w.cur = curRev
s.synced.add(w)
s.unsynced.delete(w)
}
}
2016-02-26 19:55:28 +03:00
slowWatcherGauge.Set(float64(s.unsynced.size()))
}
// kvsToEvents gets all events for the watchers from all key-value pairs
2016-04-25 22:32:58 +03:00
func kvsToEvents(wg *watcherGroup, revs, vals [][]byte) (evs []mvccpb.Event) {
for i, v := range vals {
2016-04-25 22:32:58 +03:00
var kv mvccpb.KeyValue
if err := kv.Unmarshal(v); err != nil {
2016-04-25 22:32:58 +03:00
log.Panicf("mvcc: cannot unmarshal event: %v", err)
}
2016-02-26 19:55:28 +03:00
if !wg.contains(string(kv.Key)) {
continue
}
2016-04-25 22:32:58 +03:00
ty := mvccpb.PUT
if isTombstone(revs[i]) {
2016-04-25 22:32:58 +03:00
ty = mvccpb.DELETE
// patch in mod revision so watchers won't skip
kv.ModRevision = bytesToRev(revs[i]).main
}
2016-04-25 22:32:58 +03:00
evs = append(evs, mvccpb.Event{Kv: &kv, Type: ty})
}
return evs
}
// notify notifies the fact that given event at the given rev just happened to
2016-01-03 07:20:22 +03:00
// watchers that watch on the key of the event.
2016-04-25 22:32:58 +03:00
func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) {
2016-02-26 19:55:28 +03:00
for w, eb := range newWatcherBatch(&s.synced, evs) {
if eb.revs != 1 {
2016-04-25 22:32:58 +03:00
log.Panicf("mvcc: unexpected multiple revisions in notification")
2016-02-26 19:55:28 +03:00
}
select {
case w.ch <- WatchResponse{WatchID: w.id, Events: eb.evs, Revision: s.Rev()}:
pendingEventsGauge.Add(float64(len(eb.evs)))
default:
// move slow watcher to unsynced
w.cur = rev
s.unsynced.add(w)
s.synced.delete(w)
slowWatcherGauge.Inc()
}
}
}
func (s *watchableStore) rev() int64 { return s.store.Rev() }
2016-03-03 09:34:54 +03:00
func (s *watchableStore) progress(w *watcher) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.synced.watchers[w]; ok {
select {
case w.ch <- WatchResponse{WatchID: w.id, Revision: s.rev()}:
default:
// If the ch is full, this watcher is receiving events.
// We do not need to send progress at all.
}
}
}
2016-01-03 07:20:22 +03:00
type watcher struct {
// the watcher key
key []byte
2016-02-26 19:55:28 +03:00
// end indicates the end of the range to watch.
// If end is set, the watcher is on a range.
end []byte
2016-03-07 04:41:30 +03:00
// cur is the current watcher revision of a unsynced watcher.
// cur will be updated for unsynced watcher while it is catching up.
// cur is startRev of a synced watcher.
// cur will not be updated for synced watcher.
cur int64
id WatchID
// a chan to send out the watch response.
2016-01-03 07:20:22 +03:00
// The chan might be shared with other watchers.
ch chan<- WatchResponse
}