couchbase / moss / 255

pending completion

Build # 255

Build Type

push

travis-ci

Committed by

hisundar

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/caa5c9b43f9f2568dfbaa4ecc0e85ec13670610a">caa5c9b43<a href="https://github.com/couchbase/moss/commit/caa5c9b43f9f2568dfbaa4ecc0e85ec13670610a">&quot;&gt;MB-24680: Leveled Compactions Part I - partialCompact file

Allow for partial compaction where all segments from a given
partialCompactStart point are taken, merged into one bigger segment
and appended to the end of the same file, while retaining the
segments before that partialCompactStart point.

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/31c2fc493756c642654a516345bd77f4518c46ff">31c2fc493</a>">31c2fc493</a><a href="https://github.com/couchbase/moss/commit/caa5c9b43f9f2568dfbaa4ecc0e85ec13670610a">
Reviewed-on: http://review.couchbase.org/79339
Well-Formed: Build Bot &lt;build@couchbase.com&gt;
Reviewed-by: Steve Yen &lt;steve.yen@gmail.com>
Tested-by: Sundararaman Sridharan <sundar@couchbase.com>

Run Details

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

3007 of 3628 relevant lines covered (82.88%)

84654.41 hits per line

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

66.87

/store_compact.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"
        "os"
        "time"
)

func (s *Store) compactMaybe(higher Snapshot, persistOptions StorePersistOptions) (
        bool, error) {
        if s.Options().CollectionOptions.ReadOnly {
                // Do not compact in Read-Only mode
                return false, nil
        }

        compactionConcern := persistOptions.CompactionConcern
        if compactionConcern <= 0 {
                return false, nil
        }

        footer, err := s.snapshot()
        if err != nil {
                return false, err
        }

        defer footer.DecRef()

        slocs, ss := footer.segmentLocs()

        defer footer.DecRef()
        var partialCompactStart int

        if compactionConcern == CompactionAllow {
                partialCompactStart = s.calcPartialCompactionStart(slocs, ss, higher,
                        persistOptions)
                if partialCompactStart >= 0 {
                        compactionConcern = CompactionForce
                } // else append data to the end of the same file.
        }

        if compactionConcern != CompactionForce {
                return false, nil
        }

        err = s.compact(footer, partialCompactStart, higher, persistOptions)
        if err != nil {
                return false, err
        }

        var sizeBefore, sizeAfter int64

        if len(slocs) > 0 {
                mref := slocs[0].mref
                if mref != nil && mref.fref != nil {
                        var finfo os.FileInfo
                        if partialCompactStart == 0 {
                                finfo, err = s.removeFileOnClose(mref.fref)
                        } else {
                                finfo, err = mref.fref.file.Stat()
                        }
                        if err == nil && len(finfo.Name()) > 0 {
                                // Fetch size of old file
                                sizeBefore = finfo.Size()
                        }
                }
        }

        slocs, _ = footer.segmentLocs()
        defer footer.DecRef()

        if len(slocs) > 0 {
                mref := slocs[0].mref
                if mref != nil && mref.fref != nil {
                        finfo, err := mref.fref.file.Stat()
                        if err == nil && len(finfo.Name()) > 0 {
                                // Fetch size of new file
                                sizeAfter = finfo.Size()
                        }
                }
        }

        s.m.Lock()
        s.numLastCompactionBeforeBytes = uint64(sizeBefore)
        s.numLastCompactionAfterBytes = uint64(sizeAfter)
        delta := sizeBefore - sizeAfter
        if delta > 0 {
                s.totCompactionDecreaseBytes += uint64(delta)
                if s.maxCompactionDecreaseBytes < uint64(delta) {
                        s.maxCompactionDecreaseBytes = uint64(delta)
                }
        } else if delta < 0 {
                delta = -delta
                s.totCompactionIncreaseBytes += uint64(delta)
                if s.maxCompactionIncreaseBytes < uint64(delta) {
                        s.maxCompactionIncreaseBytes = uint64(delta)
                }
        }
        s.m.Unlock()

        return true, nil
}

