couchbase / moss / 177

pending completion

Build # 177

Build Type

push

travis-ci

Committed by

abhinavdangeti

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/cc50ce321e7b7eeb80f852adbde2943393debd78">cc50ce321<a href="https://github.com/couchbase/moss/commit/cc50ce321e7b7eeb80f852adbde2943393debd78">&quot;&gt;Fix help for mossScope:dump

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/7e38fba918d44872f6d2f582876ceb1add40a24e">7e38fba91</a>">7e38fba91</a><a href="https://github.com/couchbase/moss/commit/cc50ce321e7b7eeb80f852adbde2943393debd78">
Reviewed-on: http://review.couchbase.org/73209
Reviewed-by: Steve Yen &lt;steve.yen@gmail.com>
Tested-by: abhinav dangeti <abhinav@couchbase.com>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

2706 of 3394 relevant lines covered (79.73%)

8480.61 hits per line

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

92.41

/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"
        "sort"
        "sync"
        "sync/atomic"
        "time"
)

// 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 can notify waiters like the merger via
        // waitDirtyIncomingCh (if non-nil).
        waitDirtyIncomingCh chan struct{}

        // When the persister has finished a persistence cycle, it can
        // notify waiters like the merger via waitDirtyOutgoingCh (if
        // non-nil).
        waitDirtyOutgoingCh 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 {
        if !m.options.ReadOnly {
                // Kick off merger and persister only when not in Read-Only mode
                go m.runMerger()
                go m.runPersister()
        }
        return nil
}

// Close synchronously stops background goroutines.
func (m *collection) Close() error {
        m.fireEvent(EventKindCloseStart, 0)
        startTime := time.Now()
        defer func() {
                m.fireEvent(EventKindClose, time.Now().Sub(startTime))
        }()

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

        m.m.Lock()

        close(m.stopCh)

        m.stackDirtyTopCond.Broadcast()  // Awake all ExecuteBatch()'ers.
        m.stackDirtyBaseCond.Broadcast() // Awake persister.

        if !m.options.ReadOnly {
                m.m.Unlock()

                <-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)
        }

        stackDirtyTopPrev := m.stackDirtyTop
        m.stackDirtyTop = nil

        stackDirtyMidPrev := m.stackDirtyMid
        m.stackDirtyMid = nil

        stackDirtyBasePrev := m.stackDirtyBase
        m.stackDirtyBase = nil

        stackCleanPrev := m.stackClean
        m.stackClean = nil

        m.m.Unlock()

        stackDirtyTopPrev.Close()
        stackDirtyMidPrev.Close()
        stackDirtyBasePrev.Close()
        stackCleanPrev.Close()

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

        return nil
}

func (m *collection) isClosed() bool {
        select {
        case <-m.stopCh:
                return true
        default:
                return false
        }
}

// 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) {
        if m.isClosed() {
                return nil, ErrClosed
        }

        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) {
        if m.isClosed() {
                return nil, ErrClosed
        }

        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 {
        startTime := time.Now()
        defer func() {
                m.fireEvent(EventKindBatchExecute, time.Now().Sub(startTime))
        }()

        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{options: &m.options, refs: 1}

        // notify interested handlers that we are about to execute this batch
        m.fireEvent(EventKindBatchExecuteStart, 0)

        m.m.Lock()

        for m.stackDirtyTop != nil &&
                len(m.stackDirtyTop.a) >= maxPreMergerBatches {
                if m.isClosed() {
                        m.m.Unlock()
                        return ErrClosed
                }

                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)
        }

        // check again, could have been closed while waiting
        if m.isClosed() {
                m.m.Unlock()
                return ErrClosed
        }

        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

        waitDirtyIncomingCh := m.waitDirtyIncomingCh
        m.waitDirtyIncomingCh = nil

        m.m.Unlock()

        prevStackDirtyTop.Close()

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

        atomic.AddUint64(&m.stats.TotExecuteBatchEnd, 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)
        }
}

func (m *collection) fireEvent(kind EventKind, dur time.Duration) {
        if m.options.OnEvent != nil {
                m.options.OnEvent(Event{Kind: kind, Collection: m, Duration: dur})
        }
}

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

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{options: &m.options, 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
}

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	"sort"
17	"sync"
18	"sync/atomic"
19	"time"
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 can notify waiters like the merger via
35	// waitDirtyIncomingCh (if non-nil).
36	waitDirtyIncomingCh chan struct{}
37
38	// When the persister has finished a persistence cycle, it can
39	// notify waiters like the merger via waitDirtyOutgoingCh (if
40	// non-nil).
41	waitDirtyOutgoingCh chan struct{}
42
43	// ----------------------------------------
44
45	// stackDirtyTopCond is used to wait for space in stackDirtyTop.
46	stackDirtyTopCond *sync.Cond
47
48	// stackDirtyBaseCond is used to wait for non-nil stackDirtyBase.
49	stackDirtyBaseCond *sync.Cond
50
51	// ----------------------------------------
52
53	// ExecuteBatch() will push new segments onto stackDirtyTop if
54	// there is space.
55	stackDirtyTop *segmentStack
56
57	// The merger goroutine asynchronously, atomically grabs all
58	// segments from stackDirtyTop and atomically moves them into
59	// stackDirtyMid. The merger will also merge segments in
60	// stackDirtyMid to keep its height low.
61	stackDirtyMid *segmentStack
62
63	// stackDirtyBase represents the segments currently being
64	// persisted. It is optionally populated by the merger when there
65	// are merged segments ready for persistence. Will be nil when
66	// persistence is not being used.
67	stackDirtyBase *segmentStack
68
69	// stackClean represents the segments that have been optionally
70	// persisted by the persister, and can now be safely evicted, as
71	// the lowerLevelSnapshot will contain the entries from
72	// stackClean. Will be nil when persistence is not being used.
73	stackClean *segmentStack
74
75	// lowerLevelSnapshot provides an optional, lower-level storage
76	// implementation, when using the Collection as a cache.
77	lowerLevelSnapshot *snapshotWrapper
78
79	// stats leverage sync/atomic counters.
80	stats *CollectionStats
81	}
82
83	// ------------------------------------------------------
84
85	// Start kicks off required background gouroutines.
86	func (m *collection) Start() error {	70✔
87	if !m.options.ReadOnly {	139✔
88	// Kick off merger and persister only when not in Read-Only mode	69✔
89	go m.runMerger()	69✔
90	go m.runPersister()	69✔
91	}	69✔
92	return nil	70✔
93	}
94
95	// Close synchronously stops background goroutines.
96	func (m *collection) Close() error {	61✔
97	m.fireEvent(EventKindCloseStart, 0)	61✔
98	startTime := time.Now()	61✔
99	defer func() {	122✔
100	m.fireEvent(EventKindClose, time.Now().Sub(startTime))	61✔
101	}()	61✔
102
103	atomic.AddUint64(&m.stats.TotCloseBeg, 1)	61✔
104		61✔
105	m.m.Lock()	61✔
106		61✔
107	close(m.stopCh)	61✔
108		61✔
109	m.stackDirtyTopCond.Broadcast() // Awake all ExecuteBatch()'ers.	61✔
110	m.stackDirtyBaseCond.Broadcast() // Awake persister.	61✔
111		61✔
112	if !m.options.ReadOnly {	121✔
113	m.m.Unlock()	60✔
114		60✔
115	<-m.doneMergerCh	60✔
116	atomic.AddUint64(&m.stats.TotCloseMergerDone, 1)	60✔
117		60✔
118	<-m.donePersisterCh	60✔
119	atomic.AddUint64(&m.stats.TotClosePersisterDone, 1)	60✔
120		60✔
121	m.m.Lock()	60✔
122	}	60✔
123
124	if m.lowerLevelSnapshot != nil {	90✔
125	atomic.AddUint64(&m.stats.TotCloseLowerLevelBeg, 1)	29✔
126	m.lowerLevelSnapshot.Close()	29✔
127	m.lowerLevelSnapshot = nil	29✔
128	atomic.AddUint64(&m.stats.TotCloseLowerLevelEnd, 1)	29✔
129	}	29✔
130
131	stackDirtyTopPrev := m.stackDirtyTop	61✔
132	m.stackDirtyTop = nil	61✔
133		61✔
134	stackDirtyMidPrev := m.stackDirtyMid	61✔
135	m.stackDirtyMid = nil	61✔
136		61✔
137	stackDirtyBasePrev := m.stackDirtyBase	61✔
138	m.stackDirtyBase = nil	61✔
139		61✔
140	stackCleanPrev := m.stackClean	61✔
141	m.stackClean = nil	61✔
142		61✔
143	m.m.Unlock()	61✔
144		61✔
145	stackDirtyTopPrev.Close()	61✔
146	stackDirtyMidPrev.Close()	61✔
147	stackDirtyBasePrev.Close()	61✔
148	stackCleanPrev.Close()	61✔
149		61✔
150	atomic.AddUint64(&m.stats.TotCloseEnd, 1)	61✔
151		61✔
152	return nil	61✔
153	}
154
155	func (m *collection) isClosed() bool {	1,027✔
156	select {	1,027✔
157	case <-m.stopCh:	59✔
158	return true	59✔
159	default:	968✔
160	return false	968✔
161	}
162	}
163
164	// Options returns the current options.
165	func (m *collection) Options() CollectionOptions {	1✔
166	return m.options	1✔
167	}	1✔
168
169	// Snapshot returns a stable snapshot of the key-value entries.
170	func (m *collection) Snapshot() (Snapshot, error) {	205✔
171	if m.isClosed() {	206✔
172	return nil, ErrClosed	1✔
173	}	1✔
174
175	atomic.AddUint64(&m.stats.TotSnapshotBeg, 1)	204✔
176		204✔
177	rv, _, _, _, _ := m.snapshot(0, nil)	204✔
178		204✔
179	atomic.AddUint64(&m.stats.TotSnapshotEnd, 1)	204✔
180		204✔
181	return rv, nil	204✔
182	}
183
184	// NewBatch returns a new Batch instance with hinted amount of
185	// resources expected to be required.
186	func (m *collection) NewBatch(totalOps, totalKeyValBytes int) (
187	Batch, error) {	257✔
188	if m.isClosed() {	258✔
189	return nil, ErrClosed	1✔
190	}	1✔
191
192	atomic.AddUint64(&m.stats.TotNewBatch, 1)	256✔
193	atomic.AddUint64(&m.stats.TotNewBatchTotalOps, uint64(totalOps))	256✔
194	atomic.AddUint64(&m.stats.TotNewBatchTotalKeyValBytes, uint64(totalKeyValBytes))	256✔
195		256✔
196	return newSegment(totalOps, totalKeyValBytes)	256✔
197	}
198
199	// ExecuteBatch atomically incorporates the provided Batch into the
200	// collection. The Batch instance should not be reused after
201	// ExecuteBatch() returns.
202	func (m *collection) ExecuteBatch(bIn Batch,
203	writeOptions WriteOptions) error {	255✔
204	startTime := time.Now()	255✔
205	defer func() {	510✔
206	m.fireEvent(EventKindBatchExecute, time.Now().Sub(startTime))	255✔
207	}()	255✔
208
209	atomic.AddUint64(&m.stats.TotExecuteBatchBeg, 1)	255✔
210		255✔
211	b, ok := bIn.(*segment)	255✔
212	if !ok {	255✔
213	atomic.AddUint64(&m.stats.TotExecuteBatchErr, 1)	×
214		×
215	return fmt.Errorf("wrong Batch implementation type")	×
216	}	×
217
218	if b == nil \|\| b.Len() <= 0 {	256✔
219	atomic.AddUint64(&m.stats.TotExecuteBatchEmpty, 1)	1✔
220	return nil	1✔
221	}	1✔
222
223	maxPreMergerBatches := m.options.MaxPreMergerBatches	254✔
224	if maxPreMergerBatches <= 0 {	508✔
225	maxPreMergerBatches =	254✔
226	DefaultCollectionOptions.MaxPreMergerBatches	254✔
227	}	254✔
228
229	if m.options.DeferredSort {	296✔
230	b.needSorterCh = make(chan bool, 1)	42✔
231	b.needSorterCh <- true // A ticket for the future sorter.	42✔
232	close(b.needSorterCh)	42✔
233		42✔
234	b.waitSortedCh = make(chan struct{})	42✔
235	} else {	254✔
236	sort.Sort(b)	212✔
237	}	212✔
238
239	stackDirtyTop := &segmentStack{options: &m.options, refs: 1}	254✔
240		254✔
241	// notify interested handlers that we are about to execute this batch	254✔
242	m.fireEvent(EventKindBatchExecuteStart, 0)	254✔
243		254✔
244	m.m.Lock()	254✔
245		254✔
246	for m.stackDirtyTop != nil &&	254✔
247	len(m.stackDirtyTop.a) >= maxPreMergerBatches {	254✔
248	if m.isClosed() {	×
249	m.m.Unlock()	×
250	return ErrClosed	×
251	}	×
252
253	if m.options.DeferredSort {	×
254	go b.RequestSort(false) // While waiting, might as well sort.	×
255	}	×
256
257	atomic.AddUint64(&m.stats.TotExecuteBatchWaitBeg, 1)	×
258	m.stackDirtyTopCond.Wait()	×
259	atomic.AddUint64(&m.stats.TotExecuteBatchWaitEnd, 1)	×
260	}
261
262	// check again, could have been closed while waiting
263	if m.isClosed() {	255✔
264	m.m.Unlock()	1✔
265	return ErrClosed	1✔
266	}	1✔
267
268	numDirtyTop := 0	253✔
269	if m.stackDirtyTop != nil {	325✔
270	numDirtyTop = len(m.stackDirtyTop.a)	72✔
271	}	72✔
272
273	stackDirtyTop.a = make([]Segment, 0, numDirtyTop+1)	253✔
274		253✔
275	if m.stackDirtyTop != nil {	325✔
276	stackDirtyTop.a = append(stackDirtyTop.a, m.stackDirtyTop.a...)	72✔
277	}	72✔
278
279	stackDirtyTop.a = append(stackDirtyTop.a, b)	253✔
280		253✔
281	prevStackDirtyTop := m.stackDirtyTop	253✔
282	m.stackDirtyTop = stackDirtyTop	253✔
283		253✔
284	waitDirtyIncomingCh := m.waitDirtyIncomingCh	253✔
285	m.waitDirtyIncomingCh = nil	253✔
286		253✔
287	m.m.Unlock()	253✔
288		253✔
289	prevStackDirtyTop.Close()	253✔
290		253✔
291	if waitDirtyIncomingCh != nil {	374✔
292	atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerBeg, 1)	121✔
293	close(waitDirtyIncomingCh)	121✔
294	atomic.AddUint64(&m.stats.TotExecuteBatchAwakeMergerEnd, 1)	121✔
295	}	121✔
296
297	atomic.AddUint64(&m.stats.TotExecuteBatchEnd, 1)	253✔
298		253✔
299	return nil	253✔
300	}
301
302	// ------------------------------------------------------
303
304	// Log invokes the user's configured Log callback, if any, if the
305	// debug levels are met.
306	func (m *collection) Logf(format string, a ...interface{}) {	186✔
307	if m.options.Debug > 0 &&	186✔
308	m.options.Log != nil {	187✔
309	m.options.Log(format, a...)	1✔
310	}	1✔
311	}
312
313	// OnError invokes the user's configured OnError callback, in which
314	// the application might take further action, for example, such as
315	// Close()'ing the Collection in order to fix underlying
316	// storage/resource issues.
317	func (m *collection) OnError(err error) {	2✔
318	atomic.AddUint64(&m.stats.TotOnError, 1)	2✔
319		2✔
320	if m.options.OnError != nil {	4✔
321	m.options.OnError(err)	2✔
322	}	2✔
323	}
324
325	func (m *collection) fireEvent(kind EventKind, dur time.Duration) {	937✔
326	if m.options.OnEvent != nil {	1,635✔
327	m.options.OnEvent(Event{Kind: kind, Collection: m, Duration: dur})	698✔
328	}	698✔
329	}
330
331	// ------------------------------------------------------
332
333	const snapshotSkipDirtyTop = uint32(0x00000001)
334	const snapshotSkipDirtyMid = uint32(0x00000002)
335	const snapshotSkipDirtyBase = uint32(0x00000004)
336	const snapshotSkipClean = uint32(0x00000008)
337
338	// snapshot() atomically clones the various stacks into a new, single
339	// segmentStack, controllable by skip flags, and also invokes the
340	// optional callback while holding the collection lock.
341	func (m collection) snapshot(skip uint32, cb func(segmentStack)) (
342	*segmentStack, int, int, int, int) {	392✔
343	atomic.AddUint64(&m.stats.TotSnapshotInternalBeg, 1)	392✔
344		392✔
345	rv := &segmentStack{options: &m.options, refs: 1}	392✔
346		392✔
347	heightDirtyTop := 0	392✔
348	heightDirtyMid := 0	392✔
349	heightDirtyBase := 0	392✔
350	heightClean := 0	392✔
351		392✔
352	m.m.Lock()	392✔
353		392✔
354	rv.lowerLevelSnapshot = m.lowerLevelSnapshot.addRef()	392✔
355		392✔
356	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	716✔
357	heightDirtyTop = len(m.stackDirtyTop.a)	324✔
358	}	324✔
359
360	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	516✔
361	heightDirtyMid = len(m.stackDirtyMid.a)	124✔
362	}	124✔
363
364	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	405✔
365	heightDirtyBase = len(m.stackDirtyBase.a)	13✔
366	}	13✔
367
368	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	392✔
369	heightClean = len(m.stackClean.a)	×
370	}	×
371
372	rv.a = make([]Segment, 0,	392✔
373	heightDirtyTop+heightDirtyMid+heightDirtyBase+heightClean)	392✔
374		392✔
375	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	392✔
376	rv.a = append(rv.a, m.stackClean.a...)	×
377	}	×
378
379	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	405✔
380	rv.a = append(rv.a, m.stackDirtyBase.a...)	13✔
381	}	13✔
382
383	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	516✔
384	rv.a = append(rv.a, m.stackDirtyMid.a...)	124✔
385	}	124✔
386
387	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	716✔
388	rv.a = append(rv.a, m.stackDirtyTop.a...)	324✔
389	}	324✔
390
391	if cb != nil {	578✔
392	cb(rv)	186✔
393	}	186✔
394
395	m.m.Unlock()	392✔
396		392✔
397	atomic.AddUint64(&m.stats.TotSnapshotInternalEnd, 1)	392✔
398		392✔
399	return rv, heightClean, heightDirtyBase, heightDirtyMid, heightDirtyTop	392✔
400	}

couchbase / moss / 177

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