couchbase / moss / 20

pending completion

Build # 20

Build Type

push

travis-ci

Committed by

steveyen

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/8f1c1581123b0b7fa714523d325dd21cfb8a4af2">8f1c15811<a href="https://github.com/couchbase/moss/commit/8f1c1581123b0b7fa714523d325dd21cfb8a4af2">&quot;&gt;MB-18433 - added basic counter stats

Change-Id: I</a><a class="double-link" href="https://github.com/couchbase/moss/commit/<a class="double-link" href="https://github.com/couchbase/moss/commit/0844ce4a4e810b9e8b0f0bcf56254d7824bf4f19">0844ce4a4</a>">0844ce4a4</a><a href="https://github.com/couchbase/moss/commit/8f1c1581123b0b7fa714523d325dd21cfb8a4af2">
Reviewed-on: http://review.couchbase.org/60867
Reviewed-by: Marty Schoch <marty.schoch@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

105 of 105 new or added lines in 3 files covered. (100.0%)

1190 of 1269 relevant lines covered (93.77%)

33892.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.68

/collection.go

//  Copyright (c) 2016 Couchbase, 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.

package moss

import (
        "fmt"
        "reflect"
        "sort"
        "sync"
        "sync/atomic"
)

// A collection implements the Collection interface.
type collection struct {
        options CollectionOptions

        stopCh          chan struct{}
        pingMergerCh    chan ping
        doneMergerCh    chan struct{}
        donePersisterCh chan struct{}

        m sync.Mutex // Protects the fields that follow.

        // When ExecuteBatch() has pushed a new segment onto
        // stackDirtyTop, it notifies the merger via awakeMergerCh (if
        // non-nil).
        awakeMergerCh chan struct{}

        // stackDirtyTopCond is used to wait for space in stackDirtyTop.
        stackDirtyTopCond *sync.Cond

        // stackDirtyBaseCond is used to wait for non-nil stackDirtyBase.
        stackDirtyBaseCond *sync.Cond

        // ExecuteBatch() will push new segments onto stackDirtyTop if
        // there is space.
        stackDirtyTop *segmentStack

        // The merger goroutine asynchronously, atomically grabs all
        // segments from stackDirtyTop and atomically moves them into
        // stackDirtyMid.  The merger will also merge segments in
        // stackDirtyMid to keep its height low.
        stackDirtyMid *segmentStack

        // stackDirtyBase represents the segments currently being
        // persisted.  It is optionally populated by the merger when there
        // are merged segments ready for persistence.  Will be nil when
        // persistence is not being used.
        stackDirtyBase *segmentStack

        // stackClean represents the segments that have been optionally
        // persisted by the persister, and can now be safely evicted, as
        // the lowerLevelSnapshot will contain the entries from
        // stackClean.  Will be nil when persistence is not being used.
        stackClean *segmentStack

        // lowerLevelSnapshot provides an optional, lower-level storage
        // implementation, when using the Collection as a cache.
        lowerLevelSnapshot *snapshotWrapper

        // stats leverage sync/atomic counters.
        stats *CollectionStats
}

// ------------------------------------------------------

// Start kicks off required background gouroutines.
func (m *collection) Start() error {
        go m.runMerger()
        go m.runPersister()
        return nil
}

// Close synchronously stops background goroutines.
func (m *collection) Close() error {
        atomic.AddUint64(&m.stats.TotCloseBeg, 1)

        close(m.stopCh)

        m.stackDirtyBaseCond.Signal() // Awake persister.

        <-m.doneMergerCh
        atomic.AddUint64(&m.stats.TotCloseMergerDone, 1)

        <-m.donePersisterCh
        atomic.AddUint64(&m.stats.TotClosePersisterDone, 1)

        m.m.Lock()
        if m.lowerLevelSnapshot != nil {
                atomic.AddUint64(&m.stats.TotCloseLowerLevelBeg, 1)
                m.lowerLevelSnapshot.Close()
                m.lowerLevelSnapshot = nil
                atomic.AddUint64(&m.stats.TotCloseLowerLevelEnd, 1)
        }
        m.m.Unlock()

        atomic.AddUint64(&m.stats.TotCloseEnd, 1)

        return nil
}

// Options returns the current options.
func (m *collection) Options() CollectionOptions {
        return m.options
}

// Snapshot returns a stable snapshot of the key-value entries.
func (m *collection) Snapshot() (Snapshot, error) {
        atomic.AddUint64(&m.stats.TotSnapshotBeg, 1)

        rv, _, _, _, _ := m.snapshot(0, nil)

        atomic.AddUint64(&m.stats.TotSnapshotEnd, 1)

        return rv, nil
}

// NewBatch returns a new Batch instance with hinted amount of
// resources expected to be required.
func (m *collection) NewBatch(totalOps, totalKeyValBytes int) (
        Batch, error) {
        atomic.AddUint64(&m.stats.TotNewBatch, 1)
        atomic.AddUint64(&m.stats.TotNewBatchTotalOps, uint64(totalOps))
        atomic.AddUint64(&m.stats.TotNewBatchTotalKeyValBytes, uint64(totalKeyValBytes))

        return newSegment(totalOps, totalKeyValBytes)
}

