etcd 分布式数据库概念初探

type treeIndex struct {
sync.RWMutex
tree *btree.BTree
} 

// For example: put(1.0);put(2.0);tombstone(3.0);put(4.0);tombstone(5.0) on key "foo"
// generate a keyIndex:
// key:     "foo"
// rev: 5
// generations:
//    {empty}
//    {4.0, 5.0(t)}
//    {1.0, 2.0, 3.0(t)}
//
// Compact a keyIndex removes the versions with smaller or equal to
// rev except the largest one. If the generation becomes empty
// during compaction, it will be removed. if all the generations get
// removed, the keyIndex should be removed.

// For example:
// compact(2) on the previous example
// generations:
//    {empty}
//    {4.0, 5.0(t)}
//    {2.0, 3.0(t)}
//
// compact(4)
// generations:
//    {empty}
//    {4.0, 5.0(t)}
//
// compact(5):
// generations:
//    {empty} -> key SHOULD be removed.
//
// compact(6):
// generations:
//    {empty} -> key SHOULD be removed. 

kv := mvccpb.KeyValue{
    Key:            key,
    Value:          value,
    CreateRevision: c,
    ModRevision:    rev,
    Version:        ver,
    Lease:          int64(leaseID),
} 

// syncWatchersLoop syncs the watcher in the unsynced map every 100ms.
func (s *watchableStore) syncWatchersLoop() {
defer s.wg.Done()

for {
    s.mu.RLock()
    st := time.Now()
    lastUnsyncedWatchers := s.unsynced.size()
    s.mu.RUnlock()

    unsyncedWatchers := 0
    if lastUnsyncedWatchers > 0 {
        unsyncedWatchers = s.syncWatchers()
    }
    syncDuration := time.Since(st)

    waitDuration := 100 * time.Millisecond
    // more work pending?
    if unsyncedWatchers != 0 && lastUnsyncedWatchers > unsyncedWatchers {
        // be fair to other store operations by yielding time taken
        waitDuration = syncDuration
    }

    select {
    case <-time.After(waitDuration):
    case <-s.stopc:
        return
    }
}
} 

// syncWatchers syncs unsynced watchers by:
//  1. choose a set of watchers from the unsynced watcher group
//  2. iterate over the set to get the minimum revision and remove compacted watchers
//  3. use minimum revision to get all key-value pairs and send those events to watchers
//  4. remove synced watchers in set from unsynced group and move to synced group
func (s *watchableStore) syncWatchers() int {
s.mu.Lock()
defer s.mu.Unlock()

if s.unsynced.size() == 0 {
    return 0
}

s.store.revMu.RLock()
defer s.store.revMu.RUnlock()

// 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
compactionRev := s.store.compactMainRev

wg, minRev := s.unsynced.choose(maxWatchersPerSync, 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.ReadTx()
tx.Lock()
revs, vs := tx.UnsafeRange(keyBucketName, minBytes, maxBytes, 0)
evs := kvsToEvents(wg, revs, vs)
tx.Unlock() 

// kvsToEvents gets all events for the watchers from all key-value pairs
func kvsToEvents(wg *watcherGroup, revs, vals [][]byte) (evs []mvccpb.Event) {
for i, v := range vals {
    var kv mvccpb.KeyValue
    if err := kv.Unmarshal(v); err != nil {
        plog.Panicf("cannot unmarshal event: %v", err)
    }

    if !wg.contains(string(kv.Key)) {
        continue
    }

    ty := mvccpb.PUT
    if isTombstone(revs[i]) {
        ty = mvccpb.DELETE
        // patch in mod revision so watchers won't skip
        kv.ModRevision = bytesToRev(revs[i]).main
    }
    evs = append(evs, mvccpb.Event{Kv: &kv, Type: ty})
}
return evs
} 

ibytes := newRevBytes()
idxRev := revision{main: rev, sub: int64(len(tw.changes))}
revToBytes(idxRev, ibytes)

idxRev := revision{main: tw.beginRev + 1, sub: int64(len(tw.changes))}
revToBytes(idxRev, ibytes)
ibytes = appendMarkTombstone(ibytes)


// appendMarkTombstone appends tombstone mark to normal revision bytes.
func appendMarkTombstone(b []byte) []byte {
if len(b) != revBytesLen {
    plog.Panicf("cannot append mark to non normal revision bytes")
}
return append(b, markTombstone)
}

// isTombstone checks whether the revision bytes is a tombstone.
func isTombstone(b []byte) bool {
return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
}