couchbase / moss / 305

pending completion

Build # 305

Build Type

push

travis-ci

Committed by

steveyen

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/<a class="double-link" href="https://git"><a class=hub.com/couchbase/moss/commit/<a class="double-link" href="https://git"><a class=hub.com/couchbase/moss/commit/929cc75f09a3bec7da9f993b041fbd7c9a0daa92">929cc75f0">&lt;a href=&quot;https://github.com/couchbase/moss/commit/</a><a class="double-link" href="https://github.com/couchbase/moss/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/couchbase/moss/commit/929cc75f09a3bec7da9f993b041fbd7c9a0daa92">929cc75f0</a><a href="https://github.com/couchbase/moss/commit/929cc75f09a3bec7da9f993b041fbd7c9a0daa92">&lt;a href=&quot;https://github.com/couchbase/moss/commit/929cc75f09a3bec7da9f993b041fbd7c9a0daa92&quot;&gt;&amp;quot;&amp;gt;MB-29664 - provide &amp;amp;#39;base&amp;amp;#39; seg-stack during partial compactions

If the application is configured to use partial compaction and was
also using &amp;amp;#39;merge operators&amp;amp;#39;, then there could be data loss as the
merge-operators were incorrectly not provided values from older or
lower &amp;amp;#39;base&amp;amp;#39; segments.

IMPORTANT / NOTE -- Even with this commit, there is a remaining issue
with child collections and merge-operators, in the face of a partial
compactions.  Until there is a fix for that, for now applications who
are using merge-operators in their child-collections should configure
moss to only perform full compactions instead of partial compactions.

Change-Id: I&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/couchbase/moss/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/couchbase/moss/commit/81060b01b&quot;&gt;81060b01b&lt;/a&gt;&quot;&gt;81060b01b&lt;/a&gt;&lt;a href=&quot;https://github.com/couchbase/moss/commit/929cc75f09a3bec</a>7da9f993b041fbd7c9a0daa92">
Reviewed-on: http://review.couchbase.org/94262
Well-Formed: Build Bot &lt;build@couchbase.com&gt;
Reviewed-by: Sundar Sridharan &lt;sundar@couchbase.com&gt;
Tested-by: Steve Yen &lt;steve.yen@gmail.com&gt;</a>

Run Details

23 of 23 new or added lines in 1 file covered. (100.0%)

3367 of 4052 relevant lines covered (83.09%)

7606112.46 hits per line

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

76.09

/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 {
                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 {
                                sizeBefore = finfo.Size() // Fetch old file 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 {
                                sizeAfter = finfo.Size() // Fetch new file 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.
//
// Example:                                    ||
//                                             ||
//      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   // Segments to compact (all segs when full compaction).
        var newBase *segmentStack // Segments not compacted (nil when full compaction).

        if higher != nil {
                ssHigher, ok := higher.(*segmentStack)
                if !ok {
                        return fmt.Errorf("store_compact: higher not a segmentStack")
                }

                if ssHigher.isEmpty() && len(footer.SegmentLocs) <= 1 {
                        // No incoming data to persist and 1 or fewer footer segments.
                        return ErrNothingToCompact
                }

                ssHigher.ensureFullySorted()

                newSS, newBase = 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 & 1 or fewer footer segments.
                        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()

        syncAfterBytes := 0
        if s.options != nil {
                syncAfterBytes = s.options.CompactionSyncAfterBytes
                if syncAfterBytes == 0 {
                        syncAfterBytes = DefaultStoreOptions.CompactionSyncAfterBytes
                }
        }

        compactFooter, err := s.writeSegments(newSS, newBase,
                frefCompact,
                fileCompact,
                partialCompactStart != 0, // Include deletions for partialCompactions.
                syncAfterBytes)
        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 || s.options.CompactionSyncAfterBytes > 0) {
                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) (rv, rvBase *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:]...)

                if splicePoint > 0 {
                        rvBase = &segmentStack{
                                options: footerSS.options,
                                a:       footerSS.a[0: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, rvBase
}

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, base *segmentStack,
        frefCompact *FileRef, fileCompact File,
        includeDeletes bool, syncAfterBytes int) (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{
                file:           fileCompact,
                kvsWriter:      newBufferedSectionWriter(fileCompact, kvsBegPos, 0, compactionBufferSize, s),
                bufWriter:      newBufferedSectionWriter(fileCompact, bufBegPos, 0, compactionBufferSize, s),
                syncAfterBytes: syncAfterBytes,
        }

        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, base,
                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)
                }

                // TODO: IMPORTANT: See MB-29664 - merge-operators, child
                // collections, and partial/leveled compaction does not work
                // correctly.  You need to use full compaction if you're using
                // merge-operators with child collections.  The fix will be to
                // compute and provide the right childSegStackBase to the
                // recursive writeSegments() calls.
                //
                var childSegStackBase *segmentStack

                childFooter, err := s.writeSegments(childSegStack, childSegStackBase,
                        frefCompact, fileCompact, includeDeletes, syncAfterBytes)
                if err != nil {
                        return nil, err
                }

                compactFooter.ChildFooters[cName] = childFooter
        }

        return compactFooter, nil
}

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

        // Bytes after which file's Sync() is to be invoked provided
        // sync is enabled. If <= 0, Sync() is not invoked.
        syncAfterBytes int

        // Bytes written since the last Sync().
        bytesSinceSync int

        totOperationSet   uint64
        totOperationDel   uint64
        totOperationMerge uint64
        totKeyByte        uint64
        totValByte        uint64
}

func (cw *compactWriter) Mutate(operation uint64, key, val []byte) error {
        if cw.syncAfterBytes > 0 && cw.bytesSinceSync > cw.syncAfterBytes {
                err := cw.file.Sync()
                if err != nil {
                        return err
                }
                cw.bytesSinceSync = 0
        }

        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)

        cw.bytesSinceSync += keyLen + valLen + 16 /* len(kvsBuf) */

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

couchbase / moss / 305

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