// Figure out reverse scan recursive merge size
func (s *Store) calcPartialCompactionStart(slocs SegmentLocs, ss *segmentStack,
        higher Snapshot,
        persistOptions StorePersistOptions) int {
        totUpperLen := 0
        if ss != nil && len(ss.a) >= 2 {
                for i := 1; i < len(ss.a); i++ {
                        totUpperLen += ss.a[i].Len()
                }
        }

        if higher != nil {
                higherSS, ok := higher.(*segmentStack)
                if ok {
                        higherStats := higherSS.Stats()
                        if higherStats != nil {
                                totUpperLen += int(higherStats.CurOps)
                        }
                }
        }

        if totUpperLen > 0 {
                var pct float64
                if ss != nil && len(ss.a) > 0 && ss.a[0].Len() > 0 {
                        pct = float64(totUpperLen) / float64(ss.a[0].Len())
                }

                if pct >= s.options.CompactionPercentage ||
                        (s.options.CompactionMaxSegments > 0 &&
                                len(slocs) >= s.options.CompactionMaxSegments) {
                        return 0
                }
        }

        return -1 // Don't compact, just append to end of same file.
}

func (s *Store) compact(footer *Footer, partialCompactStart int, higher Snapshot,
        persistOptions StorePersistOptions) error {
        startTime := time.Now()

        var newSS *segmentStack
        if higher != nil {
                ssHigher, ok := higher.(*segmentStack)
                if !ok {
                        return fmt.Errorf("store: can only compact higher that's a segmentStack")
                }
                ssHigher.ensureFullySorted()
                newSS = s.mergeSegStacks(footer, partialCompactStart, ssHigher)
        } else {
                newSS = footer.ss // Safe as footer ref count is held positive.
        }

        var frefCompact *FileRef
        var fileCompact File
        var err error
        if partialCompactStart == 0 {
                s.m.Lock()
                frefCompact, fileCompact, err = s.startFileLOCKED()
                s.m.Unlock()
        } else {
                frefCompact, fileCompact, err = s.startOrReuseFile()
        }
        if err != nil {
                return err
        }
        defer frefCompact.DecRef()

        compactFooter, err := s.writeSegments(newSS, frefCompact, fileCompact)
        if err != nil {
                if partialCompactStart == 0 {
                        s.removeFileOnClose(frefCompact)
                }
                return err
        }
        // Prefix restore the footer's partialCompactStart
        if partialCompactStart != 0 {
                compactFooter.spliceFooter(footer, partialCompactStart)
        }

        if s.options != nil && s.options.CompactionSync {
                persistOptions.NoSync = false
        }

        err = s.persistFooter(fileCompact, compactFooter, persistOptions)
        if err != nil {
                if partialCompactStart == 0 {
                        s.removeFileOnClose(frefCompact)
                }
                return err
        }

        err = compactFooter.loadSegments(s.options, frefCompact)
        if err != nil {
                if partialCompactStart == 0 {
                        s.removeFileOnClose(frefCompact)
                }
                return err
        }

        s.m.Lock()
        footerPrev := s.footer
        s.footer = compactFooter // Owns the frefCompact ref-count.
        if partialCompactStart == 0 {
                s.totCompactions++
        } else {
                s.totCompactionsPartial++
        }
        s.m.Unlock()

        s.histograms["CompactUsecs"].Add(
                uint64(time.Since(startTime).Nanoseconds()/1000), 1)

        if footerPrev != nil {
                footerPrev.DecRef()
        }

        return nil
}

func (s *Store) mergeSegStacks(footer *Footer, splicePoint int,
        higher *segmentStack) *segmentStack {
        var footerSS *segmentStack
        var lenFooterSS int
        if footer != nil && footer.ss != nil {
                footerSS = footer.ss
                lenFooterSS = len(footerSS.a)
        }
        rv := &segmentStack{
                options:  higher.options,
                a:        make([]Segment, 0, len(higher.a)+lenFooterSS),
                incarNum: higher.incarNum,
        }
        if footerSS != nil {
                rv.a = append(rv.a, footerSS.a[splicePoint:]...)
        }
        rv.a = append(rv.a, higher.a...)
        for cName, newStack := range higher.childSegStacks {
                if len(rv.childSegStacks) == 0 {
                        rv.childSegStacks = make(map[string]*segmentStack)
                }
                if footer == nil {
                        rv.childSegStacks[cName] = s.mergeSegStacks(nil,
                                splicePoint, newStack)
                        continue
                }

                childFooter, exists := footer.ChildFooters[cName]
                if exists {
                        if childFooter.incarNum != higher.incarNum {
                                // Fast child collection recreation, must not merge
                                // segments from prior incarnation.
                                childFooter = nil
                        }
                }
                rv.childSegStacks[cName] = s.mergeSegStacks(childFooter,
                        splicePoint, newStack)
        }
        return rv
}

