couchbase / moss / 25

pending completion

Build # 25

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/8766d1f8b0c916da249512709ee05f45aa799511">8766d1f8b<a href="https://github.com/couchbase/moss/commit/8766d1f8b0c916da249512709ee05f45aa799511">&quot;&gt;MB-18433 - options MaxDirtyOps/KeyValBytes to prevent OOM

Added MaxDirtyOps and MaxDirtyKeyValBytes settings to the
CollectionOptions, which apps can use to limit memory usage by moss.

As part of this, also renamed the awakeMergerCh to
waitDirtyIncomingCh, in order to match the newly added
waitDirtyOutgoingCh.

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/caf0fb9b4af7a11feba5bf85c1b7e31f25f64226">caf0fb9b4</a>">caf0fb9b4</a><a href="https://github.com/couchbase/moss/commit/8766d1f8b0c916da249512709ee05f45aa799511">
Reviewed-on: http://review.couchbase.org/60995
Reviewed-by: Marty Schoch <marty.schoch@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

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

1298 of 1415 relevant lines covered (91.73%)

45264.31 hits per line

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

86.49

/segment_stack.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 (
        "bytes"
        "sync"
)

// A segmentStack is a stack of segments, where higher (later) entries
// in the stack have higher precedence, and should "shadow" any
// entries of the same key from lower in the stack.  A segmentStack
// implements the Snapshot interface.
type segmentStack struct {
        collection *collection

        a []*segment

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

        refs int

        lowerLevelSnapshot *snapshotWrapper
}

func (ss *segmentStack) addRef() {
        ss.m.Lock()
        ss.refs++
        ss.m.Unlock()
}

func (ss *segmentStack) decRef() {
        ss.m.Lock()
        ss.refs--
        if ss.refs <= 0 {
                if ss.lowerLevelSnapshot != nil {
                        ss.lowerLevelSnapshot.Close()
                        ss.lowerLevelSnapshot = nil
                }
        }
        ss.m.Unlock()
}

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

// Close releases associated resources.
func (ss *segmentStack) Close() error {
        if ss != nil {
                ss.decRef()
        }

        return nil
}

// Get retrieves a val from a segmentStack.
func (ss *segmentStack) Get(key []byte,
        readOptions ReadOptions) ([]byte, error) {
        return ss.get(key, len(ss.a)-1)
}

// get() retrieves a val from a segmentStack, but only considers
// segments at or below the segStart level.
func (ss *segmentStack) get(key []byte, segStart int) ([]byte, error) {
        if segStart >= 0 {
                ss.ensureSorted(0, segStart)

                for seg := segStart; seg >= 0; seg-- {
                        b := ss.a[seg]

                        operation, k, v :=
                                b.getOperationKeyVal(b.findStartKeyInclusivePos(key))
                        if k != nil && bytes.Equal(k, key) {
                                if operation == OperationDel {
                                        return nil, nil
                                }

                                if operation == OperationMerge {
                                        return ss.getMerged(k, v, seg-1)
                                }

                                return v, nil
                        }
                }
        }

        if ss.lowerLevelSnapshot != nil {
                return ss.lowerLevelSnapshot.Get(key, ReadOptions{})
        }

        return nil, nil
}

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

// getMerged() retrieves a lower level val for a given key and returns
// a merged val, based on the configured merge operator.
func (ss *segmentStack) getMerged(key, val []byte, segStart int) (
        []byte, error) {
        mo := ss.collection.options.MergeOperator
        if mo == nil {
                return nil, ErrMergeOperatorNil
        }

        vLower, err := ss.get(key, segStart)
        if err != nil {
                return nil, err
        }

        vMerged, ok := mo.FullMerge(key, vLower, [][]byte{val})
        if !ok {
                return nil, ErrMergeOperatorFullMergeFailed
        }

        return vMerged, nil
}

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

// calcTargetTopLevel() heuristically computes a new top level that
// the segmentStack should be merged to.
func (ss *segmentStack) calcTargetTopLevel() int {
        minMergePercentage := ss.collection.options.MinMergePercentage
        if minMergePercentage <= 0 {
                minMergePercentage = DefaultCollectionOptions.MinMergePercentage
        }

        newTopLevel := 0
        maxTopLevel := len(ss.a) - 2

        for newTopLevel < maxTopLevel {
                numX0 := ss.a[newTopLevel].Len()
                numX1 := ss.a[newTopLevel+1].Len()
                if (float64(numX1) / float64(numX0)) > minMergePercentage {
                        break
                }

                newTopLevel++
        }

        return newTopLevel
}

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

