couchbase / moss / 270

pending completion

Build # 270

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/<a class="double-link" href="https://git"><a class=hub.com/couchbase/moss/commit/ee00ec710ebc2367801899cc6939e7066f2633cc">ee00ec710<a href="https://github.com/couchbase/moss/commit/ee00ec710ebc2367801899cc6939e7066f2633cc">&lt;a href=&quot;https://github.com/couchbase/moss/commit/ee00ec710ebc2367801899cc6939e7066f2633cc&quot;&gt;&quot;&gt;Revert &amp;quot;MB-25288: Revert &amp;quot;MB-24680: When idle try compactions to speed queries&amp;quot;&amp;quot;

This reverts commit &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/couchbase/moss/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/couchbase/moss/commit/dd021f4963e902002074f6db80e7b6f82e70e636">dd021f496</a>">dd021f496</a><a href="https://github.com/couchbase/moss/commit/ee00ec710ebc2367801899cc6939e7066f2633cc">.

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/c84d557fd4af5d57225daeb19a95f77dcc09a07e">c84d557fd</a>">c84d557fd</a><a href="https://github.com/couchbase/moss/commit/ee00ec710ebc2367801899cc6939e7066f2633cc">
Reviewed-on: http://review.couchbase.org/80976
Well-Formed: Build Bot <build@couchbase.com>
Reviewed-by: Sundararaman Sridharan <sundar@couchbase.com>
Tested-by: Sundararaman Sridharan <sundar@couchbase.com>

Run Details

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

3120 of 3749 relevant lines covered (83.22%)

1507764.22 hits per line

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

72.86