// ExecuteBatch atomically incorporates the provided Batch into the
// collection.  The Batch instance should not be reused after
// ExecuteBatch() returns.
func (m *collection) ExecuteBatch(bIn Batch,
        writeOptions WriteOptions) error {
        atomic.AddUint64(&m.stats.TotExecuteBatchBeg, 1)

        b, ok := bIn.(*segment)
        if !ok {
                atomic.AddUint64(&m.stats.TotExecuteBatchErr, 1)

                return fmt.Errorf("wrong Batch implementation type")
        }

        if b == nil || b.Len() <= 0 {
                atomic.AddUint64(&m.stats.TotExecuteBatchEmpty, 1)
                return nil
        }

        maxPreMergerBatches := m.options.MaxPreMergerBatches
        if maxPreMergerBatches <= 0 {
                maxPreMergerBatches =
                        DefaultCollectionOptions.MaxPreMergerBatches
        }

        if m.options.DeferredSort {
                b.needSorterCh = make(chan bool, 1)
                b.needSorterCh <- true // A ticket for the future sorter.
                close(b.needSorterCh)

                b.waitSortedCh = make(chan struct{})
        } else {
                sort.Sort(b)
        }

        stackDirtyTop := &segmentStack{collection: m, refs: 1}

        m.m.Lock()

        for m.stackDirtyTop != nil &&
                len(m.stackDirtyTop.a) >= maxPreMergerBatches {
                if m.options.DeferredSort {
                        go b.requestSort(false) // While waiting, might as well sort.
                }

                atomic.AddUint64(&m.stats.TotExecuteBatchWaitBeg, 1)
                m.stackDirtyTopCond.Wait()
                atomic.AddUint64(&m.stats.TotExecuteBatchWaitEnd, 1)
        }

        numDirtyTop := 0
        if m.stackDirtyTop != nil {
                numDirtyTop = len(m.stackDirtyTop.a)
        }

        stackDirtyTop.a = make([]*segment, 0, numDirtyTop+1)

        if m.stackDirtyTop != nil {
                stackDirtyTop.a = append(stackDirtyTop.a, m.stackDirtyTop.a...)
        }

        stackDirtyTop.a = append(stackDirtyTop.a, b)

        prevStackDirtyTop := m.stackDirtyTop
        m.stackDirtyTop = stackDirtyTop

        awakeMergerCh := m.awakeMergerCh
        m.awakeMergerCh = nil

        m.m.Unlock()

        prevStackDirtyTop.Close()

        if awakeMergerCh != nil {
                atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerBeg, 1)
                close(awakeMergerCh)
                atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerEnd, 1)
        }

        atomic.AddUint64(&m.stats.TotExecuteBatchEnd, 1)

        return nil
}

// ------------------------------------------------------

// WaitForMerger blocks until the merger has run another cycle.
// Providing a kind of "mergeAll" forces a full merge and can be
// useful for applications that are no longer performing mutations and
// that want to optimize for retrievals.
func (m *collection) WaitForMerger(kind string) error {
        atomic.AddUint64(&m.stats.TotWaitForMergerBeg, 1)

        pongCh := make(chan struct{})
        m.pingMergerCh <- ping{
                kind:   kind,
                pongCh: pongCh,
        }
        <-pongCh

        atomic.AddUint64(&m.stats.TotWaitForMergerEnd, 1)
        return nil
}

// ------------------------------------------------------

// Log invokes the user's configured Log callback, if any, if the
// debug levels are met.
func (m *collection) Logf(format string, a ...interface{}) {
        if m.options.Debug > 0 &&
                m.options.Log != nil {
                m.options.Log(format, a...)
        }
}

// OnError invokes the user's configured OnError callback, in which
// the application might take further action, for example, such as
// Close()'ing the Collection in order to fix underlying
// storage/resource issues.
func (m *collection) OnError(err error) {
        atomic.AddUint64(&m.stats.TotOnError, 1)

        if m.options.OnError != nil {
                m.options.OnError(err)
        }
}

// ------------------------------------------------------

const snapshotSkipDirtyTop = uint32(0x00000001)
const snapshotSkipDirtyMid = uint32(0x00000002)
const snapshotSkipDirtyBase = uint32(0x00000004)
const snapshotSkipClean = uint32(0x00000008)

// snapshot() atomically clones the various stacks into a new, single
// segmentStack, controllable by skip flags, and also invokes the
// optional callback while holding the collection lock.
func (m *collection) snapshot(skip uint32, cb func(*segmentStack)) (
        *segmentStack, int, int, int, int) {
        atomic.AddUint64(&m.stats.TotSnapshotInternalBeg, 1)

        rv := &segmentStack{collection: m, refs: 1}

        heightDirtyTop := 0
        heightDirtyMid := 0
        heightDirtyBase := 0
        heightClean := 0

        m.m.Lock()

        rv.lowerLevelSnapshot = m.lowerLevelSnapshot.addRef()

        if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {
                heightDirtyTop = len(m.stackDirtyTop.a)
        }

        if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {
                heightDirtyMid = len(m.stackDirtyMid.a)
        }

        if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {
                heightDirtyBase = len(m.stackDirtyBase.a)
        }

        if m.stackClean != nil && (skip&snapshotSkipClean == 0) {
                heightClean = len(m.stackClean.a)
        }

        rv.a = make([]*segment, 0,
                heightDirtyTop+heightDirtyMid+heightDirtyBase+heightClean)

        if m.stackClean != nil && (skip&snapshotSkipClean == 0) {
                rv.a = append(rv.a, m.stackClean.a...)
        }

        if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {
                rv.a = append(rv.a, m.stackDirtyBase.a...)
        }

        if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {
                rv.a = append(rv.a, m.stackDirtyMid.a...)
        }

        if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {
                rv.a = append(rv.a, m.stackDirtyTop.a...)
        }

        if cb != nil {
                cb(rv)
        }

        m.m.Unlock()

        atomic.AddUint64(&m.stats.TotSnapshotInternalEnd, 1)

        return rv, heightClean, heightDirtyBase, heightDirtyMid, heightDirtyTop
}