func (right *Footer) spliceFooter(left *Footer, splicePoint int) {
        slocs := make([]SegmentLoc, splicePoint, splicePoint+len(right.SegmentLocs))
        copy(slocs, left.SegmentLocs[0:splicePoint])
        slocs = append(slocs, right.SegmentLocs...)
        right.SegmentLocs = slocs

        for cName, childFooter := range right.ChildFooters {
                storeChildFooter, exists := left.ChildFooters[cName]
                if exists {
                        if storeChildFooter.incarNum != childFooter.incarNum {
                                // Fast child collection recreation, ok to drop store footer's
                                // segments from prior incarnation.
                                continue
                        }
                        childFooter.spliceFooter(storeChildFooter, splicePoint)
                }
        }
}

func (s *Store) writeSegments(newSS *segmentStack, frefCompact *FileRef,
        fileCompact File) (compactFooter *Footer, err error) {
        finfo, err := fileCompact.Stat()
        if err != nil {
                return nil, err
        }
        pos := finfo.Size()
        stats := newSS.Stats()

        kvsBegPos := pageAlignCeil(pos)
        bufBegPos := pageAlignCeil(kvsBegPos + 1 + (int64(8+8) * int64(stats.CurOps)))

        compactionBufferPages := 0
        if s.options != nil {
                compactionBufferPages = s.options.CompactionBufferPages
        }
        if compactionBufferPages <= 0 {
                compactionBufferPages = DefaultStoreOptions.CompactionBufferPages
        }
        compactionBufferSize := StorePageSize * compactionBufferPages

        compactWriter := &compactWriter{
                kvsWriter: newBufferedSectionWriter(fileCompact, kvsBegPos, 0, compactionBufferSize),
                bufWriter: newBufferedSectionWriter(fileCompact, bufBegPos, 0, compactionBufferSize),
        }
        onError := func(err error) error {
                compactWriter.kvsWriter.Stop()
                compactWriter.bufWriter.Stop()
                return err
        }

        err = newSS.mergeInto(0, len(newSS.a), compactWriter, nil, false, false, s.abortCh)
        if err != nil {
                return nil, onError(err)
        }

        if err = compactWriter.kvsWriter.Flush(); err != nil {
                return nil, onError(err)
        }
        if err = compactWriter.bufWriter.Flush(); err != nil {
                return nil, onError(err)
        }

        if err = compactWriter.kvsWriter.Stop(); err != nil {
                return nil, onError(err)
        }
        if err = compactWriter.bufWriter.Stop(); err != nil {
                return nil, onError(err)
        }

        compactFooter = &Footer{
                refs: 1,
                SegmentLocs: []SegmentLoc{
                        {
                                Kind:       SegmentKindBasic,
                                KvsOffset:  uint64(kvsBegPos),
                                KvsBytes:   uint64(compactWriter.kvsWriter.Offset() - kvsBegPos),
                                BufOffset:  uint64(bufBegPos),
                                BufBytes:   uint64(compactWriter.bufWriter.Offset() - bufBegPos),
                                TotOpsSet:  compactWriter.totOperationSet,
                                TotOpsDel:  compactWriter.totOperationDel,
                                TotKeyByte: compactWriter.totKeyByte,
                                TotValByte: compactWriter.totValByte,
                        },
                },
        }

        for cName, childSegStack := range newSS.childSegStacks {
                if compactFooter.ChildFooters == nil {
                        compactFooter.ChildFooters = make(map[string]*Footer)
                }
                childFooter, err := s.writeSegments(childSegStack,
                        frefCompact, fileCompact)
                if err != nil {
                        return nil, err
                }
                compactFooter.ChildFooters[cName] = childFooter
        }
        return compactFooter, nil
}

type compactWriter struct {
        file      File
        kvsWriter *bufferedSectionWriter
        bufWriter *bufferedSectionWriter

        totOperationSet   uint64
        totOperationDel   uint64
        totOperationMerge uint64
        totKeyByte        uint64
        totValByte        uint64
}

