couchbase / moss / 193

Build # 193

Build Type

push

travis-ci

Committed by Sreekanth Sivasankaran

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/5dc185c3c6dca4cdb2249804feaeae84b1db5663">5dc185c3c<a href="https://github.com/couchbase/moss/commit/5dc185c3c6dca4cdb2249804feaeae84b1db5663">&quot;&gt;MB-22410 - New CloseEx(options) api for fast store closings

New CloseEx(options) api in Store for an aborted close.
Improving comments.
Adding nil checks in Store Close(),and generic errorcode.

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/50364c75cc883ed92b50ad7c7c88b4ef00b4e984">50364c75c</a>">50364c75c</a><a href="https://github.com/couchbase/moss/commit/5dc185c3c6dca4cdb2249804feaeae84b1db5663">
Reviewed-on: http://review.couchbase.org/74385
Reviewed-by: Steve Yen <steve.yen@gmail.com>
Tested-by: Sreekanth Sivasankaran

Run Details

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

3164 of 3999 relevant lines covered (79.12%)

10199.71 hits per line

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

82.61

/segment_stack_merge.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

// calcTargetTopLevel() heuristically computes a new top level that
// the segmentStack should be merged to.
func (ss *segmentStack) calcTargetTopLevel() int {
        var minMergePercentage float64
        if ss.options != nil {
                minMergePercentage = ss.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(mergeAll bool, base *segmentStack) (
        *segmentStack, uint64, error) {
        newTopLevel := 0
        var numFullMerges uint64
        if !mergeAll {
                // If we have not been asked to merge all segments,
                // then heuristically calc a newTopLevel.
                newTopLevel = ss.calcTargetTopLevel()
        }
        if newTopLevel <= 0 {
                numFullMerges++
        }

        // ----------------------------------------------------
        // First, rough estimate the bytes neeeded.

        var totOps int
        var totKeyBytes, totValBytes uint64
        for i := newTopLevel; i < len(ss.a); i++ {
                totOps += ss.a[i].Len()
                nk, nv := ss.a[i].NumKeyValBytes()
                totKeyBytes += nk
                totValBytes += nv
        }

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

        mergedSegment, err := newSegment(totOps, int(totKeyBytes+totValBytes))
        if err != nil {
                return nil, 0, err
        }

        err = ss.mergeInto(newTopLevel, len(ss.a), mergedSegment, base, true,
                true, nil)
        if err != nil {
                return nil, 0, err
        }

        a := make([]Segment, 0, newTopLevel+1)
        a = append(a, ss.a[0:newTopLevel]...)
        a = append(a, mergedSegment)

        rv := &segmentStack{
                options:            ss.options,
                a:                  a,
                refs:               1,
                lowerLevelSnapshot: ss.lowerLevelSnapshot.addRef(),
                incarNum:           ss.incarNum,
        }

        // ---------------------------------------------------
        // Recursively merge all the child segmentStacks with the base stack
        // Dropping any deleted collections present in base but not in me.
        for cName, childSegStack := range ss.childSegStacks {
                var baseSegStack *segmentStack
                if base != nil {
                        var exists bool
                        if base.childSegStacks != nil {
                                baseSegStack, exists = base.childSegStacks[cName]
                        }
                        if exists {
                                if baseSegStack.incarNum != childSegStack.incarNum {
                                        // The base segment stack carries a child collection
                                        // which was subsequently recreated.
                                        baseSegStack = nil
                                        // The dirtyBase's old segmentStacks will be closed by
                                        // the collection merger after successful merge.
                                }
                        }
                }
                mergedChildStack, fullMerges, err := childSegStack.merge(mergeAll,
                        baseSegStack)
                if err != nil {
                        rv.Close()
                        return nil, numFullMerges, err
                }
                numFullMerges += fullMerges
                if len(rv.childSegStacks) == 0 {
                        rv.childSegStacks = make(map[string]*segmentStack)
                }
                rv.childSegStacks[cName] = mergedChildStack
        }

        return rv, numFullMerges, nil
}

func (ss *segmentStack) mergeInto(minSegmentLevel, maxSegmentHeight int,
        dest SegmentMutator, base *segmentStack, includeDeletions, optimizeTail bool,
        cancelCh chan struct{}) error {
        cancelCheckEvery := ss.options.MergerCancelCheckEvery
        if cancelCheckEvery <= 0 {
                cancelCheckEvery = DefaultCollectionOptions.MergerCancelCheckEvery
        }

        iter, err := ss.startIterator(nil, nil, IteratorOptions{
                IncludeDeletions: includeDeletions,
                SkipLowerLevel:   true,
                MinSegmentLevel:  minSegmentLevel,
                MaxSegmentHeight: maxSegmentHeight,
                base:             base,
        })
        if err != nil {
                return err
        }

        defer iter.Close()

        readOptions := ReadOptions{NoCopyValue: true}

OUTER:
        for i := 0; true; i++ {
                if cancelCh != nil && i%cancelCheckEvery == 0 {
                        select {
                        case <-cancelCh:
                                return ErrAborted
                        default:
                                // NO-OP.
                        }
                }

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

                if optimizeTail && 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 = dest.Mutate(op, k, v)
                                if err != nil {
                                        return 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 = ss.get(key, len(ss.a)-1, base, readOptions)
                        if err != nil {
                                return err
                        }

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

                err = dest.Mutate(op, key, val)
                if err != nil {
                        return err
                }

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

        return nil
}

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	// calcTargetTopLevel() heuristically computes a new top level that
15	// the segmentStack should be merged to.
16	func (ss *segmentStack) calcTargetTopLevel() int {	357✔
17	var minMergePercentage float64	357✔
18	if ss.options != nil {	714✔
19	minMergePercentage = ss.options.MinMergePercentage	357✔
20	}	357✔
21	if minMergePercentage <= 0 {	714✔
22	minMergePercentage = DefaultCollectionOptions.MinMergePercentage	357✔
23	}	357✔
24
25	newTopLevel := 0	357✔
26	maxTopLevel := len(ss.a) - 2	357✔
27		357✔
28	for newTopLevel < maxTopLevel {	627✔
29	numX0 := ss.a[newTopLevel].Len()	270✔
30	numX1 := ss.a[newTopLevel+1].Len()	270✔
31	if (float64(numX1) / float64(numX0)) > minMergePercentage {	392✔
32	break	122✔
33	}
34
35	newTopLevel++	148✔
36	}
37
38	return newTopLevel	357✔
39	}
40
41	// ------------------------------------------------------
42
43	// merge() returns a new segmentStack, merging all the segments that
44	// are at the given newTopLevel and higher.
45	func (ss segmentStack) merge(mergeAll bool, base segmentStack) (
46	*segmentStack, uint64, error) {	361✔
47	newTopLevel := 0	361✔
48	var numFullMerges uint64	361✔
49	if !mergeAll {	718✔
50	// If we have not been asked to merge all segments,	357✔
51	// then heuristically calc a newTopLevel.	357✔
52	newTopLevel = ss.calcTargetTopLevel()	357✔
53	}	357✔
54	if newTopLevel <= 0 {	637✔
55	numFullMerges++	276✔
56	}	276✔
57
58	// ----------------------------------------------------
59	// First, rough estimate the bytes neeeded.
60
61	var totOps int	361✔
62	var totKeyBytes, totValBytes uint64	361✔
63	for i := newTopLevel; i < len(ss.a); i++ {	1,703✔
64	totOps += ss.a[i].Len()	1,342✔
65	nk, nv := ss.a[i].NumKeyValBytes()	1,342✔
66	totKeyBytes += nk	1,342✔
67	totValBytes += nv	1,342✔
68	}	1,342✔
69
70	// ----------------------------------------------------
71	// Next, use an iterator for the actual merge.
72
73	mergedSegment, err := newSegment(totOps, int(totKeyBytes+totValBytes))	361✔
74	if err != nil {	361✔
75	return nil, 0, err	×
76	}	×
77
78	err = ss.mergeInto(newTopLevel, len(ss.a), mergedSegment, base, true,	361✔
79	true, nil)	361✔
80	if err != nil {	361✔
81	return nil, 0, err	×
82	}	×
83
84	a := make([]Segment, 0, newTopLevel+1)	361✔
85	a = append(a, ss.a[0:newTopLevel]...)	361✔
86	a = append(a, mergedSegment)	361✔
87		361✔
88	rv := &segmentStack{	361✔
89	options: ss.options,	361✔
90	a: a,	361✔
91	refs: 1,	361✔
92	lowerLevelSnapshot: ss.lowerLevelSnapshot.addRef(),	361✔
93	incarNum: ss.incarNum,	361✔
94	}	361✔
95		361✔
96	// ---------------------------------------------------	361✔
97	// Recursively merge all the child segmentStacks with the base stack	361✔
98	// Dropping any deleted collections present in base but not in me.	361✔
99	for cName, childSegStack := range ss.childSegStacks {	585✔
100	var baseSegStack *segmentStack	224✔
101	if base != nil {	443✔
102	var exists bool	219✔
103	if base.childSegStacks != nil {	438✔
104	baseSegStack, exists = base.childSegStacks[cName]	219✔
105	}	219✔
106	if exists {	438✔
107	if baseSegStack.incarNum != childSegStack.incarNum {	219✔
108	// The base segment stack carries a child collection	×
109	// which was subsequently recreated.	×
110	baseSegStack = nil	×
111	// The dirtyBase's old segmentStacks will be closed by	×
112	// the collection merger after successful merge.	×
113	}	×
114	}
115	}
116	mergedChildStack, fullMerges, err := childSegStack.merge(mergeAll,	224✔
117	baseSegStack)	224✔
118	if err != nil {	224✔
119	rv.Close()	×
120	return nil, numFullMerges, err	×
121	}	×
122	numFullMerges += fullMerges	224✔
123	if len(rv.childSegStacks) == 0 {	314✔
124	rv.childSegStacks = make(map[string]*segmentStack)	90✔
125	}	90✔
126	rv.childSegStacks[cName] = mergedChildStack	224✔
127	}
128
129	return rv, numFullMerges, nil	361✔
130	}
131
132	func (ss *segmentStack) mergeInto(minSegmentLevel, maxSegmentHeight int,
133	dest SegmentMutator, base *segmentStack, includeDeletions, optimizeTail bool,
134	cancelCh chan struct{}) error {	434✔
135	cancelCheckEvery := ss.options.MergerCancelCheckEvery	434✔
136	if cancelCheckEvery <= 0 {	868✔
137	cancelCheckEvery = DefaultCollectionOptions.MergerCancelCheckEvery	434✔
138	}	434✔
139
140	iter, err := ss.startIterator(nil, nil, IteratorOptions{	434✔
141	IncludeDeletions: includeDeletions,	434✔
142	SkipLowerLevel: true,	434✔
143	MinSegmentLevel: minSegmentLevel,	434✔
144	MaxSegmentHeight: maxSegmentHeight,	434✔
145	base: base,	434✔
146	})	434✔
147	if err != nil {	434✔
148	return err	×
149	}	×
150
151	defer iter.Close()	434✔
152		434✔
153	readOptions := ReadOptions{NoCopyValue: true}	434✔
154		434✔
155	OUTER:	434✔
156	for i := 0; true; i++ {	102,740✔
157	if cancelCh != nil && i%cancelCheckEvery == 0 {	102,379✔
158	select {	73✔
159	case <-cancelCh:	×
160	return ErrAborted	×
161	default:	73✔
162	// NO-OP.
163	}
164	}
165
166	entryEx, key, val, err := iter.CurrentEx()	102,306✔
167	if err == ErrIteratorDone {	102,462✔
168	break	156✔
169	}
170	if err != nil {	102,150✔
171	return err	×
172	}	×
173
174	if optimizeTail && len(iter.cursors) == 1 {	102,285✔
175	// When only 1 cursor remains, copy the remains of the	135✔
176	// last segment more directly instead of Next()'ing	135✔
177	// through the iterator.	135✔
178	cursor := iter.cursors[0]	135✔
179		135✔
180	segment := iter.ss.a[cursor.ssIndex]	135✔
181	segmentOps := segment.Len()	135✔
182		135✔
183	for pos := cursor.pos; pos < segmentOps; pos++ {	18,670✔
184	op, k, v := segment.GetOperationKeyVal(pos)	18,535✔
185		18,535✔
186	err = dest.Mutate(op, k, v)	18,535✔
187	if err != nil {	18,535✔
188	return err	×
189	}	×
190	}
191
192	break OUTER	135✔
193	}
194
195	op := entryEx.Operation	102,015✔
196	if op == OperationMerge {	102,025✔
197	// TODO: the merge operator implementation is currently	10✔
198	// inefficient and not lazy enough right now.	10✔
199	val, err = ss.get(key, len(ss.a)-1, base, readOptions)	10✔
200	if err != nil {	10✔
201	return err	×
202	}	×
203
204	if val == nil {	10✔
205	op = OperationDel	×
206	} else {	10✔
207	op = OperationSet	10✔
208	}	10✔
209	}
210
211	err = dest.Mutate(op, key, val)	102,015✔
212	if err != nil {	102,015✔
213	return err	×
214	}	×
215
216	err = iter.Next()	102,015✔
217	if err == ErrIteratorDone {	102,158✔
218	break	143✔
219	}
220	if err != nil {	101,872✔
221	return err	×
222	}	×
223	}
224
225	return nil	434✔
226	}

couchbase / moss / 193

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