// merge() returns a new segmentStack, merging all the segments that
// are at the given newTopLevel and higher.
func (ss *segmentStack) merge(newTopLevel int) (*segmentStack, error) {
        // ----------------------------------------------------
        // First, rough estimate the bytes neeeded.

        totOps := ss.a[newTopLevel].Len()
        totBytes := ss.a[newTopLevel].NumKeyValBytes()

        iterPrealloc, err := ss.StartIterator(nil, nil, IteratorOptions{
                IncludeDeletions: true,
                SkipLowerLevel:   true,
                MinSegmentLevel:  newTopLevel + 1,
                MaxSegmentHeight: len(ss.a),
        })
        if err != nil {
                return nil, err
        }

        defer iterPrealloc.Close()

        for {
                _, key, val, err := iterPrealloc.CurrentEx()
                if err == ErrIteratorDone {
                        break
                }
                if err != nil {
                        return nil, err
                }

                totOps++
                totBytes += len(key) + len(val)

                err = iterPrealloc.Next()
                if err == ErrIteratorDone {
                        break
                }
                if err != nil {
                        return nil, err
                }
        }

        // ----------------------------------------------------
        // Next, use an iterator for the actual merge.

        mergedSegment, err := newSegment(totOps, totBytes)
        if err != nil {
                return nil, err
        }

        iter, err := ss.startIterator(nil, nil, IteratorOptions{
                IncludeDeletions: true,
                SkipLowerLevel:   true,
                MinSegmentLevel:  newTopLevel,
                MaxSegmentHeight: len(ss.a),
        })
        if err != nil {
                return nil, err
        }

        defer iter.Close()

        var readOptions ReadOptions

OUTER:
        for {
                entryEx, key, val, err := iter.CurrentEx()
                if err == ErrIteratorDone {
                        break
                }
                if err != nil {
                        return nil, err
                }

                if len(iter.cursors) == 1 {
                        // When only 1 cursor remains, copy the remains of the
                        // last segment more directly instead of Next()'ing
                        // through the iterator.
                        cursor := &iter.cursors[0]

                        segment := iter.ss.a[cursor.ssIndex]
                        segmentOps := segment.Len()

                        for pos := cursor.pos; pos < segmentOps; pos++ {
                                op, k, v := segment.getOperationKeyVal(pos)

                                err = mergedSegment.mutate(op, k, v)
                                if err != nil {
                                        return nil, err
                                }
                        }

                        break OUTER
                }

                op := entryEx.Operation
                if op == OperationMerge {
                        // TODO: the merge operator implementation is currently
                        // inefficient and not lazy enough right now.
                        val, err = iter.ss.Get(key, readOptions)
                        if err != nil {
                                return nil, err
                        }

                        if val == nil {
                                op = OperationDel
                        } else {
                                op = OperationSet
                        }
                }

                err = mergedSegment.mutate(op, key, val)
                if err != nil {
                        return nil, err
                }

                err = iter.Next()
                if err == ErrIteratorDone {
                        break
                }
                if err != nil {
                        return nil, err
                }
        }

        var a []*segment

        a = append(a, ss.a[0:newTopLevel]...)
        a = append(a, mergedSegment)

        return &segmentStack{collection: ss.collection, a: a, refs: 1}, nil
}

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

func (ss *segmentStack) ensureSorted(minSeg, maxSeg int) {
        if !ss.collection.options.DeferredSort {
                return
        }

        sorted := true // Two phases allows for more concurrent sorting.
        for seg := maxSeg; seg >= minSeg; seg-- {
                sorted = sorted && ss.a[seg].requestSort(false)
        }

        if !sorted {
                for seg := maxSeg; seg >= minSeg; seg-- {
                        ss.a[seg].requestSort(true)
                }
        }
}

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

