couchbase / moss / 12

pending completion

Build # 12

Build Type

push

travis-ci

Committed by

steveyen

Commit Message

Added helper funcs to hide segment's kvs/buf slices

Only the segment implementation should know of its underlying kvs and
buf slices, so using helper functions that provide for this
information hiding.

Change-Id: I34fdbc5e6
Reviewed-on: http://review.couchbase.org/60611
Reviewed-by: Marty Schoch <marty.schoch@gmail.com>
Tested-by: Steve Yen <steve.yen@gmail.com>

Run Details

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

1025 of 1099 relevant lines covered (93.27%)

49035.45 hits per line

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

90.18

/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 += 1
        ss.m.Unlock()
}

func (ss *segmentStack) decRef() {
        ss.m.Lock()
        ss.refs -= 1
        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() retreives 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 {
                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 += 1
        }

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

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()	2✔
37	ss.refs += 1	2✔
38	ss.m.Unlock()	2✔
39	}	1✔
40
41	func (ss *segmentStack) decRef() {	1✔
42	ss.m.Lock()	82✔
43	ss.refs -= 1	82✔
44	if ss.refs <= 0 {	82✔
45	if ss.lowerLevelSnapshot != nil {	74✔
46	ss.lowerLevelSnapshot.Close()	1✔
47	ss.lowerLevelSnapshot = nil	1✔
48	}	1✔
49	}	1✔
50	ss.m.Unlock()	82✔
51	}	1✔
52
53	// ------------------------------------------------------
54
55	// Close releases associated resources.
56	func (ss *segmentStack) Close() error {	1✔
57	if ss != nil {	96✔
58	ss.decRef()	82✔
59	}	1✔
60		1✔
61	return nil	96✔
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)	100✔
68	}	1✔
69
70	// get() retreives 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 {	111✔
74	for seg := segStart; seg >= 0; seg-- {	101✔
75	b := ss.a[seg]	380✔
76		1✔
77	operation, k, v :=	380✔
78	b.getOperationKeyVal(b.findStartKeyInclusivePos(key))	1✔
79	if k != nil && bytes.Equal(k, key) {	380✔
80	if operation == OperationDel {	59✔
81	return nil, nil	25✔
82	}	1✔
83		1✔
84	if operation == OperationMerge {	34✔
85	return ss.getMerged(k, v, seg-1)	7✔
86	}	1✔
87		1✔
88	return v, nil	27✔
89	}	1✔
90	}	1✔
91	}	1✔
92		1✔
93	if ss.lowerLevelSnapshot != nil {	52✔
94	return ss.lowerLevelSnapshot.Get(key, ReadOptions{})	×
95	}	1✔
96		1✔
97	return nil, nil	52✔
98	}	1✔
99
100	// ------------------------------------------------------
101
102	// getMerged() retrieves a lower level val for a given key and returns
103	// a merged val, based on the configured merge operator.
104	func (ss *segmentStack) getMerged(key, val []byte, segStart int) (	1✔
105	[]byte, error) {	1✔
106	mo := ss.collection.options.MergeOperator	11✔
107	if mo == nil {	11✔
108	return nil, ErrMergeOperatorNil	×
109	}	1✔
110		1✔
111	vLower, err := ss.get(key, segStart)	11✔
112	if err != nil {	11✔
113	return nil, err	×
114	}	1✔
115		1✔
116	vMerged, ok := mo.FullMerge(key, vLower, [][]byte{val})	11✔
117	if !ok {	11✔
118	return nil, ErrMergeOperatorFullMergeFailed	×
119	}	1✔
120		1✔
121	return vMerged, nil	11✔
122	}	1✔
123
124	// ------------------------------------------------------
125
126	// calcTargetTopLevel() heuristically computes a new top level that
127	// the segmentStack should be merged to.
128	func (ss *segmentStack) calcTargetTopLevel() int {	1✔
129	minMergePercentage := ss.collection.options.MinMergePercentage	2✔
130	if minMergePercentage <= 0 {	2✔
131	minMergePercentage = DefaultCollectionOptions.MinMergePercentage	2✔
132	}	1✔
133		1✔
134	newTopLevel := 0	2✔
135	maxTopLevel := len(ss.a) - 2	2✔
136		1✔
137	for newTopLevel < maxTopLevel {	2✔
138	numX0 := ss.a[newTopLevel].Len()	2✔
139	numX1 := ss.a[newTopLevel+1].Len()	2✔
140	if (float64(numX1) / float64(numX0)) > minMergePercentage {	2✔
141	break	2✔
142	}	1✔
143		1✔
144	newTopLevel += 1	×
145	}	1✔
146		1✔
147	return newTopLevel	2✔
148	}	1✔
149
150	// ------------------------------------------------------
151
152	// merge() returns a new segmentStack, merging all the segments that
153	// are at the given newTopLevel and higher.
154	func (ss segmentStack) merge(newTopLevel int) (segmentStack, error) {	1✔
155	// ----------------------------------------------------	1✔
156	// First, rough estimate the bytes neeeded.	1✔
157		1✔
158	totOps := ss.a[newTopLevel].Len()	2✔
159	totBytes := ss.a[newTopLevel].NumKeyValBytes()	2✔
160		1✔
161	iterPrealloc, err := ss.StartIterator(nil, nil, IteratorOptions{	2✔
162	IncludeDeletions: true,	1✔
163	SkipLowerLevel: true,	1✔
164	MinSegmentLevel: newTopLevel + 1,	1✔
165	MaxSegmentHeight: len(ss.a),	1✔
166	})	1✔
167	if err != nil {	2✔
168	return nil, err	×
169	}	1✔
170		1✔
171	defer iterPrealloc.Close()	2✔
172		1✔
173	for {	2✔
174	_, key, val, err := iterPrealloc.CurrentEx()	9✔
175	if err == ErrIteratorDone {	9✔
176	break	×
177	}	1✔
178	if err != nil {	9✔
179	return nil, err	×
180	}	1✔
181		1✔
182	totOps += 1	9✔
183	totBytes += len(key) + len(val)	9✔
184		1✔
185	err = iterPrealloc.Next()	9✔
186	if err == ErrIteratorDone {	9✔
187	break	2✔
188	}	1✔
189	if err != nil {	7✔
190	return nil, err	×
191	}	1✔
192	}	1✔
193		1✔
194	// ----------------------------------------------------	1✔
195	// Next, use an iterator for the actual merge.	1✔
196		1✔
197	mergedSegment, err := newSegment(totOps, totBytes)	2✔
198	if err != nil {	2✔
199	return nil, err	×
200	}	1✔
201		1✔
202	iter, err := ss.startIterator(nil, nil, IteratorOptions{	2✔
203	IncludeDeletions: true,	1✔
204	SkipLowerLevel: true,	1✔
205	MinSegmentLevel: newTopLevel,	1✔
206	MaxSegmentHeight: len(ss.a),	1✔
207	})	1✔
208	if err != nil {	2✔
209	return nil, err	×
210	}	1✔
211		1✔
212	defer iter.Close()	2✔
213		1✔
214	var readOptions ReadOptions	2✔
215		1✔
216	OUTER:	2✔
217	for {	1✔
218	entryEx, key, val, err := iter.CurrentEx()	9✔
219	if err == ErrIteratorDone {	9✔
220	break	×
221	}	1✔
222	if err != nil {	9✔
223	return nil, err	×
224	}	1✔
225		1✔
226	if len(iter.cursors) == 1 {	9✔
227	// When only 1 cursor remains, copy the remains of the	1✔
228	// last segment more directly instead of Next()'ing	1✔
229	// through the iterator.	1✔
230	cursor := &iter.cursors[0]	1✔
231		1✔
232	segment := iter.ss.a[cursor.ssIndex]	1✔
233	segmentOps := segment.Len()	1✔
234		1✔
235	for pos := cursor.pos; pos < segmentOps; pos++ {	1✔
236	op, k, v := segment.getOperationKeyVal(pos)	1✔
237		1✔
238	err = mergedSegment.mutate(op, k, v)	1✔
239	if err != nil {	1✔
240	return nil, err	×
241	}	1✔
242	}	1✔
243		1✔
244	break OUTER	1✔
245	}	1✔
246		1✔
247	op := entryEx.Operation	8✔
248	if op == OperationMerge {	8✔
249	// TODO: the merge operator implementation is currently	1✔
250	// inefficient and not lazy enough right now.	1✔
251	val, err = iter.ss.Get(key, readOptions)	×
252	if err != nil {	×
253	return nil, err	×
254	}	1✔
255		1✔
256	if val == nil {	×
257	op = OperationDel	×
258	} else {	1✔
259	op = OperationSet	×
260	}	1✔
261	}	1✔
262		1✔
263	err = mergedSegment.mutate(op, key, val)	8✔
264	if err != nil {	8✔
265	return nil, err	×
266	}	1✔
267		1✔
268	err = iter.Next()	8✔
269	if err == ErrIteratorDone {	8✔
270	break	1✔
271	}	1✔
272	if err != nil {	7✔
273	return nil, err	×
274	}	1✔
275	}	1✔
276		1✔
277	var a []*segment	2✔
278		1✔
279	a = append(a, ss.a[0:newTopLevel]...)	2✔
280	a = append(a, mergedSegment)	2✔
281		1✔
282	return &segmentStack{collection: ss.collection, a: a, refs: 1}, nil	2✔
283	}	1✔

couchbase / moss / 12

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