/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, _ := footer.segmentLocs()

        defer footer.DecRef()
        var partialCompactStart int

        if compactionConcern == CompactionAllow {
                // First compute size of the incoming batch of data.
                var incomingDataSize uint64
                if higher != nil {
                        higherSS, ok := higher.(*segmentStack)
                        if ok {
                                higherStats := higherSS.Stats()
                                if higherStats != nil {
                                        incomingDataSize = higherStats.CurBytes
                                }
                        }
                }

                // Try leveled compaction to same file.
                var doCompact bool
                partialCompactStart, doCompact = calcPartialCompactionStart(slocs,
                        incomingDataSize, s.options)
                if doCompact {
                        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 {
                if err == ErrNothingToCompact {
                        return false, nil // swallow the error internally.
                }
                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
}

// calcPartialCompactionStart - returns an index into the segment locations
// from which point it would be better to merge segments into a larger one.
// Return Values:
//  0 => Full compaction into a new file.
// >0 => Partially Compact all the segments starting at this return index,
//       and append  this one segment at the end of the file while retaining all
//       segments before this starting segment.
//  false => append data to the end of the file.
//  true  => perform compaction.
//                                         ||
//                                         ||
//  Say CompactionLevelMaxSegments = 2     ||
//                                         ||
//     ||                  ||              ||
//  || ||               || || ??           ||
//  || ||          ==>  || || ??     ==>   ||
//  || ||               || || ??           ||
//  || ||               || || ??           ||
//  || ||  || ||  ##    || || ??           ||
//  || ||  || ||  ##    || || ??           ||
// ----------------->  ----------->      -----> (new file)
//  level1 level0 new  level1 hits max!  final file
//
func calcPartialCompactionStart(slocs SegmentLocs, newDataSize uint64,
        options *StoreOptions) (compStartIdx int, doCompact bool) {
        maxSegmentsPerLevel := options.CompactionLevelMaxSegments
        if maxSegmentsPerLevel == 0 {
                maxSegmentsPerLevel = DefaultStoreOptions.CompactionLevelMaxSegments
        }
        fragmentationThreshold := options.CompactionPercentage
        if fragmentationThreshold == 0.0 {
                fragmentationThreshold = DefaultStoreOptions.CompactionPercentage
        }
        levelMultiplier := options.CompactionLevelMultiplier
        if levelMultiplier == 0 {
                levelMultiplier = DefaultStoreOptions.CompactionLevelMultiplier
        }
        if newDataSize == 0 { // Idle compaction => attempt full compaction.
                return 0, true
        }
        if len(slocs) < maxSegmentsPerLevel {
                return -1, false // No segments => append to end of same file.
        }

        determineExponent := func(segSize, curLevelSize uint64, curLevel int) int {
                newLevel := curLevel
                growBy := uint64(levelMultiplier)
                for sz := curLevelSize * growBy; sz <= segSize && sz > 0; sz *= growBy {
                        newLevel++
                }
                return newLevel
        }

        compStartIdx = -1 // Assume we need to append to end of file.
        // sizeSoFar represents the estimated size of the partially compacted
        // future segment if we were to start compacting from current segment.
        sizeSoFar := newDataSize
        curLevelSize := newDataSize
        curLevel := 0
        numInLevel := 1 // Incoming batch is assumed to be appended in L0.

        for idx := len(slocs) - 1; idx >= 0; idx-- {
                segSize := slocs[idx].TotKeyByte + slocs[idx].TotValByte
                newLevel := determineExponent(segSize, curLevelSize, curLevel)
                if newLevel > curLevel {
                        break
                }
                numInLevel++
                sizeSoFar += segSize
                if numInLevel > maxSegmentsPerLevel {
                        compStartIdx = idx
                        curLevel = determineExponent(sizeSoFar, curLevelSize, curLevel)
                        numInLevel = 1
                        curLevelSize = sizeSoFar
                }
        }

        if compStartIdx > 0 && fragmentationThreshold > 0 {
                totDataSize := uint64(0)
                for idx := 0; idx < len(slocs); idx++ {
                        totDataSize += slocs[idx].TotKeyByte + slocs[idx].TotValByte
                }
                finfo, err := slocs[0].mref.fref.file.Stat()
                if err != nil {
                        return 0, true
                }
                predictedDataSize := int64(totDataSize) + int64(newDataSize)
                predictedFileSize := finfo.Size() + int64(curLevelSize)

                staleDataSize := predictedFileSize - predictedDataSize
                fragPercentage := float64(staleDataSize) / float64(predictedFileSize)
                if fragPercentage > fragmentationThreshold {
                        return 0, true // File is too fragmented for partial compaction.
                }
        }
        if compStartIdx >= 0 {
                doCompact = true
        }

        return compStartIdx, doCompact
}

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")
                }
                if ssHigher.isEmpty() && len(footer.SegmentLocs) == 1 {
                        // No incoming data to persist and just 1 segment in footer.
                        return ErrNothingToCompact // Nothing to compact.
                }
                ssHigher.ensureFullySorted()
                newSS = s.mergeSegStacks(footer, partialCompactStart, ssHigher)
        } else {
                newSS = footer.ss      // Safe as footer ref count is held positive.
                if len(newSS.a) == 1 { // No incoming data and just 1 segment in footer.
                        return ErrNothingToCompact // no need to perform compaction.
                }
        }

        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,
                partialCompactStart != 0) // include deletions for partialCompactions.
        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, includeDeletes bool) (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, s),
                bufWriter: newBufferedSectionWriter(fileCompact, bufBegPos, 0, compactionBufferSize, s),
        }
        onError := func(err error) error {
                compactWriter.kvsWriter.Stop()
                compactWriter.bufWriter.Stop()
                return err
        }
        s.m.Lock()
        s.totCompactionBeforeBytes += stats.CurBytes
        s.m.Unlock()

        err = newSS.mergeInto(0, len(newSS.a), compactWriter, nil, includeDeletes, 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, includeDeletes)
                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) {	325✔
22	if s.Options().CollectionOptions.ReadOnly {	326✔
23	// Do not compact in Read-Only mode	1✔
24	return false, nil	1✔
25	}	1✔
26
27	compactionConcern := persistOptions.CompactionConcern	324✔
28	if compactionConcern <= 0 {	372✔
29	return false, nil	48✔
30	}	48✔
31
32	footer, err := s.snapshot()	276✔
33	if err != nil {	276✔
34	return false, err	×
35	}	×
36
37	defer footer.DecRef()	276✔
38		276✔
39	slocs, _ := footer.segmentLocs()	276✔
40		276✔
41	defer footer.DecRef()	276✔
42	var partialCompactStart int	276✔
43		276✔
44	if compactionConcern == CompactionAllow {	478✔
45	// First compute size of the incoming batch of data.	202✔
46	var incomingDataSize uint64	202✔
47	if higher != nil {	404✔
48	higherSS, ok := higher.(*segmentStack)	202✔
49	if ok {	404✔
50	higherStats := higherSS.Stats()	202✔
51	if higherStats != nil {	404✔
52	incomingDataSize = higherStats.CurBytes	202✔
53	}	202✔
54	}
55	}
56
57	// Try leveled compaction to same file.
58	var doCompact bool	202✔
59	partialCompactStart, doCompact = calcPartialCompactionStart(slocs,	202✔
60	incomingDataSize, s.options)	202✔
61	if doCompact {	242✔
62	compactionConcern = CompactionForce	40✔
63	} // else append data to the end of the same file.	40✔
64	}
65
66	if compactionConcern != CompactionForce {	438✔
67	return false, nil	162✔
68	}	162✔
69
70	err = s.compact(footer, partialCompactStart, higher, persistOptions)	114✔
71	if err != nil {	114✔
72	if err == ErrNothingToCompact {	×
73	return false, nil // swallow the error internally.	×
74	}	×
75	return false, err	×
76	}
77
78	var sizeBefore, sizeAfter int64	114✔
79		114✔
80	if len(slocs) > 0 {	222✔
81	mref := slocs[0].mref	108✔
82	if mref != nil && mref.fref != nil {	216✔
83	var finfo os.FileInfo	108✔
84	if partialCompactStart == 0 {	202✔
85	finfo, err = s.removeFileOnClose(mref.fref)	94✔
86	} else {	108✔
87	finfo, err = mref.fref.file.Stat()	14✔
88	}	14✔
89	if err == nil && len(finfo.Name()) > 0 {	216✔
90	// Fetch size of old file	108✔
91	sizeBefore = finfo.Size()	108✔
92	}	108✔
93	}
94	}
95
96	slocs, _ = footer.segmentLocs()	114✔
97	defer footer.DecRef()	114✔
98		114✔
99	if len(slocs) > 0 {	222✔
100	mref := slocs[0].mref	108✔
101	if mref != nil && mref.fref != nil {	216✔
102	finfo, err := mref.fref.file.Stat()	108✔
103	if err == nil && len(finfo.Name()) > 0 {	216✔
104	// Fetch size of new file	108✔
105	sizeAfter = finfo.Size()	108✔
106	}	108✔
107	}
108	}
109
110	s.m.Lock()	114✔
111	s.numLastCompactionBeforeBytes = uint64(sizeBefore)	114✔
112	s.numLastCompactionAfterBytes = uint64(sizeAfter)	114✔
113	delta := sizeBefore - sizeAfter	114✔
114	if delta > 0 {	114✔
115	s.totCompactionDecreaseBytes += uint64(delta)	×
116	if s.maxCompactionDecreaseBytes < uint64(delta) {	×
117	s.maxCompactionDecreaseBytes = uint64(delta)	×
118	}	×
119	} else if delta < 0 {	114✔
120	delta = -delta	×
121	s.totCompactionIncreaseBytes += uint64(delta)	×
122	if s.maxCompactionIncreaseBytes < uint64(delta) {	×
123	s.maxCompactionIncreaseBytes = uint64(delta)	×
124	}	×
125	}
126	s.m.Unlock()	114✔
127		114✔
128	return true, nil	114✔
129	}
130
131	// calcPartialCompactionStart - returns an index into the segment locations
132	// from which point it would be better to merge segments into a larger one.
133	// Return Values:
134	// 0 => Full compaction into a new file.
135	// >0 => Partially Compact all the segments starting at this return index,
136	// and append this one segment at the end of the file while retaining all
137	// segments before this starting segment.
138	// false => append data to the end of the file.
139	// true => perform compaction.
140	// \|\|
141	// \|\|
142	// Say CompactionLevelMaxSegments = 2 \|\|
143	// \|\|
144	// \|\| \|\| \|\|
145	// \|\| \|\| \|\| \|\| ?? \|\|
146	// \|\| \|\| ==> \|\| \|\| ?? ==> \|\|
147	// \|\| \|\| \|\| \|\| ?? \|\|
148	// \|\| \|\| \|\| \|\| ?? \|\|
149	// \|\| \|\| \|\| \|\| ## \|\| \|\| ?? \|\|
150	// \|\| \|\| \|\| \|\| ## \|\| \|\| ?? \|\|
151	// -----------------> -----------> -----> (new file)
152	// level1 level0 new level1 hits max! final file
153	//
154	func calcPartialCompactionStart(slocs SegmentLocs, newDataSize uint64,
155	options *StoreOptions) (compStartIdx int, doCompact bool) {	202✔
156	maxSegmentsPerLevel := options.CompactionLevelMaxSegments	202✔
157	if maxSegmentsPerLevel == 0 {	202✔
158	maxSegmentsPerLevel = DefaultStoreOptions.CompactionLevelMaxSegments	×
159	}	×
160	fragmentationThreshold := options.CompactionPercentage	202✔
161	if fragmentationThreshold == 0.0 {	221✔
162	fragmentationThreshold = DefaultStoreOptions.CompactionPercentage	19✔
163	}	19✔
164	levelMultiplier := options.CompactionLevelMultiplier	202✔
165	if levelMultiplier == 0 {	202✔
166	levelMultiplier = DefaultStoreOptions.CompactionLevelMultiplier	×
167	}	×
168	if newDataSize == 0 { // Idle compaction => attempt full compaction.	203✔
169	return 0, true	1✔
170	}	1✔
171	if len(slocs) < maxSegmentsPerLevel {	281✔
172	return -1, false // No segments => append to end of same file.	80✔
173	}	80✔
174
175	determineExponent := func(segSize, curLevelSize uint64, curLevel int) int {	903✔
176	newLevel := curLevel	782✔
177	growBy := uint64(levelMultiplier)	782✔
178	for sz := curLevelSize * growBy; sz <= segSize && sz > 0; sz *= growBy {	960✔
179	newLevel++	178✔
180	}	178✔
181	return newLevel	782✔
182	}
183
184	compStartIdx = -1 // Assume we need to append to end of file.	121✔
185	// sizeSoFar represents the estimated size of the partially compacted	121✔
186	// future segment if we were to start compacting from current segment.	121✔
187	sizeSoFar := newDataSize	121✔
188	curLevelSize := newDataSize	121✔
189	curLevel := 0	121✔
190	numInLevel := 1 // Incoming batch is assumed to be appended in L0.	121✔
191		121✔
192	for idx := len(slocs) - 1; idx >= 0; idx-- {	862✔
193	segSize := slocs[idx].TotKeyByte + slocs[idx].TotValByte	741✔
194	newLevel := determineExponent(segSize, curLevelSize, curLevel)	741✔
195	if newLevel > curLevel {	830✔
196	break	89✔
197	}
198	numInLevel++	652✔
199	sizeSoFar += segSize	652✔
200	if numInLevel > maxSegmentsPerLevel {	693✔
201	compStartIdx = idx	41✔
202	curLevel = determineExponent(sizeSoFar, curLevelSize, curLevel)	41✔
203	numInLevel = 1	41✔
204	curLevelSize = sizeSoFar	41✔
205	}	41✔
206	}
207
208	if compStartIdx > 0 && fragmentationThreshold > 0 {	135✔
209	totDataSize := uint64(0)	14✔
210	for idx := 0; idx < len(slocs); idx++ {	205✔
211	totDataSize += slocs[idx].TotKeyByte + slocs[idx].TotValByte	191✔
212	}	191✔
213	finfo, err := slocs[0].mref.fref.file.Stat()	14✔
214	if err != nil {	14✔
215	return 0, true	×
216	}	×
217	predictedDataSize := int64(totDataSize) + int64(newDataSize)	14✔
218	predictedFileSize := finfo.Size() + int64(curLevelSize)	14✔
219		14✔
220	staleDataSize := predictedFileSize - predictedDataSize	14✔
221	fragPercentage := float64(staleDataSize) / float64(predictedFileSize)	14✔
222	if fragPercentage > fragmentationThreshold {	14✔
223	return 0, true // File is too fragmented for partial compaction.	×
224	}	×
225	}
226	if compStartIdx >= 0 {	160✔
227	doCompact = true	39✔
228	}	39✔
229
230	return compStartIdx, doCompact	121✔
231	}
232
233	func (s Store) compact(footer Footer, partialCompactStart int, higher Snapshot,
234	persistOptions StorePersistOptions) error {	114✔
235	startTime := time.Now()	114✔
236		114✔
237	var newSS *segmentStack	114✔
238	if higher != nil {	228✔
239	ssHigher, ok := higher.(*segmentStack)	114✔
240	if !ok {	114✔
241	return fmt.Errorf("store: can only compact higher that's a segmentStack")	×
242	}	×
243	if ssHigher.isEmpty() && len(footer.SegmentLocs) == 1 {	114✔
244	// No incoming data to persist and just 1 segment in footer.	×
245	return ErrNothingToCompact // Nothing to compact.	×
246	}	×
247	ssHigher.ensureFullySorted()	114✔
248	newSS = s.mergeSegStacks(footer, partialCompactStart, ssHigher)	114✔
249	} else {	×
250	newSS = footer.ss // Safe as footer ref count is held positive.	×
251	if len(newSS.a) == 1 { // No incoming data and just 1 segment in footer.	×
252	return ErrNothingToCompact // no need to perform compaction.	×
253	}	×
254	}
255
256	var frefCompact *FileRef	114✔
257	var fileCompact File	114✔
258	var err error	114✔
259	if partialCompactStart == 0 {	214✔
260	s.m.Lock()	100✔
261	frefCompact, fileCompact, err = s.startFileLOCKED()	100✔
262	s.m.Unlock()	100✔
263	} else {	114✔
264	frefCompact, fileCompact, err = s.startOrReuseFile()	14✔
265	}	14✔
266	if err != nil {	114✔
267	return err	×
268	}	×
269	defer frefCompact.DecRef()	114✔
270		114✔
271	compactFooter, err := s.writeSegments(newSS, frefCompact, fileCompact,	114✔
272	partialCompactStart != 0) // include deletions for partialCompactions.	114✔
273	if err != nil {	114✔
274	if partialCompactStart == 0 {	×
275	s.removeFileOnClose(frefCompact)	×
276	}	×
277	return err	×
278	}
279	// Prefix restore the footer's partialCompactStart
280	if partialCompactStart != 0 {	128✔
281	compactFooter.spliceFooter(footer, partialCompactStart)	14✔
282	}	14✔
283
284	if s.options != nil && s.options.CompactionSync {	153✔
285	persistOptions.NoSync = false	39✔
286	}	39✔
287
288	err = s.persistFooter(fileCompact, compactFooter, persistOptions)	114✔
289	if err != nil {	114✔
290	if partialCompactStart == 0 {	×
291	s.removeFileOnClose(frefCompact)	×
292	}	×
293	return err	×
294	}
295
296	err = compactFooter.loadSegments(s.options, frefCompact)	114✔
297	if err != nil {	114✔
298	if partialCompactStart == 0 {	×
299	s.removeFileOnClose(frefCompact)	×
300	}	×
301	return err	×
302	}
303
304	s.m.Lock()	114✔
305	footerPrev := s.footer	114✔
306	s.footer = compactFooter // Owns the frefCompact ref-count.	114✔
307	if partialCompactStart == 0 {	214✔
308	s.totCompactions++	100✔
309	} else {	114✔
310	s.totCompactionsPartial++	14✔
311	}	14✔
312	s.m.Unlock()	114✔
313		114✔
314	s.histograms["CompactUsecs"].Add(	114✔
315	uint64(time.Since(startTime).Nanoseconds()/1000), 1)	114✔
316		114✔
317	if footerPrev != nil {	227✔
318	footerPrev.DecRef()	113✔
319	}	113✔
320
321	return nil	114✔
322	}
323
324	func (s Store) mergeSegStacks(footer Footer, splicePoint int,
325	higher segmentStack) segmentStack {	114✔
326	var footerSS *segmentStack	114✔
327	var lenFooterSS int	114✔
328	if footer != nil && footer.ss != nil {	228✔
329	footerSS = footer.ss	114✔
330	lenFooterSS = len(footerSS.a)	114✔
331	}	114✔
332	rv := &segmentStack{	114✔
333	options: higher.options,	114✔
334	a: make([]Segment, 0, len(higher.a)+lenFooterSS),	114✔
335	incarNum: higher.incarNum,	114✔
336	}	114✔
337	if footerSS != nil {	228✔
338	rv.a = append(rv.a, footerSS.a[splicePoint:]...)	114✔
339	}	114✔
340	rv.a = append(rv.a, higher.a...)	114✔
341	for cName, newStack := range higher.childSegStacks {	114✔
342	if len(rv.childSegStacks) == 0 {	×
343	rv.childSegStacks = make(map[string]*segmentStack)	×
344	}	×
345	if footer == nil {	×
346	rv.childSegStacks[cName] = s.mergeSegStacks(nil,	×
347	splicePoint, newStack)	×
348	continue	×
349	}
350
351	childFooter, exists := footer.ChildFooters[cName]	×
352	if exists {	×
353	if childFooter.incarNum != higher.incarNum {	×
354	// Fast child collection recreation, must not merge	×
355	// segments from prior incarnation.	×
356	childFooter = nil	×
357	}	×
358	}
359	rv.childSegStacks[cName] = s.mergeSegStacks(childFooter,	×
360	splicePoint, newStack)	×
361	}
362	return rv	114✔
363	}
364
365	func (right Footer) spliceFooter(left Footer, splicePoint int) {	14✔
366	slocs := make([]SegmentLoc, splicePoint, splicePoint+len(right.SegmentLocs))	14✔
367	copy(slocs, left.SegmentLocs[0:splicePoint])	14✔
368	slocs = append(slocs, right.SegmentLocs...)	14✔
369	right.SegmentLocs = slocs	14✔
370		14✔
371	for cName, childFooter := range right.ChildFooters {	14✔
372	storeChildFooter, exists := left.ChildFooters[cName]	×
373	if exists {	×
374	if storeChildFooter.incarNum != childFooter.incarNum {	×
375	// Fast child collection recreation, ok to drop store footer's	×
376	// segments from prior incarnation.	×
377	continue	×
378	}
379	childFooter.spliceFooter(storeChildFooter, splicePoint)	×
380	}
381	}
382	}
383
384	func (s Store) writeSegments(newSS segmentStack, frefCompact *FileRef,
385	fileCompact File, includeDeletes bool) (compactFooter *Footer, err error) {	114✔
386	finfo, err := fileCompact.Stat()	114✔
387	if err != nil {	114✔
388	return nil, err	×
389	}	×
390	pos := finfo.Size()	114✔
391	stats := newSS.Stats()	114✔
392		114✔
393	kvsBegPos := pageAlignCeil(pos)	114✔
394	bufBegPos := pageAlignCeil(kvsBegPos + 1 + (int64(8+8) * int64(stats.CurOps)))	114✔
395		114✔
396	compactionBufferPages := 0	114✔
397	if s.options != nil {	228✔
398	compactionBufferPages = s.options.CompactionBufferPages	114✔
399	}	114✔
400	if compactionBufferPages <= 0 {	189✔
401	compactionBufferPages = DefaultStoreOptions.CompactionBufferPages	75✔
402	}	75✔
403	compactionBufferSize := StorePageSize * compactionBufferPages	114✔
404		114✔
405	compactWriter := &compactWriter{	114✔
406	kvsWriter: newBufferedSectionWriter(fileCompact, kvsBegPos, 0, compactionBufferSize, s),	114✔
407	bufWriter: newBufferedSectionWriter(fileCompact, bufBegPos, 0, compactionBufferSize, s),	114✔
408	}	114✔
409	onError := func(err error) error {	114✔
410	compactWriter.kvsWriter.Stop()	×
411	compactWriter.bufWriter.Stop()	×
412	return err	×
413	}	×
414	s.m.Lock()	114✔
415	s.totCompactionBeforeBytes += stats.CurBytes	114✔
416	s.m.Unlock()	114✔
417		114✔
418	err = newSS.mergeInto(0, len(newSS.a), compactWriter, nil, includeDeletes, false, s.abortCh)	114✔
419	if err != nil {	114✔
420	return nil, onError(err)	×
421	}	×
422
423	if err = compactWriter.kvsWriter.Flush(); err != nil {	114✔
424	return nil, onError(err)	×
425	}	×
426	if err = compactWriter.bufWriter.Flush(); err != nil {	114✔
427	return nil, onError(err)	×
428	}	×
429
430	if err = compactWriter.kvsWriter.Stop(); err != nil {	114✔
431	return nil, onError(err)	×
432	}	×
433	if err = compactWriter.bufWriter.Stop(); err != nil {	114✔
434	return nil, onError(err)	×
435	}	×
436
437	compactFooter = &Footer{	114✔
438	refs: 1,	114✔
439	SegmentLocs: []SegmentLoc{	114✔
440	{	114✔
441	Kind: SegmentKindBasic,	114✔
442	KvsOffset: uint64(kvsBegPos),	114✔
443	KvsBytes: uint64(compactWriter.kvsWriter.Offset() - kvsBegPos),	114✔
444	BufOffset: uint64(bufBegPos),	114✔
445	BufBytes: uint64(compactWriter.bufWriter.Offset() - bufBegPos),	114✔
446	TotOpsSet: compactWriter.totOperationSet,	114✔
447	TotOpsDel: compactWriter.totOperationDel,	114✔
448	TotKeyByte: compactWriter.totKeyByte,	114✔
449	TotValByte: compactWriter.totValByte,	114✔
450	},	114✔
451	},	114✔
452	}	114✔
453		114✔
454	for cName, childSegStack := range newSS.childSegStacks {	114✔
455	if compactFooter.ChildFooters == nil {	×
456	compactFooter.ChildFooters = make(map[string]*Footer)	×
457	}	×
458	childFooter, err := s.writeSegments(childSegStack,	×
459	frefCompact, fileCompact, includeDeletes)	×
460	if err != nil {	×
461	return nil, err	×
462	}	×
463	compactFooter.ChildFooters[cName] = childFooter	×
464	}
465	return compactFooter, nil	114✔
466	}
467
468	type compactWriter struct {
469	file File
470	kvsWriter *bufferedSectionWriter
471	bufWriter *bufferedSectionWriter
472
473	totOperationSet uint64
474	totOperationDel uint64
475	totOperationMerge uint64
476	totKeyByte uint64
477	totValByte uint64
478	}
479
480	func (cw *compactWriter) Mutate(operation uint64, key, val []byte) error {	4,945,346✔
481	keyStart := cw.bufWriter.Written()	4,945,346✔
482	_, err := cw.bufWriter.Write(key)	4,945,346✔
483	if err != nil {	4,945,346✔
484	return err	×
485	}	×
486
487	_, err = cw.bufWriter.Write(val)	4,945,346✔
488	if err != nil {	4,945,346✔
489	return err	×
490	}	×
491
492	keyLen := len(key)	4,945,346✔
493	valLen := len(val)	4,945,346✔
494		4,945,346✔
495	opKlVl := encodeOpKeyLenValLen(operation, keyLen, valLen)	4,945,346✔
496		4,945,346✔
497	if keyLen <= 0 && valLen <= 0 {	4,945,346✔
498	keyStart = 0	×
499	}	×
500
501	pair := []uint64{opKlVl, uint64(keyStart)}	4,945,346✔
502	kvsBuf, err := Uint64SliceToByteSlice(pair)	4,945,346✔
503	if err != nil {	4,945,346✔
504	return err	×
505	}	×
506
507	_, err = cw.kvsWriter.Write(kvsBuf)	4,945,346✔
508	if err != nil {	4,945,346✔
509	return err	×
510	}	×
511
512	switch operation {	4,945,346✔
513	case OperationSet:	4,945,343✔
514	cw.totOperationSet++	4,945,343✔
515	case OperationDel:	3✔
516	cw.totOperationDel++	3✔
517	case OperationMerge:	×
518	cw.totOperationMerge++	×
519	default:	×
520	}
521
522	cw.totKeyByte += uint64(keyLen)	4,945,346✔
523	cw.totValByte += uint64(valLen)	4,945,346✔
524		4,945,346✔
525	return nil	4,945,346✔
526	}

couchbase / moss / 270

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