func (cw *compactWriter) Mutate(operation uint64, key, val []byte) error {
        keyStart := cw.bufWriter.Written()

        _, err := cw.bufWriter.Write(key)
        if err != nil {
                return err
        }

        _, err = cw.bufWriter.Write(val)
        if err != nil {
                return err
        }

        keyLen := len(key)
        valLen := len(val)

        opKlVl := encodeOpKeyLenValLen(operation, keyLen, valLen)

        if keyLen <= 0 && valLen <= 0 {
                keyStart = 0
        }

        pair := []uint64{opKlVl, uint64(keyStart)}
        kvsBuf, err := Uint64SliceToByteSlice(pair)
        if err != nil {
                return err
        }

        _, err = cw.kvsWriter.Write(kvsBuf)
        if err != nil {
                return err
        }

        switch operation {
        case OperationSet:
                cw.totOperationSet++
        case OperationDel:
                cw.totOperationDel++
        case OperationMerge:
                cw.totOperationMerge++
        default:
        }

        cw.totKeyByte += uint64(keyLen)
        cw.totValByte += uint64(valLen)

        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	import (
15	"fmt"
16	"os"
17	"time"
18	)
19
20	func (s *Store) compactMaybe(higher Snapshot, persistOptions StorePersistOptions) (
21	bool, error) {	194✔
22	if s.Options().CollectionOptions.ReadOnly {	195✔
23	// Do not compact in Read-Only mode	1✔
24	return false, nil	1✔
25	}	1✔
26
27	compactionConcern := persistOptions.CompactionConcern	193✔
28	if compactionConcern <= 0 {	260✔
29	return false, nil	67✔
30	}	67✔
31
32	footer, err := s.snapshot()	126✔
33	if err != nil {	126✔
34	return false, err	×
35	}	×
36
37	defer footer.DecRef()	126✔
38		126✔
39	slocs, ss := footer.segmentLocs()	126✔
40		126✔
41	defer footer.DecRef()	126✔
42	var partialCompactStart int	126✔
43		126✔
44	if compactionConcern == CompactionAllow {	177✔
45	partialCompactStart = s.calcPartialCompactionStart(slocs, ss, higher,	51✔
46	persistOptions)	51✔
47	if partialCompactStart >= 0 {	69✔
48	compactionConcern = CompactionForce	18✔
49	} // else append data to the end of the same file.	18✔
50	}
51
52	if compactionConcern != CompactionForce {	159✔
53	return false, nil	33✔
54	}	33✔
55
56	err = s.compact(footer, partialCompactStart, higher, persistOptions)	93✔
57	if err != nil {	93✔
58	return false, err	×
59	}	×
60
61	var sizeBefore, sizeAfter int64	93✔
62		93✔
63	if len(slocs) > 0 {	179✔
64	mref := slocs[0].mref	86✔
65	if mref != nil && mref.fref != nil {	172✔
66	var finfo os.FileInfo	86✔
67	if partialCompactStart == 0 {	172✔
68	finfo, err = s.removeFileOnClose(mref.fref)	86✔
69	} else {	86✔
70	finfo, err = mref.fref.file.Stat()	×
71	}	×
72	if err == nil && len(finfo.Name()) > 0 {	172✔
73	// Fetch size of old file	86✔
74	sizeBefore = finfo.Size()	86✔
75	}	86✔
76	}
77	}
78
79	slocs, _ = footer.segmentLocs()	93✔
80	defer footer.DecRef()	93✔
81		93✔
82	if len(slocs) > 0 {	179✔
83	mref := slocs[0].mref	86✔
84	if mref != nil && mref.fref != nil {	172✔
85	finfo, err := mref.fref.file.Stat()	86✔
86	if err == nil && len(finfo.Name()) > 0 {	172✔
87	// Fetch size of new file	86✔
88	sizeAfter = finfo.Size()	86✔
89	}	86✔
90	}
91	}
92
93	s.m.Lock()	93✔
94	s.numLastCompactionBeforeBytes = uint64(sizeBefore)	93✔
95	s.numLastCompactionAfterBytes = uint64(sizeAfter)	93✔
96	delta := sizeBefore - sizeAfter	93✔
97	if delta > 0 {	93✔
98	s.totCompactionDecreaseBytes += uint64(delta)	×
99	if s.maxCompactionDecreaseBytes < uint64(delta) {	×
100	s.maxCompactionDecreaseBytes = uint64(delta)	×
101	}	×
102	} else if delta < 0 {	93✔
103	delta = -delta	×
104	s.totCompactionIncreaseBytes += uint64(delta)	×
105	if s.maxCompactionIncreaseBytes < uint64(delta) {	×
106	s.maxCompactionIncreaseBytes = uint64(delta)	×
107	}	×
108	}
109	s.m.Unlock()	93✔
110		93✔
111	return true, nil	93✔
112	}
113
114	// Figure out reverse scan recursive merge size
115	func (s Store) calcPartialCompactionStart(slocs SegmentLocs, ss segmentStack,
116	higher Snapshot,
117	persistOptions StorePersistOptions) int {	51✔
118	totUpperLen := 0	51✔
119	if ss != nil && len(ss.a) >= 2 {	79✔
120	for i := 1; i < len(ss.a); i++ {	131✔
121	totUpperLen += ss.a[i].Len()	103✔
122	}	103✔
123	}
124
125	if higher != nil {	102✔
126	higherSS, ok := higher.(*segmentStack)	51✔
127	if ok {	102✔
128	higherStats := higherSS.Stats()	51✔
129	if higherStats != nil {	102✔
130	totUpperLen += int(higherStats.CurOps)	51✔
131	}	51✔
132	}
133	}
134
135	if totUpperLen > 0 {	102✔
136	var pct float64	51✔
137	if ss != nil && len(ss.a) > 0 && ss.a[0].Len() > 0 {	98✔
138	pct = float64(totUpperLen) / float64(ss.a[0].Len())	47✔
139	}	47✔
140
141	if pct >= s.options.CompactionPercentage \|\|	51✔
142	(s.options.CompactionMaxSegments > 0 &&	51✔
143	len(slocs) >= s.options.CompactionMaxSegments) {	69✔
144	return 0	18✔
145	}	18✔
146	}
147
148	return -1 // Don't compact, just append to end of same file.	33✔
149	}
150
151	func (s Store) compact(footer Footer, partialCompactStart int, higher Snapshot,
152	persistOptions StorePersistOptions) error {	93✔
153	startTime := time.Now()	93✔
154		93✔
155	var newSS *segmentStack	93✔
156	if higher != nil {	186✔
157	ssHigher, ok := higher.(*segmentStack)	93✔
158	if !ok {	93✔
159	return fmt.Errorf("store: can only compact higher that's a segmentStack")	×
160	}	×
161	ssHigher.ensureFullySorted()	93✔
162	newSS = s.mergeSegStacks(footer, partialCompactStart, ssHigher)	93✔
163	} else {	×
164	newSS = footer.ss // Safe as footer ref count is held positive.	×
165	}	×
166
167	var frefCompact *FileRef	93✔
168	var fileCompact File	93✔
169	var err error	93✔
170	if partialCompactStart == 0 {	186✔
171	s.m.Lock()	93✔
172	frefCompact, fileCompact, err = s.startFileLOCKED()	93✔
173	s.m.Unlock()	93✔
174	} else {	93✔
175	frefCompact, fileCompact, err = s.startOrReuseFile()	×
176	}	×
177	if err != nil {	93✔
178	return err	×
179	}	×
180	defer frefCompact.DecRef()	93✔
181		93✔
182	compactFooter, err := s.writeSegments(newSS, frefCompact, fileCompact)	93✔
183	if err != nil {	93✔
184	if partialCompactStart == 0 {	×
185	s.removeFileOnClose(frefCompact)	×
186	}	×
187	return err	×
188	}
189	// Prefix restore the footer's partialCompactStart
190	if partialCompactStart != 0 {	93✔
191	compactFooter.spliceFooter(footer, partialCompactStart)	×
192	}	×
193
194	if s.options != nil && s.options.CompactionSync {	110✔
195	persistOptions.NoSync = false	17✔
196	}	17✔
197
198	err = s.persistFooter(fileCompact, compactFooter, persistOptions)	93✔
199	if err != nil {	93✔
200	if partialCompactStart == 0 {	×
201	s.removeFileOnClose(frefCompact)	×
202	}	×
203	return err	×
204	}
205
206	err = compactFooter.loadSegments(s.options, frefCompact)	93✔
207	if err != nil {	93✔
208	if partialCompactStart == 0 {	×
209	s.removeFileOnClose(frefCompact)	×
210	}	×
211	return err	×
212	}
213
214	s.m.Lock()	93✔
215	footerPrev := s.footer	93✔
216	s.footer = compactFooter // Owns the frefCompact ref-count.	93✔
217	if partialCompactStart == 0 {	186✔
218	s.totCompactions++	93✔
219	} else {	93✔
220	s.totCompactionsPartial++	×
221	}	×
222	s.m.Unlock()	93✔
223		93✔
224	s.histograms["CompactUsecs"].Add(	93✔
225	uint64(time.Since(startTime).Nanoseconds()/1000), 1)	93✔
226		93✔
227	if footerPrev != nil {	185✔
228	footerPrev.DecRef()	92✔
229	}	92✔
230
231	return nil	93✔
232	}
233
234	func (s Store) mergeSegStacks(footer Footer, splicePoint int,
235	higher segmentStack) segmentStack {	93✔
236	var footerSS *segmentStack	93✔
237	var lenFooterSS int	93✔
238	if footer != nil && footer.ss != nil {	186✔
239	footerSS = footer.ss	93✔
240	lenFooterSS = len(footerSS.a)	93✔
241	}	93✔
242	rv := &segmentStack{	93✔
243	options: higher.options,	93✔
244	a: make([]Segment, 0, len(higher.a)+lenFooterSS),	93✔
245	incarNum: higher.incarNum,	93✔
246	}	93✔
247	if footerSS != nil {	186✔
248	rv.a = append(rv.a, footerSS.a[splicePoint:]...)	93✔
249	}	93✔
250	rv.a = append(rv.a, higher.a...)	93✔
251	for cName, newStack := range higher.childSegStacks {	93✔
252	if len(rv.childSegStacks) == 0 {	×
253	rv.childSegStacks = make(map[string]*segmentStack)	×
254	}	×
255	if footer == nil {	×
256	rv.childSegStacks[cName] = s.mergeSegStacks(nil,	×
257	splicePoint, newStack)	×
258	continue	×
259	}
260
261	childFooter, exists := footer.ChildFooters[cName]	×
262	if exists {	×
263	if childFooter.incarNum != higher.incarNum {	×
264	// Fast child collection recreation, must not merge	×
265	// segments from prior incarnation.	×
266	childFooter = nil	×
267	}	×
268	}
269	rv.childSegStacks[cName] = s.mergeSegStacks(childFooter,	×
270	splicePoint, newStack)	×
271	}
272	return rv	93✔
273	}
274
275	func (right Footer) spliceFooter(left Footer, splicePoint int) {	×
276	slocs := make([]SegmentLoc, splicePoint, splicePoint+len(right.SegmentLocs))	×
277	copy(slocs, left.SegmentLocs[0:splicePoint])	×
278	slocs = append(slocs, right.SegmentLocs...)	×
279	right.SegmentLocs = slocs	×
280		×
281	for cName, childFooter := range right.ChildFooters {	×
282	storeChildFooter, exists := left.ChildFooters[cName]	×
283	if exists {	×
284	if storeChildFooter.incarNum != childFooter.incarNum {	×
285	// Fast child collection recreation, ok to drop store footer's	×
286	// segments from prior incarnation.	×
287	continue	×
288	}
289	childFooter.spliceFooter(storeChildFooter, splicePoint)	×
290	}
291	}
292	}
293
294	func (s Store) writeSegments(newSS segmentStack, frefCompact *FileRef,
295	fileCompact File) (compactFooter *Footer, err error) {	93✔
296	finfo, err := fileCompact.Stat()	93✔
297	if err != nil {	93✔
298	return nil, err	×
299	}	×
300	pos := finfo.Size()	93✔
301	stats := newSS.Stats()	93✔
302		93✔
303	kvsBegPos := pageAlignCeil(pos)	93✔
304	bufBegPos := pageAlignCeil(kvsBegPos + 1 + (int64(8+8) * int64(stats.CurOps)))	93✔
305		93✔
306	compactionBufferPages := 0	93✔
307	if s.options != nil {	186✔
308	compactionBufferPages = s.options.CompactionBufferPages	93✔
309	}	93✔
310	if compactionBufferPages <= 0 {	169✔
311	compactionBufferPages = DefaultStoreOptions.CompactionBufferPages	76✔
312	}	76✔
313	compactionBufferSize := StorePageSize * compactionBufferPages	93✔
314		93✔
315	compactWriter := &compactWriter{	93✔
316	kvsWriter: newBufferedSectionWriter(fileCompact, kvsBegPos, 0, compactionBufferSize),	93✔
317	bufWriter: newBufferedSectionWriter(fileCompact, bufBegPos, 0, compactionBufferSize),	93✔
318	}	93✔
319	onError := func(err error) error {	93✔
320	compactWriter.kvsWriter.Stop()	×
321	compactWriter.bufWriter.Stop()	×
322	return err	×
323	}	×
324
325	err = newSS.mergeInto(0, len(newSS.a), compactWriter, nil, false, false, s.abortCh)	93✔
326	if err != nil {	93✔
327	return nil, onError(err)	×
328	}	×
329
330	if err = compactWriter.kvsWriter.Flush(); err != nil {	93✔
331	return nil, onError(err)	×
332	}	×
333	if err = compactWriter.bufWriter.Flush(); err != nil {	93✔
334	return nil, onError(err)	×
335	}	×
336
337	if err = compactWriter.kvsWriter.Stop(); err != nil {	93✔
338	return nil, onError(err)	×
339	}	×
340	if err = compactWriter.bufWriter.Stop(); err != nil {	93✔
341	return nil, onError(err)	×
342	}	×
343
344	compactFooter = &Footer{	93✔
345	refs: 1,	93✔
346	SegmentLocs: []SegmentLoc{	93✔
347	{	93✔
348	Kind: SegmentKindBasic,	93✔
349	KvsOffset: uint64(kvsBegPos),	93✔
350	KvsBytes: uint64(compactWriter.kvsWriter.Offset() - kvsBegPos),	93✔
351	BufOffset: uint64(bufBegPos),	93✔
352	BufBytes: uint64(compactWriter.bufWriter.Offset() - bufBegPos),	93✔
353	TotOpsSet: compactWriter.totOperationSet,	93✔
354	TotOpsDel: compactWriter.totOperationDel,	93✔
355	TotKeyByte: compactWriter.totKeyByte,	93✔
356	TotValByte: compactWriter.totValByte,	93✔
357	},	93✔
358	},	93✔
359	}	93✔
360		93✔
361	for cName, childSegStack := range newSS.childSegStacks {	93✔
362	if compactFooter.ChildFooters == nil {	×
363	compactFooter.ChildFooters = make(map[string]*Footer)	×
364	}	×
365	childFooter, err := s.writeSegments(childSegStack,	×
366	frefCompact, fileCompact)	×
367	if err != nil {	×
368	return nil, err	×
369	}	×
370	compactFooter.ChildFooters[cName] = childFooter	×
371	}
372	return compactFooter, nil	93✔
373	}
374
375	type compactWriter struct {
376	file File
377	kvsWriter *bufferedSectionWriter
378	bufWriter *bufferedSectionWriter
379
380	totOperationSet uint64
381	totOperationDel uint64
382	totOperationMerge uint64
383	totKeyByte uint64
384	totValByte uint64
385	}
386
387	func (cw *compactWriter) Mutate(operation uint64, key, val []byte) error {	746,990✔
388	keyStart := cw.bufWriter.Written()	746,990✔
389		746,990✔
390	_, err := cw.bufWriter.Write(key)	746,990✔
391	if err != nil {	746,990✔
392	return err	×
393	}	×
394
395	_, err = cw.bufWriter.Write(val)	746,990✔
396	if err != nil {	746,990✔
397	return err	×
398	}	×
399
400	keyLen := len(key)	746,990✔
401	valLen := len(val)	746,990✔
402		746,990✔
403	opKlVl := encodeOpKeyLenValLen(operation, keyLen, valLen)	746,990✔
404		746,990✔
405	if keyLen <= 0 && valLen <= 0 {	746,990✔
406	keyStart = 0	×
407	}	×
408
409	pair := []uint64{opKlVl, uint64(keyStart)}	746,990✔
410	kvsBuf, err := Uint64SliceToByteSlice(pair)	746,990✔
411	if err != nil {	746,990✔
412	return err	×
413	}	×
414
415	_, err = cw.kvsWriter.Write(kvsBuf)	746,990✔
416	if err != nil {	746,990✔
417	return err	×
418	}	×
419
420	switch operation {	746,990✔
421	case OperationSet:	746,990✔
422	cw.totOperationSet++	746,990✔
423	case OperationDel:	×
424	cw.totOperationDel++	×
425	case OperationMerge:	×
426	cw.totOperationMerge++	×
427	default:	×
428	}
429
430	cw.totKeyByte += uint64(keyLen)	746,990✔
431	cw.totValByte += uint64(valLen)	746,990✔
432		746,990✔
433	return nil	746,990✔
434	}

couchbase / moss / 255

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