couchbase / moss / 298

pending completion

Build # 298

Build Type

push

travis-ci

Committed by

abhinavdangeti

Commit Message

<a href="https://github.com/couchbase/moss/commit/<a class=hub.com/couchbase/moss/commit/010c98934dc0b4e69f2d27c5abdfd36bbf6de3ee">010c98934<a href="https://github.com/couchbase/moss/commit/010c98934dc0b4e69f2d27c5abdfd36bbf6de3ee">&quot;&gt;MB-25705: Set defaults for segment keys index

+ As performace showed a substantial improvement in
  query throughput with the segment keys index, setting
  the following default store options:
    - SegmentKeysIndexMaxBytes:    100000
    - SegmentKeysIndexMinKeyBytes: 10000000
+ Note that advanced users can fine tune or even disable
  this feature by setting these parameters to a value less
  than 0.

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/67fa652c8ac748cba2fd18a217c8ca4e2b90b632">67fa652c8</a>">67fa652c8</a><a href="https://github.com/couchbase/moss/commit/010c98934dc0b4e69f2d27c5abdfd36bbf6de3ee">
Reviewed-on: http://review.couchbase.org/84944
Well-Formed: Build Bot &lt;build@couchbase.com&gt;
Reviewed-by: Steve Yen &lt;steve.yen@gmail.com>
Tested-by: abhinav dangeti <abhinav@couchbase.com>

Run Details

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

3354 of 4012 relevant lines covered (83.6%)

6363764.23 hits per line

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

76.06

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

couchbase / moss / 298

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