// ------------------------------------------------------

// runMerger() implements the background merger task.
func (m *collection) runMerger() {
        defer func() {
                close(m.doneMergerCh)

                atomic.AddUint64(&m.stats.TotMergerEnd, 1)
        }()

        pings := []ping{}

        defer func() {
                replyToPings(pings)
                pings = pings[0:0]
        }()

OUTER:
        for {
                atomic.AddUint64(&m.stats.TotMergerLoop, 1)

                // ---------------------------------------------
                // Notify ping'ers from the previous loop.

                replyToPings(pings)
                pings = pings[0:0]

                // ---------------------------------------------
                // Wait for new stackDirtyTop entries and/or pings.

                var awakeMergerCh chan struct{}

                m.m.Lock()

                if m.stackDirtyTop == nil || len(m.stackDirtyTop.a) <= 0 {
                        m.awakeMergerCh = make(chan struct{})
                        awakeMergerCh = m.awakeMergerCh
                }

                m.m.Unlock()

                mergeAll := false

                if awakeMergerCh != nil {
                        atomic.AddUint64(&m.stats.TotMergerWaitBeg, 1)

                        select {
                        case <-m.stopCh:
                                return

                        case ping := <-m.pingMergerCh:
                                pings = append(pings, ping)
                                if ping.kind == "mergeAll" {
                                        mergeAll = true
                                }

                        case <-awakeMergerCh:
                                // NO-OP.
                        }

                        atomic.AddUint64(&m.stats.TotMergerWaitEnd, 1)
                }

                pings, mergeAll =
                        receivePings(m.pingMergerCh, pings, "mergeAll", mergeAll)

                // ---------------------------------------------
                // Atomically ingest stackDirtyTop into stackDirtyMid.

                var stackDirtyTopPrev *segmentStack
                var stackDirtyMidPrev *segmentStack

                stackDirtyMid, _, _, prevLenDirtyMid, prevLenDirtyTop :=
                        m.snapshot(snapshotSkipClean|snapshotSkipDirtyBase,
                                func(ss *segmentStack) {
                                        // m.stackDirtyMid takes 1 refs, and
                                        // stackDirtyMid takes 1 refs.
                                        ss.refs++

                                        stackDirtyTopPrev = m.stackDirtyTop
                                        m.stackDirtyTop = nil

                                        stackDirtyMidPrev = m.stackDirtyMid
                                        m.stackDirtyMid = ss

                                        // Awake any writers that are waiting for more space
                                        // in stackDirtyTop.
                                        m.stackDirtyTopCond.Signal()
                                })

                stackDirtyTopPrev.Close()
                stackDirtyMidPrev.Close()

                // ---------------------------------------------
                // Merge multiple stackDirtyMid layers.

                if len(stackDirtyMid.a) > 1 {
                        newTopLevel := 0

                        if !mergeAll {
                                // If we have not been asked to merge all segments,
                                // then heuristically calc a newTopLevel.
                                newTopLevel = stackDirtyMid.calcTargetTopLevel()
                        }

                        if newTopLevel <= 0 {
                                atomic.AddUint64(&m.stats.TotMergerAll, 1)
                        }

                        atomic.AddUint64(&m.stats.TotMergerInternalBeg, 1)

                        mergedStackDirtyMid, err := stackDirtyMid.merge(newTopLevel)
                        if err != nil {
                                atomic.AddUint64(&m.stats.TotMergerInternalErr, 1)

                                m.Logf("collection: runMerger stackDirtyMid.merge,"+
                                        " newTopLevel: %d, err: %v", newTopLevel, err)

                                m.OnError(err)

                                continue OUTER
                        }

                        atomic.AddUint64(&m.stats.TotMergerInternalEnd, 1)

                        stackDirtyMid.Close()

                        mergedStackDirtyMid.addRef()
                        stackDirtyMid = mergedStackDirtyMid

                        m.m.Lock()
                        stackDirtyMidPrev = m.stackDirtyMid
                        m.stackDirtyMid = mergedStackDirtyMid
                        m.m.Unlock()

                        stackDirtyMidPrev.Close()
                } else {
                        atomic.AddUint64(&m.stats.TotMergerInternalSkip, 1)
                }

                lenDirtyMid := len(stackDirtyMid.a)
                if lenDirtyMid > 0 {
                        topDirtyMid := stackDirtyMid.a[lenDirtyMid-1]

                        m.Logf("collection: runMerger,"+
                                " dirtyTop prev height: %2d,"+
                                " dirtyMid height: %2d (%2d),"+
                                " dirtyMid top (%0.2f kvs cap, %0.2f buf cap) # entries: %d",
                                prevLenDirtyTop, lenDirtyMid, lenDirtyMid-prevLenDirtyMid,
                                float64(len(topDirtyMid.kvs))/float64(cap(topDirtyMid.kvs)),
                                float64(len(topDirtyMid.buf))/float64(cap(topDirtyMid.buf)),
                                topDirtyMid.Len())
                }

                stackDirtyMid.Close()

                // ---------------------------------------------
                // Notify persister.

                if m.options.LowerLevelUpdate != nil {
                        m.m.Lock()
                        if m.stackDirtyBase == nil &&
                                m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0 {
                                atomic.AddUint64(&m.stats.TotMergerLowerLevelNotify, 1)

                                m.stackDirtyBase = m.stackDirtyMid
                                m.stackDirtyMid = nil

                                m.stackDirtyBaseCond.Signal()
                        } else {
                                atomic.AddUint64(&m.stats.TotMergerLowerLevelNotifySkip, 1)
                        }
                        m.m.Unlock()
                }

                atomic.AddUint64(&m.stats.TotMergerLoopRepeat, 1)
        }

        // TODO: Concurrent merging of disjoint slices of stackDirtyMid
        // instead of the current, single-threaded merger?
        //
        // TODO: A busy merger means no feeding of the persister?
        //
        // TODO: Delay merger until lots of deletion tombstones?
        //
        // TODO: The base layer is likely the largest, so instead of heap
        // merging the base layer entries, treat the base layer with
        // special case to binary search to find better start points?
        //
        // TODO: Dynamically calc'ed soft max dirty top height, for
        // read-heavy (favor lower) versus write-heavy (favor higher)
        // situations?
}