// SegmentStackStats represents the stats for a segmentStack.
type SegmentStackStats struct {
        CurOps      uint64
        CurBytes    uint64 // Counts key-val bytes only, not metadata.
        CurSegments uint64
}

// AddTo adds the values from this SegmentStackStats to the dest
// SegmentStackStats.
func (sss *SegmentStackStats) AddTo(dest *SegmentStackStats) {
        if sss == nil {
                return
        }

        dest.CurOps += sss.CurOps
        dest.CurBytes += sss.CurBytes
        dest.CurBytes += sss.CurBytes
}

// Stats returns the stats for this segment stack.
func (ss *segmentStack) Stats() *SegmentStackStats {
        rv := &SegmentStackStats{CurSegments: uint64(len(ss.a))}
        for _, seg := range ss.a {
                rv.CurOps += uint64(seg.Len())
                rv.CurBytes += uint64(seg.NumKeyValBytes())
        }
        return rv
}

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	"bytes"
16	"sync"
17	)
18
19	// A segmentStack is a stack of segments, where higher (later) entries
20	// in the stack have higher precedence, and should "shadow" any
21	// entries of the same key from lower in the stack. A segmentStack
22	// implements the Snapshot interface.
23	type segmentStack struct {
24	collection *collection
25
26	a []*segment
27
28	m sync.Mutex // Protects the fields the follow.
29
30	refs int
31
32	lowerLevelSnapshot *snapshotWrapper
33	}
34
35	func (ss *segmentStack) addRef() {	1✔
36	ss.m.Lock()	3✔
37	ss.refs++	3✔
38	ss.m.Unlock()	3✔
39	}	1✔
40
41	func (ss *segmentStack) decRef() {	1✔
42	ss.m.Lock()	98✔
43	ss.refs--	98✔
44	if ss.refs <= 0 {	98✔
45	if ss.lowerLevelSnapshot != nil {	84✔
46	ss.lowerLevelSnapshot.Close()	6✔
47	ss.lowerLevelSnapshot = nil	6✔
48	}	1✔
49	}	1✔
50	ss.m.Unlock()	98✔
51	}	1✔
52
53	// ------------------------------------------------------
54
55	// Close releases associated resources.
56	func (ss *segmentStack) Close() error {	1✔
57	if ss != nil {	121✔
58	ss.decRef()	98✔
59	}	1✔
60		1✔
61	return nil	121✔
62	}	1✔
63
64	// Get retrieves a val from a segmentStack.
65	func (ss *segmentStack) Get(key []byte,	1✔
66	readOptions ReadOptions) ([]byte, error) {	1✔
67	return ss.get(key, len(ss.a)-1)	9,199✔
68	}	1✔
69
70	// get() retrieves a val from a segmentStack, but only considers
71	// segments at or below the segStart level.
72	func (ss *segmentStack) get(key []byte, segStart int) ([]byte, error) {	1✔
73	if segStart >= 0 {	9,210✔
74	ss.ensureSorted(0, segStart)	6,167✔
75		1✔
76	for seg := segStart; seg >= 0; seg-- {	6,167✔
77	b := ss.a[seg]	6,379✔
78		1✔
79	operation, k, v :=	6,379✔
80	b.getOperationKeyVal(b.findStartKeyInclusivePos(key))	1✔
81	if k != nil && bytes.Equal(k, key) {	6,379✔
82	if operation == OperationDel {	6,125✔
83	return nil, nil	2,047✔
84	}	1✔
85		1✔
86	if operation == OperationMerge {	4,078✔
87	return ss.getMerged(k, v, seg-1)	7✔
88	}	1✔
89		1✔
90	return v, nil	4,071✔
91	}	1✔
92	}	1✔
93	}	1✔
94		1✔
95	if ss.lowerLevelSnapshot != nil {	3,085✔
96	return ss.lowerLevelSnapshot.Get(key, ReadOptions{})	3,033✔
97	}	1✔
98		1✔
99	return nil, nil	52✔
100	}	1✔
101
102	// ------------------------------------------------------
103
104	// getMerged() retrieves a lower level val for a given key and returns
105	// a merged val, based on the configured merge operator.
106	func (ss *segmentStack) getMerged(key, val []byte, segStart int) (	1✔
107	[]byte, error) {	1✔
108	mo := ss.collection.options.MergeOperator	11✔
109	if mo == nil {	11✔
110	return nil, ErrMergeOperatorNil	×
111	}	1✔
112		1✔
113	vLower, err := ss.get(key, segStart)	11✔
114	if err != nil {	11✔
115	return nil, err	×
116	}	1✔
117		1✔
118	vMerged, ok := mo.FullMerge(key, vLower, [][]byte{val})	11✔
119	if !ok {	11✔
120	return nil, ErrMergeOperatorFullMergeFailed	×
121	}	1✔
122		1✔
123	return vMerged, nil	11✔
124	}	1✔
125
126	// ------------------------------------------------------
127
128	// calcTargetTopLevel() heuristically computes a new top level that
129	// the segmentStack should be merged to.
130	func (ss *segmentStack) calcTargetTopLevel() int {	1✔
131	minMergePercentage := ss.collection.options.MinMergePercentage	3✔
132	if minMergePercentage <= 0 {	3✔
133	minMergePercentage = DefaultCollectionOptions.MinMergePercentage	3✔
134	}	1✔
135		1✔
136	newTopLevel := 0	3✔
137	maxTopLevel := len(ss.a) - 2	3✔
138		1✔
139	for newTopLevel < maxTopLevel {	3✔
140	numX0 := ss.a[newTopLevel].Len()	2✔
141	numX1 := ss.a[newTopLevel+1].Len()	2✔
142	if (float64(numX1) / float64(numX0)) > minMergePercentage {	2✔
143	break	2✔
144	}	1✔
145		1✔
146	newTopLevel++	×
147	}	1✔
148		1✔
149	return newTopLevel	3✔
150	}	1✔
151
152	// ------------------------------------------------------
153
154	// merge() returns a new segmentStack, merging all the segments that
155	// are at the given newTopLevel and higher.
156	func (ss segmentStack) merge(newTopLevel int) (segmentStack, error) {	1✔
157	// ----------------------------------------------------	1✔
158	// First, rough estimate the bytes neeeded.	1✔
159		1✔
160	totOps := ss.a[newTopLevel].Len()	3✔
161	totBytes := ss.a[newTopLevel].NumKeyValBytes()	3✔
162		1✔
163	iterPrealloc, err := ss.StartIterator(nil, nil, IteratorOptions{	3✔
164	IncludeDeletions: true,	1✔
165	SkipLowerLevel: true,	1✔
166	MinSegmentLevel: newTopLevel + 1,	1✔
167	MaxSegmentHeight: len(ss.a),	1✔
168	})	1✔
169	if err != nil {	3✔
170	return nil, err	×
171	}	1✔
172		1✔
173	defer iterPrealloc.Close()	3✔
174		1✔
175	for {	3✔
176	_, key, val, err := iterPrealloc.CurrentEx()	9✔
177	if err == ErrIteratorDone {	9✔
178	break	×
179	}	1✔
180	if err != nil {	9✔
181	return nil, err	×
182	}	1✔
183		1✔
184	totOps++	9✔
185	totBytes += len(key) + len(val)	9✔
186		1✔
187	err = iterPrealloc.Next()	9✔
188	if err == ErrIteratorDone {	9✔
189	break	3✔
190	}	1✔
191	if err != nil {	6✔
192	return nil, err	×
193	}	1✔
194	}	1✔
195		1✔
196	// ----------------------------------------------------	1✔
197	// Next, use an iterator for the actual merge.	1✔
198		1✔
199	mergedSegment, err := newSegment(totOps, totBytes)	3✔
200	if err != nil {	3✔
201	return nil, err	×
202	}	1✔
203		1✔
204	iter, err := ss.startIterator(nil, nil, IteratorOptions{	3✔
205	IncludeDeletions: true,	1✔
206	SkipLowerLevel: true,	1✔
207	MinSegmentLevel: newTopLevel,	1✔
208	MaxSegmentHeight: len(ss.a),	1✔
209	})	1✔
210	if err != nil {	3✔
211	return nil, err	×
212	}	1✔
213		1✔
214	defer iter.Close()	3✔
215		1✔
216	var readOptions ReadOptions	3✔
217		1✔
218	OUTER:	3✔
219	for {	1✔
220	entryEx, key, val, err := iter.CurrentEx()	10✔
221	if err == ErrIteratorDone {	10✔
222	break	×
223	}	1✔
224	if err != nil {	10✔
225	return nil, err	×
226	}	1✔
227		1✔
228	if len(iter.cursors) == 1 {	10✔
229	// When only 1 cursor remains, copy the remains of the	1✔
230	// last segment more directly instead of Next()'ing	1✔
231	// through the iterator.	1✔
232	cursor := &iter.cursors[0]	2✔
233		1✔
234	segment := iter.ss.a[cursor.ssIndex]	2✔
235	segmentOps := segment.Len()	2✔
236		1✔
237	for pos := cursor.pos; pos < segmentOps; pos++ {	2✔
238	op, k, v := segment.getOperationKeyVal(pos)	2✔
239		1✔
240	err = mergedSegment.mutate(op, k, v)	2✔
241	if err != nil {	2✔
242	return nil, err	×
243	}	1✔
244	}	1✔
245		1✔
246	break OUTER	2✔
247	}	1✔
248		1✔
249	op := entryEx.Operation	8✔
250	if op == OperationMerge {	8✔
251	// TODO: the merge operator implementation is currently	1✔
252	// inefficient and not lazy enough right now.	1✔
253	val, err = iter.ss.Get(key, readOptions)	×
254	if err != nil {	×
255	return nil, err	×
256	}	1✔
257		1✔
258	if val == nil {	×
259	op = OperationDel	×
260	} else {	1✔
261	op = OperationSet	×
262	}	1✔
263	}	1✔
264		1✔
265	err = mergedSegment.mutate(op, key, val)	8✔
266	if err != nil {	8✔
267	return nil, err	×
268	}	1✔
269		1✔
270	err = iter.Next()	8✔
271	if err == ErrIteratorDone {	8✔
272	break	1✔
273	}	1✔
274	if err != nil {	7✔
275	return nil, err	×
276	}	1✔
277	}	1✔
278		1✔
279	var a []*segment	3✔
280		1✔
281	a = append(a, ss.a[0:newTopLevel]...)	3✔
282	a = append(a, mergedSegment)	3✔
283		1✔
284	return &segmentStack{collection: ss.collection, a: a, refs: 1}, nil	3✔
285	}	1✔
286
287	// ------------------------------------------------------
288
289	func (ss *segmentStack) ensureSorted(minSeg, maxSeg int) {	1✔
290	if !ss.collection.options.DeferredSort {	6,359✔
291	return	6,359✔
292	}	1✔
293		1✔
294	sorted := true // Two phases allows for more concurrent sorting.	×
295	for seg := maxSeg; seg >= minSeg; seg-- {	×
296	sorted = sorted && ss.a[seg].requestSort(false)	×
297	}	1✔
298		1✔
299	if !sorted {	×
300	for seg := maxSeg; seg >= minSeg; seg-- {	×
301	ss.a[seg].requestSort(true)	×
302	}	1✔
303	}	1✔
304	}	1✔
305
306	// ------------------------------------------------------
307
308	// SegmentStackStats represents the stats for a segmentStack.
309	type SegmentStackStats struct {
310	CurOps uint64
311	CurBytes uint64 // Counts key-val bytes only, not metadata.
312	CurSegments uint64
313	}
314
315	// AddTo adds the values from this SegmentStackStats to the dest
316	// SegmentStackStats.
317	func (sss SegmentStackStats) AddTo(dest SegmentStackStats) {	1✔
318	if sss == nil {	3✔
319	return	3✔
320	}	1✔
321		1✔
322	dest.CurOps += sss.CurOps	×
323	dest.CurBytes += sss.CurBytes	×
324	dest.CurBytes += sss.CurBytes	×
325	}	1✔
326
327	// Stats returns the stats for this segment stack.
328	func (ss segmentStack) Stats() SegmentStackStats {	1✔
329	rv := &SegmentStackStats{CurSegments: uint64(len(ss.a))}	×
330	for _, seg := range ss.a {	×
331	rv.CurOps += uint64(seg.Len())	×
332	rv.CurBytes += uint64(seg.NumKeyValBytes())	×
333	}	1✔
334	return rv	×
335	}	1✔

couchbase / moss / 25

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