couchbase / moss / 11

pending completion

Build # 11

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/259baf203261f01e333ba5fc338c83554f9bea36">259baf203<a href="https://github.com/couchbase/moss/commit/259baf203261f01e333ba5fc338c83554f9bea36">&quot;&gt;renamed to Logf() to pass go vet

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/0a97ee410a3dfa211f970a8c6c80664ad2d7e329">0a97ee410</a>">0a97ee410</a><a href="https://github.com/couchbase/moss/commit/259baf203261f01e333ba5fc338c83554f9bea36">
Reviewed-on: http://review.couchbase.org/60610
Reviewed-by: Steve Yen <steve.yen@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

1027 of 1096 relevant lines covered (93.7%)

53681.94 hits per line

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

94.83

/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"
)

// 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.
        stackDirtyTop *segmentStack

        // The merger asynchronously grabs all segments from stackDirtyTop
        // and atomically moves them into stackDirtyMid.
        stackDirtyMid *segmentStack

        // stackDirtyBase represents the segments currently being
        // optionally persisted.  Will be nil when persistence is not
        // being used.
        stackDirtyBase *segmentStack

        // stackClean represents the segments that have been persisted,
        // and can be safely evicted, as the lowerLevelSnapshot will have
        // those entries.  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
}

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

// 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 {
        close(m.stopCh)

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

        <-m.doneMergerCh
        <-m.donePersisterCh

        m.m.Lock()
        if m.lowerLevelSnapshot != nil {
                m.lowerLevelSnapshot.Close()
                m.lowerLevelSnapshot = nil
        }
        m.m.Unlock()

        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) {
        rv, _, _, _, _ := m.snapshot(0, nil)

        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) {
        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 {
        b, ok := bIn.(*segment)
        if !ok {
                return fmt.Errorf("wrong Batch implementation type")
        }

        if b == nil || len(b.kvs) <= 0 {
                return nil
        }

        maxStackDirtyTopHeight := m.options.MaxStackDirtyTopHeight
        if maxStackDirtyTopHeight <= 0 {
                maxStackDirtyTopHeight =
                        DefaultCollectionOptions.MaxStackDirtyTopHeight
        }

        sort.Sort(b)

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

        m.m.Lock()

        for m.stackDirtyTop != nil &&
                len(m.stackDirtyTop.a) >= maxStackDirtyTopHeight {
                m.stackDirtyTopCond.Wait()
        }

        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 {
                close(awakeMergerCh)
        }

        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 {
        pongCh := make(chan struct{})
        m.pingMergerCh <- ping{
                kind:   kind,
                pongCh: pongCh,
        }
        <-pongCh
        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) {
        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) {
        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()

        return rv, heightClean, heightDirtyBase, heightDirtyMid, heightDirtyTop
}

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

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

        pings := []ping{}

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

OUTER:
        for {
                // ---------------------------------------------
                // 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 {
                        select {
                        case <-m.stopCh:
                                return

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

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

                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 += 1

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

                        mergedStackDirtyMid, err := stackDirtyMid.merge(newTopLevel)
                        if err != nil {
                                m.Logf("collection: runMerger stackDirtyMid.merge,"+
                                        " newTopLevel: %d, err: %v", newTopLevel, err)

                                m.OnError(err)

                                continue OUTER
                        }

                        stackDirtyMid.Close()

                        mergedStackDirtyMid.addRef()
                        stackDirtyMid = mergedStackDirtyMid

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

                        stackDirtyMidPrev.Close()
                }

                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)),
                                len(topDirtyMid.kvs)/2)
                }

                stackDirtyMid.Close()

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

                if m.options.LowerLevelUpdate != nil {
                        m.m.Lock()
                        if m.stackDirtyBase == nil &&
                                m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0 {
                                m.stackDirtyBase = m.stackDirtyMid
                                m.stackDirtyMid = nil

                                m.stackDirtyBaseCond.Signal()
                        }
                        m.m.Unlock()
                }
        }

        // 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?
}

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	)
19
20	// A collection implements the Collection interface.
21	type collection struct {
22	options CollectionOptions
23
24	stopCh chan struct{}
25	pingMergerCh chan ping
26	doneMergerCh chan struct{}
27	donePersisterCh chan struct{}
28
29	m sync.Mutex // Protects the fields that follow.
30
31	// When ExecuteBatch() has pushed a new segment onto
32	// stackDirtyTop, it notifies the merger via awakeMergerCh (if
33	// non-nil).
34	awakeMergerCh chan struct{}
35
36	// stackDirtyTopCond is used to wait for space in stackDirtyTop.
37	stackDirtyTopCond *sync.Cond
38
39	// stackDirtyBaseCond is used to wait for non-nil stackDirtyBase.
40	stackDirtyBaseCond *sync.Cond
41
42	// ExecuteBatch() will push new segments onto stackDirtyTop.
43	stackDirtyTop *segmentStack
44
45	// The merger asynchronously grabs all segments from stackDirtyTop
46	// and atomically moves them into stackDirtyMid.
47	stackDirtyMid *segmentStack
48
49	// stackDirtyBase represents the segments currently being
50	// optionally persisted. Will be nil when persistence is not
51	// being used.
52	stackDirtyBase *segmentStack
53
54	// stackClean represents the segments that have been persisted,
55	// and can be safely evicted, as the lowerLevelSnapshot will have
56	// those entries. Will be nil when persistence is not being used.
57	stackClean *segmentStack
58
59	// lowerLevelSnapshot provides an optional, lower-level storage
60	// implementation, when using the Collection as a cache.
61	lowerLevelSnapshot *snapshotWrapper
62	}
63
64	// ------------------------------------------------------
65
66	// Start kicks off required background gouroutines.
67	func (m *collection) Start() error {	1✔
68	go m.runMerger()	7✔
69	go m.runPersister()	7✔
70	return nil	7✔
71	}	1✔
72
73	// Close synchronously stops background goroutines.
74	func (m *collection) Close() error {	1✔
75	close(m.stopCh)	6✔
76		1✔
77	m.stackDirtyBaseCond.Signal() // Awake persister.	6✔
78		1✔
79	<-m.doneMergerCh	6✔
80	<-m.donePersisterCh	6✔
81		1✔
82	m.m.Lock()	6✔
83	if m.lowerLevelSnapshot != nil {	6✔
84	m.lowerLevelSnapshot.Close()	1✔
85	m.lowerLevelSnapshot = nil	1✔
86	}	1✔
87	m.m.Unlock()	6✔
88		1✔
89	return nil	6✔
90	}	1✔
91
92	// Options returns the current options.
93	func (m *collection) Options() CollectionOptions {	1✔
94	return m.options	×
95	}	1✔
96
97	// Snapshot returns a stable snapshot of the key-value entries.
98	func (m *collection) Snapshot() (Snapshot, error) {	1✔
99	rv, _, _, _, _ := m.snapshot(0, nil)	109✔
100		1✔
101	return rv, nil	109✔
102	}	1✔
103
104	// NewBatch returns a new Batch instance with hinted amount of
105	// resources expected to be required.
106	func (m *collection) NewBatch(totalOps, totalKeyValBytes int) (	1✔
107	Batch, error) {	1✔
108	return newSegment(totalOps, totalKeyValBytes)	28✔
109	}	1✔
110
111	// ExecuteBatch atomically incorporates the provided Batch into the
112	// collection. The Batch instance should not be reused after
113	// ExecuteBatch() returns.
114	func (m *collection) ExecuteBatch(bIn Batch,	1✔
115	writeOptions WriteOptions) error {	1✔
116	b, ok := bIn.(*segment)	27✔
117	if !ok {	27✔
118	return fmt.Errorf("wrong Batch implementation type")	×
119	}	1✔
120		1✔
121	if b == nil \|\| len(b.kvs) <= 0 {	27✔
122	return nil	1✔
123	}	1✔
124		1✔
125	maxStackDirtyTopHeight := m.options.MaxStackDirtyTopHeight	26✔
126	if maxStackDirtyTopHeight <= 0 {	26✔
127	maxStackDirtyTopHeight =	26✔
128	DefaultCollectionOptions.MaxStackDirtyTopHeight	1✔
129	}	1✔
130		1✔
131	sort.Sort(b)	26✔
132		1✔
133	stackDirtyTop := &segmentStack{collection: m, refs: 1}	26✔
134		1✔
135	m.m.Lock()	26✔
136		1✔
137	for m.stackDirtyTop != nil &&	26✔
138	len(m.stackDirtyTop.a) >= maxStackDirtyTopHeight {	1✔
139	m.stackDirtyTopCond.Wait()	1✔
140	}	1✔
141		1✔
142	numDirtyTop := 0	26✔
143	if m.stackDirtyTop != nil {	26✔
144	numDirtyTop = len(m.stackDirtyTop.a)	19✔
145	}	1✔
146		1✔
147	stackDirtyTop.a = make([]*segment, 0, numDirtyTop+1)	26✔
148		1✔
149	if m.stackDirtyTop != nil {	26✔
150	stackDirtyTop.a = append(stackDirtyTop.a, m.stackDirtyTop.a...)	19✔
151	}	1✔
152		1✔
153	stackDirtyTop.a = append(stackDirtyTop.a, b)	26✔
154		1✔
155	prevStackDirtyTop := m.stackDirtyTop	26✔
156	m.stackDirtyTop = stackDirtyTop	26✔
157		1✔
158	awakeMergerCh := m.awakeMergerCh	26✔
159	m.awakeMergerCh = nil	26✔
160		1✔
161	m.m.Unlock()	26✔
162		1✔
163	prevStackDirtyTop.Close()	26✔
164		1✔
165	if awakeMergerCh != nil {	26✔
166	close(awakeMergerCh)	2✔
167	}	1✔
168		1✔
169	return nil	26✔
170	}	1✔
171
172	// ------------------------------------------------------
173
174	// WaitForMerger blocks until the merger has run another cycle.
175	// Providing a kind of "mergeAll" forces a full merge and can be
176	// useful for applications that are no longer performing mutations and
177	// that want to optimize for retrievals.
178	func (m *collection) WaitForMerger(kind string) error {	1✔
179	pongCh := make(chan struct{})	×
180	m.pingMergerCh <- ping{	×
181	kind: kind,	1✔
182	pongCh: pongCh,	1✔
183	}	1✔
184	<-pongCh	×
185	return nil	×
186	}	1✔
187
188	// ------------------------------------------------------
189
190	// Log invokes the user's configured Log callback, if any, if the
191	// debug levels are met.
192	func (m *collection) Logf(format string, a ...interface{}) {	1✔
193	if m.options.Debug > 0 &&	3,886,306✔
194	m.options.Log != nil {	1✔
195	m.options.Log(format, a...)	×
196	}	1✔
197	}	1✔
198
199	// OnError invokes the user's configured OnError callback, in which
200	// the application might take further action, for example, such as
201	// Close()'ing the Collection in order to fix underlying
202	// storage/resource issues.
203	func (m *collection) OnError(err error) {	1✔
204	if m.options.OnError != nil {	3,886,299✔
205	m.options.OnError(err)	3,886,299✔
206	}	1✔
207	}	1✔
208
209	// ------------------------------------------------------
210
211	const snapshotSkipDirtyTop = uint32(0x00000001)
212	const snapshotSkipDirtyMid = uint32(0x00000002)
213	const snapshotSkipDirtyBase = uint32(0x00000004)
214	const snapshotSkipClean = uint32(0x00000008)
215
216	// snapshot() atomically clones the various stacks into a new, single
217	// segmentStack, controllable by skip flags, and also invokes the
218	// optional callback while holding the collection lock.
219	func (m collection) snapshot(skip uint32, cb func(segmentStack)) (	1✔
220	*segmentStack, int, int, int, int) {	1✔
221	rv := &segmentStack{collection: m, refs: 1}	116✔
222		1✔
223	heightDirtyTop := 0	116✔
224	heightDirtyMid := 0	116✔
225	heightDirtyBase := 0	116✔
226	heightClean := 0	116✔
227		1✔
228	m.m.Lock()	116✔
229		1✔
230	rv.lowerLevelSnapshot = m.lowerLevelSnapshot.addRef()	116✔
231		1✔
232	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	116✔
233	heightDirtyTop = len(m.stackDirtyTop.a)	106✔
234	}	1✔
235		1✔
236	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	116✔
237	heightDirtyMid = len(m.stackDirtyMid.a)	3✔
238	}	1✔
239		1✔
240	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	116✔
241	heightDirtyBase = len(m.stackDirtyBase.a)	×
242	}	1✔
243		1✔
244	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	116✔
245	heightClean = len(m.stackClean.a)	×
246	}	1✔
247		1✔
248	rv.a = make([]*segment, 0,	116✔
249	heightDirtyTop+heightDirtyMid+heightDirtyBase+heightClean)	1✔
250		1✔
251	if m.stackClean != nil && (skip&snapshotSkipClean == 0) {	116✔
252	rv.a = append(rv.a, m.stackClean.a...)	×
253	}	1✔
254		1✔
255	if m.stackDirtyBase != nil && (skip&snapshotSkipDirtyBase == 0) {	116✔
256	rv.a = append(rv.a, m.stackDirtyBase.a...)	×
257	}	1✔
258		1✔
259	if m.stackDirtyMid != nil && (skip&snapshotSkipDirtyMid == 0) {	116✔
260	rv.a = append(rv.a, m.stackDirtyMid.a...)	3✔
261	}	1✔
262		1✔
263	if m.stackDirtyTop != nil && (skip&snapshotSkipDirtyTop == 0) {	116✔
264	rv.a = append(rv.a, m.stackDirtyTop.a...)	106✔
265	}	1✔
266		1✔
267	if cb != nil {	116✔
268	cb(rv)	7✔
269	}	1✔
270		1✔
271	m.m.Unlock()	116✔
272		1✔
273	return rv, heightClean, heightDirtyBase, heightDirtyMid, heightDirtyTop	116✔
274	}	1✔
275
276	// ------------------------------------------------------
277
278	// runMerger() implements the background merger task.
279	func (m *collection) runMerger() {	1✔
280	defer close(m.doneMergerCh)	7✔
281		1✔
282	pings := []ping{}	7✔
283		1✔
284	defer func() {	1✔
285	replyToPings(pings)	6✔
286	pings = pings[0:0]	6✔
287	}()	1✔
288		1✔
289	OUTER:	7✔
290	for {	1✔
291	// ---------------------------------------------	1✔
292	// Notify ping'ers from the previous loop.	1✔
293		1✔
294	replyToPings(pings)	14✔
295	pings = pings[0:0]	14✔
296		1✔
297	// ---------------------------------------------	1✔
298	// Wait for new stackDirtyTop entries and/or pings.	1✔
299		1✔
300	var awakeMergerCh chan struct{}	14✔
301		1✔
302	m.m.Lock()	14✔
303		1✔
304	if m.stackDirtyTop == nil \|\| len(m.stackDirtyTop.a) <= 0 {	14✔
305	m.awakeMergerCh = make(chan struct{})	9✔
306	awakeMergerCh = m.awakeMergerCh	9✔
307	}	1✔
308		1✔
309	m.m.Unlock()	14✔
310		1✔
311	mergeAll := false	14✔
312		1✔
313	if awakeMergerCh != nil {	14✔
314	select {	9✔
315	case <-m.stopCh:	1✔
316	return	6✔
317		1✔
318	case ping := <-m.pingMergerCh:	1✔
319	pings = append(pings, ping)	×
320	if ping.kind == "mergeAll" {	×
321	mergeAll = true	×
322	}	1✔
323		1✔
324	case <-awakeMergerCh:	1✔
325	// NO-OP.	1✔
326	}	1✔
327	}	1✔
328		1✔
329	pings, mergeAll =	7✔
330	receivePings(m.pingMergerCh, pings, "mergeAll", mergeAll)	1✔
331		1✔
332	// ---------------------------------------------	1✔
333	// Atomically ingest stackDirtyTop into stackDirtyMid.	1✔
334		1✔
335	var stackDirtyTopPrev *segmentStack	7✔
336	var stackDirtyMidPrev *segmentStack	7✔
337		1✔
338	stackDirtyMid, _, _, prevLenDirtyMid, prevLenDirtyTop :=	7✔
339	m.snapshot(snapshotSkipClean\|snapshotSkipDirtyBase,	1✔
340	func(ss *segmentStack) {	1✔
341	// m.stackDirtyMid takes 1 refs, and	1✔
342	// stackDirtyMid takes 1 refs.	1✔
343	ss.refs += 1	7✔
344		1✔
345	stackDirtyTopPrev = m.stackDirtyTop	7✔
346	m.stackDirtyTop = nil	7✔
347		1✔
348	stackDirtyMidPrev = m.stackDirtyMid	7✔
349	m.stackDirtyMid = ss	7✔
350		1✔
351	// Awake any writers that are waiting for more space	1✔
352	// in stackDirtyTop.	1✔
353	m.stackDirtyTopCond.Signal()	7✔
354	})	1✔
355		1✔
356	stackDirtyTopPrev.Close()	7✔
357	stackDirtyMidPrev.Close()	7✔
358		1✔
359	// ---------------------------------------------	1✔
360	// Merge multiple stackDirtyMid layers.	1✔
361		1✔
362	if len(stackDirtyMid.a) > 1 {	7✔
363	newTopLevel := 0	3✔
364		1✔
365	if !mergeAll {	3✔
366	// If we have not been asked to merge all segments,	1✔
367	// then heuristically calc a newTopLevel.	1✔
368	newTopLevel = stackDirtyMid.calcTargetTopLevel()	3✔
369	}	1✔
370		1✔
371	mergedStackDirtyMid, err := stackDirtyMid.merge(newTopLevel)	3✔
372	if err != nil {	3✔
373	m.Logf("collection: runMerger stackDirtyMid.merge,"+	×
374	" newTopLevel: %d, err: %v", newTopLevel, err)	1✔
375		1✔
376	m.OnError(err)	×
377		1✔
378	continue OUTER	×
379	}	1✔
380		1✔
381	stackDirtyMid.Close()	3✔
382		1✔
383	mergedStackDirtyMid.addRef()	3✔
384	stackDirtyMid = mergedStackDirtyMid	3✔
385		1✔
386	m.m.Lock()	3✔
387	stackDirtyMidPrev = m.stackDirtyMid	3✔
388	m.stackDirtyMid = mergedStackDirtyMid	3✔
389	m.m.Unlock()	3✔
390		1✔
391	stackDirtyMidPrev.Close()	3✔
392	}	1✔
393		1✔
394	lenDirtyMid := len(stackDirtyMid.a)	7✔
395	if lenDirtyMid > 0 {	7✔
396	topDirtyMid := stackDirtyMid.a[lenDirtyMid-1]	7✔
397		1✔
398	m.Logf("collection: runMerger,"+	7✔
399	" dirtyTop prev height: %2d,"+	1✔
400	" dirtyMid height: %2d (%2d),"+	1✔
401	" dirtyMid top (%0.2f kvs cap, %0.2f buf cap) # entries: %d",	1✔
402	prevLenDirtyTop, lenDirtyMid, lenDirtyMid-prevLenDirtyMid,	1✔
403	float64(len(topDirtyMid.kvs))/float64(cap(topDirtyMid.kvs)),	1✔
404	float64(len(topDirtyMid.buf))/float64(cap(topDirtyMid.buf)),	1✔
405	len(topDirtyMid.kvs)/2)	1✔
406	}	1✔
407		1✔
408	stackDirtyMid.Close()	7✔
409		1✔
410	// ---------------------------------------------	1✔
411	// Notify persister.	1✔
412		1✔
413	if m.options.LowerLevelUpdate != nil {	7✔
414	m.m.Lock()	3✔
415	if m.stackDirtyBase == nil &&	3✔
416	m.stackDirtyMid != nil && len(m.stackDirtyMid.a) > 0 {	1✔
417	m.stackDirtyBase = m.stackDirtyMid	3✔
418	m.stackDirtyMid = nil	3✔
419		1✔
420	m.stackDirtyBaseCond.Signal()	3✔
421	}	1✔
422	m.m.Unlock()	3✔
423	}	1✔
424	}	1✔
425		1✔
426	// TODO: Concurrent merging of disjoint slices of stackDirtyMid	1✔
427	// instead of the current, single-threaded merger?	1✔
428	//	1✔
429	// TODO: A busy merger means no feeding of the persister?	1✔
430	//	1✔
431	// TODO: Delay merger until lots of deletion tombstones?	1✔
432	//	1✔
433	// TODO: The base layer is likely the largest, so instead of heap	1✔
434	// merging the base layer entries, treat the base layer with	1✔
435	// special case to binary search to find better start points?	1✔
436	//	1✔
437	// TODO: Dynamically calc'ed soft max dirty top height, for	1✔
438	// read-heavy (favor lower) versus write-heavy (favor higher)	1✔
439	// situations?	1✔
440	}	1✔

couchbase / moss / 11

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