// ------------------------------------------------------

// Stats returns stats for this collection.
func (m *collection) Stats() (*CollectionStats, error) {
        rv := &CollectionStats{}
        m.stats.AtomicCopyTo(rv)
        return rv, nil
}

// AtomicCopyTo copies stats from s to r (from source to result).
func (s *CollectionStats) AtomicCopyTo(r *CollectionStats) {
        rve := reflect.ValueOf(r).Elem()
        sve := reflect.ValueOf(s).Elem()
        svet := sve.Type()
        for i := 0; i < svet.NumField(); i++ {
                rvef := rve.Field(i)
                svef := sve.Field(i)
                if rvef.CanAddr() && svef.CanAddr() {
                        rvefp := rvef.Addr().Interface()
                        svefp := svef.Addr().Interface()
                        atomic.StoreUint64(rvefp.(*uint64),
                                atomic.LoadUint64(svefp.(*uint64)))
                }
        }
}

1	// Copyright (c) 2016 Couchbase, Inc.
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the
4	// License. You may obtain a copy of the License at
5	// http://www.apache.org/licenses/LICENSE-2.0
6	// Unless required by applicable law or agreed to in writing,
7	// software distributed under the License is distributed on an "AS
8	// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
9	// express or implied. See the License for the specific language
10	// governing permissions and limitations under the License.
11
12	package moss
13
14	import (
15	"fmt"
16	"reflect"
17	"sort"
18	"sync"
19	"sync/atomic"
20	)
21
22	// A collection implements the Collection interface.
23	type collection struct {
24	options CollectionOptions
25
26	stopCh chan struct{}
27	pingMergerCh chan ping
28	doneMergerCh chan struct{}
29	donePersisterCh chan struct{}
30
31	m sync.Mutex // Protects the fields that follow.
32
33	// When ExecuteBatch() has pushed a new segment onto
34	// stackDirtyTop, it notifies the merger via awakeMergerCh (if
35	// non-nil).
36	awakeMergerCh chan struct{}
37
38	// stackDirtyTopCond is used to wait for space in stackDirtyTop.
39	stackDirtyTopCond *sync.Cond
40
41	// stackDirtyBaseCond is used to wait for non-nil stackDirtyBase.
42	stackDirtyBaseCond *sync.Cond
43
44	// ExecuteBatch() will push new segments onto stackDirtyTop if
45	// there is space.
46	stackDirtyTop *segmentStack
47
48	// The merger goroutine asynchronously, atomically grabs all
49	// segments from stackDirtyTop and atomically moves them into
50	// stackDirtyMid. The merger will also merge segments in
51	// stackDirtyMid to keep its height low.
52	stackDirtyMid *segmentStack
53
54	// stackDirtyBase represents the segments currently being
55	// persisted. It is optionally populated by the merger when there
56	// are merged segments ready for persistence. Will be nil when
57	// persistence is not being used.
58	stackDirtyBase *segmentStack
59
60	// stackClean represents the segments that have been optionally
61	// persisted by the persister, and can now be safely evicted, as
62	// the lowerLevelSnapshot will contain the entries from
63	// stackClean. Will be nil when persistence is not being used.
64	stackClean *segmentStack
65
66	// lowerLevelSnapshot provides an optional, lower-level storage
67	// implementation, when using the Collection as a cache.
68	lowerLevelSnapshot *snapshotWrapper
69
70	// stats leverage sync/atomic counters.
71	stats *CollectionStats
72	}
73
74	// ------------------------------------------------------
75
76	// Start kicks off required background gouroutines.
77	func (m *collection) Start() error {	1✔
78	go m.runMerger()	10✔
79	go m.runPersister()	10✔
80	return nil	10✔
81	}	1✔
82
83	// Close synchronously stops background goroutines.
84	func (m *collection) Close() error {	1✔
85	atomic.AddUint64(&m.stats.TotCloseBeg, 1)	9✔
86		1✔
87	close(m.stopCh)	9✔
88		1✔
89	m.stackDirtyBaseCond.Signal() // Awake persister.	9✔
90		1✔
91	<-m.doneMergerCh	9✔
92	atomic.AddUint64(&m.stats.TotCloseMergerDone, 1)	9✔
93		1✔
94	<-m.donePersisterCh	9✔
95	atomic.AddUint64(&m.stats.TotClosePersisterDone, 1)	9✔
96		1✔
97	m.m.Lock()	9✔
98	if m.lowerLevelSnapshot != nil {	9✔
99	atomic.AddUint64(&m.stats.TotCloseLowerLevelBeg, 1)	3✔
100	m.lowerLevelSnapshot.Close()	3✔
101	m.lowerLevelSnapshot = nil	3✔
102	atomic.AddUint64(&m.stats.TotCloseLowerLevelEnd, 1)	3✔
103	}	1✔
104	m.m.Unlock()	9✔
105		1✔
106	atomic.AddUint64(&m.stats.TotCloseEnd, 1)	9✔
107		1✔
108	return nil	9✔
109	}	1✔
110
111	// Options returns the current options.
112	func (m *collection) Options() CollectionOptions {	1✔
113	return m.options	×
114	}	1✔
115
116	// Snapshot returns a stable snapshot of the key-value entries.
117	func (m *collection) Snapshot() (Snapshot, error) {	1✔
118	atomic.AddUint64(&m.stats.TotSnapshotBeg, 1)	127✔
119		1✔
120	rv, _, _, _, _ := m.snapshot(0, nil)	127✔
121		1✔
122	atomic.AddUint64(&m.stats.TotSnapshotEnd, 1)	127✔
123		1✔
124	return rv, nil	127✔
125	}	1✔
126
127	// NewBatch returns a new Batch instance with hinted amount of
128	// resources expected to be required.
129	func (m *collection) NewBatch(totalOps, totalKeyValBytes int) (	1✔
130	Batch, error) {	1✔
131	atomic.AddUint64(&m.stats.TotNewBatch, 1)	32✔
132	atomic.AddUint64(&m.stats.TotNewBatchTotalOps, uint64(totalOps))	32✔
133	atomic.AddUint64(&m.stats.TotNewBatchTotalKeyValBytes, uint64(totalKeyValBytes))	32✔
134		1✔
135	return newSegment(totalOps, totalKeyValBytes)	32✔
136	}	1✔
137
138	// ExecuteBatch atomically incorporates the provided Batch into the
139	// collection. The Batch instance should not be reused after
140	// ExecuteBatch() returns.
141	func (m *collection) ExecuteBatch(bIn Batch,	1✔
142	writeOptions WriteOptions) error {	1✔
143	atomic.AddUint64(&m.stats.TotExecuteBatchBeg, 1)	31✔
144		1✔
145	b, ok := bIn.(*segment)	31✔
146	if !ok {	31✔
147	atomic.AddUint64(&m.stats.TotExecuteBatchErr, 1)	×
148		1✔
149	return fmt.Errorf("wrong Batch implementation type")	×
150	}	1✔
151		1✔
152	if b == nil \|\| b.Len() <= 0 {	31✔
153	atomic.AddUint64(&m.stats.TotExecuteBatchEmpty, 1)	1✔
154	return nil	1✔
155	}	1✔
156		1✔
157	maxPreMergerBatches := m.options.MaxPreMergerBatches	30✔
158	if maxPreMergerBatches <= 0 {	30✔
159	maxPreMergerBatches =	30✔
160	DefaultCollectionOptions.MaxPreMergerBatches	1✔
161	}	1✔
162		1✔
163	if m.options.DeferredSort {	30✔
164	b.needSorterCh = make(chan bool, 1)	×
165	b.needSorterCh <- true // A ticket for the future sorter.	×
166	close(b.needSorterCh)	×
167		1✔
168	b.waitSortedCh = make(chan struct{})	×
169	} else {	1✔
170	sort.Sort(b)	30✔
171	}	1✔
172		1✔
173	stackDirtyTop := &segmentStack{collection: m, refs: 1}	30✔
174		1✔
175	m.m.Lock()	30✔
176		1✔
177	for m.stackDirtyTop != nil &&	30✔
178	len(m.stackDirtyTop.a) >= maxPreMergerBatches {	1✔
179	if m.options.DeferredSort {	1✔
180	go b.requestSort(false) // While waiting, might as well sort.	×
181	}	1✔
182		1✔
183	atomic.AddUint64(&m.stats.TotExecuteBatchWaitBeg, 1)	1✔
184	m.stackDirtyTopCond.Wait()	1✔
185	atomic.AddUint64(&m.stats.TotExecuteBatchWaitEnd, 1)	1✔
186	}	1✔
187		1✔
188	numDirtyTop := 0	30✔
189	if m.stackDirtyTop != nil {	30✔
190	numDirtyTop = len(m.stackDirtyTop.a)	19✔
191	}	1✔
192		1✔
193	stackDirtyTop.a = make([]*segment, 0, numDirtyTop+1)	30✔
194		1✔
195	if m.stackDirtyTop != nil {	30✔
196	stackDirtyTop.a = append(stackDirtyTop.a, m.stackDirtyTop.a...)	19✔
197	}	1✔
198		1✔
199	stackDirtyTop.a = append(stackDirtyTop.a, b)	30✔
200		1✔
201	prevStackDirtyTop := m.stackDirtyTop	30✔
202	m.stackDirtyTop = stackDirtyTop	30✔
203		1✔
204	awakeMergerCh := m.awakeMergerCh	30✔
205	m.awakeMergerCh = nil	30✔
206		1✔
207	m.m.Unlock()	30✔
208		1✔
209	prevStackDirtyTop.Close()	30✔
210		1✔
211	if awakeMergerCh != nil {	30✔
212	atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerBeg, 1)	4✔
213	close(awakeMergerCh)	4✔
214	atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerEnd, 1)	4✔
215	}	1✔
216		1✔
217	atomic.AddUint64(&m.stats.TotExecuteBatchEnd, 1)	30✔
218		1✔
219	return nil	30✔
220	}	1✔
221
222	// ------------------------------------------------------
223
224	// WaitForMerger blocks until the merger has run another cycle.
225	// Providing a kind of "mergeAll" forces a full merge and can be
226	// useful for applications that are no longer performing mutations and
227	// that want to optimize for retrievals.
228	func (m *collection) WaitForMerger(kind string) error {	1✔
229	atomic.AddUint64(&m.stats.TotWaitForMergerBeg, 1)	×
230		1✔
231	pongCh := make(chan struct{})	×
232	m.pingMergerCh <- ping{	×
233	kind: kind,	1✔
234	pongCh: pongCh,	1✔
235	}	1✔
236	<-pongCh	×
237		1✔
238	atomic.AddUint64(&m.stats.TotWaitForMergerEnd, 1)	×
239	return nil	×
240	}	1✔
241
242	// ------------------------------------------------------
243
244	// Log invokes the user's configured Log callback, if any, if the
245	// debug levels are met.
246	func (m *collection) Logf(format string, a ...interface{}) {	1✔
247	if m.options.Debug > 0 &&	2,200,377✔
248	m.options.Log != nil {	1✔
249	m.options.Log(format, a...)	×
250	}	1✔
251	}	1✔
252
253	// OnError invokes the user's configured OnError callback, in which
254	// the application might take further action, for example, such as
255	// Close()'ing the Collection in order to fix underlying
256	// storage/resource issues.
257	func (m *collection) OnError(err error) {	1✔
258	atomic.AddUint64(&m.stats.TotOnError, 1)	2,200,399✔
259		1✔
260	if m.options.OnError != nil {	2,200,399✔
261	m.options.OnError(err)	2,200,414✔
262	}	1✔
263	}	1✔
264
265	// ------------------------------------------------------
266
267	const snapshotSkipDirtyTop = uint32(0x00000001)
268	const snapshotSkipDirtyMid = uint32(0x00000002)
269	const snapshotSkipDirtyBase = uint32(0x00000004)
270	const snapshotSkipClean = uint32(0x00000008)
271
272	// snapshot() atomically clones the various stacks into a new, single
273	// segmentStack, controllable by skip flags, and also invokes the
274	// optional callback while holding the collection lock.
275	func (m collection) snapshot(skip uint32, cb func(segmentStack)) (	1✔
276	*segmentStack, int, int, int, int) {	1✔
277	atomic.AddUint64(&m.stats.TotSnapshotInternalBeg, 1)	137✔
278		1✔
279	rv := &segmentStack{collection: m, refs: 1}	137✔
280		1✔
281	heightDirtyTop := 0	137✔
282	heightDirtyMid := 0	137✔
283	heightDirtyBase := 0	137✔
284	heightClean := 0	137✔
285		1✔
286	m.m.Lock()	137✔
287		1✔
288	rv.lowerLevelSnapshot = m.lowerLevelSnapshot.addRef()	137✔
289		1✔
290	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	137✔
291	heightDirtyTop = len(m.stackDirtyTop.a)	121✔
292	}	1✔
293		1✔
294	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	137✔
295	heightDirtyMid = len(m.stackDirtyMid.a)	2✔
296	}	1✔
297		1✔
298	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	137✔
299	heightDirtyBase = len(m.stackDirtyBase.a)	×
300	}	1✔
301		1✔
302	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	137✔
303	heightClean = len(m.stackClean.a)	×
304	}	1✔
305		1✔
306	rv.a = make([]*segment, 0,	137✔
307	heightDirtyTop+heightDirtyMid+heightDirtyBase+heightClean)	1✔
308		1✔
309	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	137✔
310	rv.a = append(rv.a, m.stackClean.a...)	×
311	}	1✔
312		1✔
313	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	137✔
314	rv.a = append(rv.a, m.stackDirtyBase.a...)	×
315	}	1✔
316		1✔
317	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	137✔
318	rv.a = append(rv.a, m.stackDirtyMid.a...)	2✔
319	}	1✔
320		1✔
321	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	137✔
322	rv.a = append(rv.a, m.stackDirtyTop.a...)	121✔
323	}	1✔
324		1✔
325	if cb != nil {	137✔
326	cb(rv)	10✔
327	}	1✔
328		1✔
329	m.m.Unlock()	137✔
330		1✔
331	atomic.AddUint64(&m.stats.TotSnapshotInternalEnd, 1)	137✔
332		1✔
333	return rv, heightClean, heightDirtyBase, heightDirtyMid, heightDirtyTop	137✔
334	}	1✔
335
336	// ------------------------------------------------------
337
338	// runMerger() implements the background merger task.
339	func (m *collection) runMerger() {	1✔
340	defer func() {	1✔
341	close(m.doneMergerCh)	9✔
342		1✔
343	atomic.AddUint64(&m.stats.TotMergerEnd, 1)	9✔
344	}()	1✔
345		1✔
346	pings := []ping{}	10✔
347		1✔
348	defer func() {	1✔
349	replyToPings(pings)	9✔
350	pings = pings[0:0]	9✔
351	}()	1✔
352		1✔
353	OUTER:	10✔
354	for {	1✔
355	atomic.AddUint64(&m.stats.TotMergerLoop, 1)	20✔
356		1✔
357	// ---------------------------------------------	1✔
358	// Notify ping'ers from the previous loop.	1✔
359		1✔
360	replyToPings(pings)	20✔
361	pings = pings[0:0]	20✔
362		1✔
363	// ---------------------------------------------	1✔
364	// Wait for new stackDirtyTop entries and/or pings.	1✔
365		1✔
366	var awakeMergerCh chan struct{}	20✔
367		1✔
368	m.m.Lock()	20✔
369		1✔
370	if m.stackDirtyTop == nil \|\| len(m.stackDirtyTop.a) <= 0 {	20✔
371	m.awakeMergerCh = make(chan struct{})	13✔
372	awakeMergerCh = m.awakeMergerCh	13✔
373	}	1✔
374		1✔
375	m.m.Unlock()	20✔
376		1✔
377	mergeAll := false	20✔
378		1✔
379	if awakeMergerCh != nil {	20✔
380	atomic.AddUint64(&m.stats.TotMergerWaitBeg, 1)	13✔
381		1✔
382	select {	13✔
383	case <-m.stopCh:	1✔
384	return	9✔
385		1✔
386	case ping := <-m.pingMergerCh:	1✔
387	pings = append(pings, ping)	×
388	if ping.kind == "mergeAll" {	×
389	mergeAll = true	×
390	}	1✔
391		1✔
392	case <-awakeMergerCh:	1✔
393	// NO-OP.	1✔
394	}	1✔
395		1✔
396	atomic.AddUint64(&m.stats.TotMergerWaitEnd, 1)	3✔
397	}	1✔
398		1✔
399	pings, mergeAll =	10✔
400	receivePings(m.pingMergerCh, pings, "mergeAll", mergeAll)	1✔
401		1✔
402	// ---------------------------------------------	1✔
403	// Atomically ingest stackDirtyTop into stackDirtyMid.	1✔
404		1✔
405	var stackDirtyTopPrev *segmentStack	10✔
406	var stackDirtyMidPrev *segmentStack	10✔
407		1✔
408	stackDirtyMid, _, _, prevLenDirtyMid, prevLenDirtyTop :=	10✔
409	m.snapshot(snapshotSkipClean\|snapshotSkipDirtyBase,	1✔
410	func(ss *segmentStack) {	1✔
411	// m.stackDirtyMid takes 1 refs, and	1✔
412	// stackDirtyMid takes 1 refs.	1✔
413	ss.refs++	10✔
414		1✔
415	stackDirtyTopPrev = m.stackDirtyTop	10✔
416	m.stackDirtyTop = nil	10✔
417		1✔
418	stackDirtyMidPrev = m.stackDirtyMid	10✔
419	m.stackDirtyMid = ss	10✔
420		1✔
421	// Awake any writers that are waiting for more space	1✔
422	// in stackDirtyTop.	1✔
423	m.stackDirtyTopCond.Signal()	10✔
424	})	1✔
425		1✔
426	stackDirtyTopPrev.Close()	10✔
427	stackDirtyMidPrev.Close()	10✔
428		1✔
429	// ---------------------------------------------	1✔
430	// Merge multiple stackDirtyMid layers.	1✔
431		1✔
432	if len(stackDirtyMid.a) > 1 {	10✔
433	newTopLevel := 0	2✔
434		1✔
435	if !mergeAll {	2✔
436	// If we have not been asked to merge all segments,	1✔
437	// then heuristically calc a newTopLevel.	1✔
438	newTopLevel = stackDirtyMid.calcTargetTopLevel()	2✔
439	}	1✔
440		1✔
441	if newTopLevel <= 0 {	2✔
442	atomic.AddUint64(&m.stats.TotMergerAll, 1)	2✔
443	}	1✔
444		1✔
445	atomic.AddUint64(&m.stats.TotMergerInternalBeg, 1)	2✔
446		1✔
447	mergedStackDirtyMid, err := stackDirtyMid.merge(newTopLevel)	2✔
448	if err != nil {	2✔
449	atomic.AddUint64(&m.stats.TotMergerInternalErr, 1)	×
450		1✔
451	m.Logf("collection: runMerger stackDirtyMid.merge,"+	×
452	" newTopLevel: %d, err: %v", newTopLevel, err)	1✔
453		1✔
454	m.OnError(err)	×
455		1✔
456	continue OUTER	×
457	}	1✔
458		1✔
459	atomic.AddUint64(&m.stats.TotMergerInternalEnd, 1)	2✔
460		1✔
461	stackDirtyMid.Close()	2✔
462		1✔
463	mergedStackDirtyMid.addRef()	2✔
464	stackDirtyMid = mergedStackDirtyMid	2✔
465		1✔
466	m.m.Lock()	2✔
467	stackDirtyMidPrev = m.stackDirtyMid	2✔
468	m.stackDirtyMid = mergedStackDirtyMid	2✔
469	m.m.Unlock()	2✔
470		1✔
471	stackDirtyMidPrev.Close()	2✔
472	} else {	1✔
473	atomic.AddUint64(&m.stats.TotMergerInternalSkip, 1)	8✔
474	}	1✔
475		1✔
476	lenDirtyMid := len(stackDirtyMid.a)	10✔
477	if lenDirtyMid > 0 {	10✔
478	topDirtyMid := stackDirtyMid.a[lenDirtyMid-1]	10✔
479		1✔
480	m.Logf("collection: runMerger,"+	10✔
481	" dirtyTop prev height: %2d,"+	1✔
482	" dirtyMid height: %2d (%2d),"+	1✔
483	" dirtyMid top (%0.2f kvs cap, %0.2f buf cap) # entries: %d",	1✔
484	prevLenDirtyTop, lenDirtyMid, lenDirtyMid-prevLenDirtyMid,	1✔
485	float64(len(topDirtyMid.kvs))/float64(cap(topDirtyMid.kvs)),	1✔
486	float64(len(topDirtyMid.buf))/float64(cap(topDirtyMid.buf)),	1✔
487	topDirtyMid.Len())	1✔
488	}	1✔
489		1✔
490	stackDirtyMid.Close()	10✔
491		1✔
492	// ---------------------------------------------	1✔
493	// Notify persister.	1✔
494		1✔
495	if m.options.LowerLevelUpdate != nil {	10✔
496	m.m.Lock()	7✔
497	if m.stackDirtyBase == nil &&	7✔
498	m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0 {	1✔
499	atomic.AddUint64(&m.stats.TotMergerLowerLevelNotify, 1)	7✔
500		1✔
501	m.stackDirtyBase = m.stackDirtyMid	7✔
502	m.stackDirtyMid = nil	7✔
503		1✔
504	m.stackDirtyBaseCond.Signal()	7✔
505	} else {	1✔
506	atomic.AddUint64(&m.stats.TotMergerLowerLevelNotifySkip, 1)	×
507	}	1✔
508	m.m.Unlock()	7✔
509	}	1✔
510		1✔
511	atomic.AddUint64(&m.stats.TotMergerLoopRepeat, 1)	10✔
512	}	1✔
513		1✔
514	// TODO: Concurrent merging of disjoint slices of stackDirtyMid	1✔
515	// instead of the current, single-threaded merger?	1✔
516	//	1✔
517	// TODO: A busy merger means no feeding of the persister?	1✔
518	//	1✔
519	// TODO: Delay merger until lots of deletion tombstones?	1✔
520	//	1✔
521	// TODO: The base layer is likely the largest, so instead of heap	1✔
522	// merging the base layer entries, treat the base layer with	1✔
523	// special case to binary search to find better start points?	1✔
524	//	1✔
525	// TODO: Dynamically calc'ed soft max dirty top height, for	1✔
526	// read-heavy (favor lower) versus write-heavy (favor higher)	1✔
527	// situations?	1✔
528	}	1✔
529
530	// ------------------------------------------------------
531
532	// Stats returns stats for this collection.
533	func (m collection) Stats() (CollectionStats, error) {	1✔
534	rv := &CollectionStats{}	1✔
535	m.stats.AtomicCopyTo(rv)	1✔
536	return rv, nil	1✔
537	}	1✔
538
539	// AtomicCopyTo copies stats from s to r (from source to result).
540	func (s CollectionStats) AtomicCopyTo(r CollectionStats) {	1✔
541	rve := reflect.ValueOf(r).Elem()	3✔
542	sve := reflect.ValueOf(s).Elem()	3✔
543	svet := sve.Type()	3✔
544	for i := 0; i < svet.NumField(); i++ {	3✔
545	rvef := rve.Field(i)	132✔
546	svef := sve.Field(i)	132✔
547	if rvef.CanAddr() && svef.CanAddr() {	132✔
548	rvefp := rvef.Addr().Interface()	132✔
549	svefp := svef.Addr().Interface()	132✔
550	atomic.StoreUint64(rvefp.(*uint64),	132✔
551	atomic.LoadUint64(svefp.(*uint64)))	1✔
552	}	1✔
553	}	1✔
554	}	1✔

couchbase / moss / 20

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous