couchbase / moss / 199

pending completion

Build # 199

Build Type

push

travis-ci

Committed by

abhinavdangeti

Commit Message

MB-20256: Add keylen, vallen, ExecuteBatch related histograms

+ Adding the following histograms to the collection interface:
    - MutationKeyBytes
    - MutationValBytes
    - ExecuteBatchBytes
    - ExecuteBatchOpsCount

+ This change introduces a separate non-blocking go routine
  that will be responsible for populating the histograms
  with key-val stats.

                                                                            Before (ns/op)  After (ns/op)   Delta
BenchmarkStore_numItems1M_keySize20_valSize100_batchSize100-8               1509367857      1572045343      +4.15%
BenchmarkStore_numItems1M_keySize20_valSize100_batchSize100_randomLoad-8    1964633255      1814830967      -7.62%
BenchmarkStore_numItems1M_keySize20_valSize1000_batchSize100_randomLoad-8   7130239871      7166151755      +0.50%

Change-Id: I4f619bd74
Reviewed-on: http://review.couchbase.org/73986
Reviewed-by: Steve Yen <steve.yen@gmail.com>
Tested-by: abhinav dangeti <abhinav@couchbase.com>

Run Details

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

3190 of 4031 relevant lines covered (79.14%)

10424.83 hits per line

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

88.13

/segment.go

//  Copyright (c) 2016 Couchbase, Inc.
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the
//  License. You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//  Unless required by applicable law or agreed to in writing,
//  software distributed under the License is distributed on an "AS
//  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//  express or implied. See the License for the specific language
//  governing permissions and limitations under the License.

package moss

import (
        "bytes"
        "fmt"
        "sort"
)

// BASIC_SEGMENT_KIND is the code for a basic, persistable segment
// implementation, which represents a segment as two arrays: an array
// of contiguous key-val bytes [key0, val0, key1, val1, ... keyN,
// valN], and an array of offsets plus lengths into the first array.
var BASIC_SEGMENT_KIND = "a"

func init() {
        SegmentLoaders[BASIC_SEGMENT_KIND] = loadBasicSegment
}

// A Segment represents the read-oriented interface for a segment.
type Segment interface {
        // Returns the kind of segment, used for persistence.
        Kind() string

        // Len returns the number of ops in the segment.
        Len() int

        // NumKeyValBytes returns the number of bytes used for key-val data.
        NumKeyValBytes() (uint64, uint64)

        // FindStartKeyInclusivePos() returns the logical entry position for
        // the given (inclusive) start key.  With segment keys of [b, d, f],
        // looking for 'c' will return 1.  Looking for 'd' will return 1.
        // Looking for 'g' will return 3.  Looking for 'a' will return 0.
        FindStartKeyInclusivePos(startKeyInclusive []byte) int

        // GetOperationKeyVal() returns the operation, key, val for a given
        // logical entry position in the segment.
        GetOperationKeyVal(pos int) (operation uint64, key []byte, val []byte)

        // Returns true if the segment is already sorted, and returns
        // false if the sorting is only asynchronously scheduled.
        RequestSort(synchronous bool) bool
}

// A SegmentMutator represents the mutation methods of a segment.
type SegmentMutator interface {
        Mutate(operation uint64, key, val []byte) error
}

// A SegmentPersister represents a segment that can be persisted.
type SegmentPersister interface {
        Persist(file File) (SegmentLoc, error)
}

// A segment is a basic implementation of the segment related
// interfaces and represents a sequence of key-val entries or
// operations.  A segment's kvs will be sorted by key when the segment
// is pushed into the collection.  A segment implements the Batch
// interface.
type segment struct {
        // Each key-val operation is encoded as 2 uint64's...
        // - operation (see: maskOperation) |
        //       key length (see: maskKeyLength) |
        //       val length (see: maskValLength).
        // - start index into buf for key-val bytes.
        kvs []uint64

        // Contiguous backing memory for the keys and vals of the segment.
        buf []byte

        // If this segment needs sorting, then needSorterCh will be
        // non-nil and also the first goroutine that reads successfully
        // from needSorterCh becomes the sorter of this segment.  All
        // other goroutines must instead wait on the waitSortedCh.
        needSorterCh chan bool

        // Once the sorter of this segment is done sorting the kvs, it
        // close()'s the waitSortedCh, treating waitSortedCh like a
        // one-way latch.  The needSorterCh and waitSortedCh will either
        // be nil or non-nil together.  A segment that was "born
        // sorted" will have needSorterCh and waitSortedCh as both nil.
        waitSortedCh chan struct{}

        totOperationSet   uint64
        totOperationDel   uint64
        totOperationMerge uint64
        totKeyByte        uint64
        totValByte        uint64
}

// See the OperationXxx consts.
const maskOperation = uint64(0x0F00000000000000)

// Max key length is 2^24, from 24 bits key length.
const maskKeyLength = uint64(0x00FFFFFF00000000)

const maxKeyLength = 1<<24 - 1

// Max val length is 2^28, from 28 bits val length.
const maskValLength = uint64(0x000000000FFFFFFF)

const maxValLength = 1<<28 - 1

const maskRESERVED = uint64(0xF0000000F0000000)

// newSegment() allocates a segment with hinted amount of resources.
func newSegment(totalOps, totalKeyValBytes int) (*segment, error) {
        return &segment{
                kvs: make([]uint64, 0, totalOps*2),
                buf: make([]byte, 0, totalKeyValBytes),
        }, nil
}

func (a *segment) Kind() string { return BASIC_SEGMENT_KIND }

// Close releases resources associated with the segment.
func (a *segment) Close() error {
        return nil
}

// Set copies the key and val bytes into the segment as a "set"
// mutation.  The key must be unique (not repeated) within the
// segment.
func (a *segment) Set(key, val []byte) error {
        return a.mutate(OperationSet, key, val)
}

// Del copies the key bytes into the segment as a "deletion" mutation.
// The key must be unique (not repeated) within the segment.
func (a *segment) Del(key []byte) error {
        return a.mutate(OperationDel, key, nil)
}

// Merge creates or updates a key-val entry in the Collection via the
// MergeOperator defined in the CollectionOptions.  The key must be
// unique (not repeated) within the segment.
func (a *segment) Merge(key, val []byte) error {
        return a.mutate(OperationMerge, key, val)
}

// ------------------------------------------------------

// Alloc provides a slice of bytes "owned" by the segment, to reduce
// extra copying of memory.  See the Collection.NewBatch() method.
func (a *segment) Alloc(numBytes int) ([]byte, error) {
        bufLen := len(a.buf)
        bufCap := cap(a.buf)

        if numBytes > bufCap-bufLen {
                return nil, ErrAllocTooLarge
        }

        rv := a.buf[bufLen : bufLen+numBytes]

        a.buf = a.buf[0 : bufLen+numBytes]

        return rv, nil
}

// AllocSet is like Set(), but the caller must provide []byte
// parameters that came from Alloc(), for less buffer copying.
func (a *segment) AllocSet(keyFromAlloc, valFromAlloc []byte) error {
        bufCap := cap(a.buf)

        keyStart := bufCap - cap(keyFromAlloc)

        return a.mutateEx(OperationSet,
                keyStart, len(keyFromAlloc), len(valFromAlloc))
}

// AllocDel is like Del(), but the caller must provide []byte
// parameters that came from Alloc(), for less buffer copying.
func (a *segment) AllocDel(keyFromAlloc []byte) error {
        bufCap := cap(a.buf)

        keyStart := bufCap - cap(keyFromAlloc)

        return a.mutateEx(OperationDel,
                keyStart, len(keyFromAlloc), 0)
}

// AllocMerge is like Merge(), but the caller must provide []byte
// parameters that came from Alloc(), for less buffer copying.
func (a *segment) AllocMerge(keyFromAlloc, valFromAlloc []byte) error {
        bufCap := cap(a.buf)

        keyStart := bufCap - cap(keyFromAlloc)

        return a.mutateEx(OperationMerge,
                keyStart, len(keyFromAlloc), len(valFromAlloc))
}

// ------------------------------------------------------

func (a *segment) Mutate(operation uint64, key, val []byte) error {
        return a.mutate(operation, key, val)
}

func (a *segment) mutate(operation uint64, key, val []byte) error {
        keyStart := len(a.buf)
        a.buf = append(a.buf, key...)
        keyLength := len(a.buf) - keyStart

        valStart := len(a.buf)
        a.buf = append(a.buf, val...)
        valLength := len(a.buf) - valStart

        return a.mutateEx(operation, keyStart, keyLength, valLength)
}

func (a *segment) mutateEx(operation uint64,
        keyStart, keyLength, valLength int) error {
        if keyLength > maxKeyLength {
                return ErrKeyTooLarge
        }
        if valLength > maxValLength {
                return ErrValueTooLarge
        }

        if keyLength <= 0 && valLength <= 0 {
                keyStart = 0
        }

        opKlVl := encodeOpKeyLenValLen(operation, keyLength, valLength)

        a.kvs = append(a.kvs, opKlVl, uint64(keyStart))

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

        a.totKeyByte += uint64(keyLength)
        a.totValByte += uint64(valLength)

        return nil
}

// ------------------------------------------------------

// NumKeyValBytes returns the number of bytes used for key-val data.
func (a *segment) NumKeyValBytes() (uint64, uint64) {
        return a.totKeyByte, a.totValByte
}

// ------------------------------------------------------

// Len returns the number of ops in the segment.
func (a *segment) Len() int {
        return len(a.kvs) / 2
}

func (a *segment) Swap(i, j int) {
        x := i * 2
        y := j * 2

        // Operation + key length + val length.
        a.kvs[x], a.kvs[y] = a.kvs[y], a.kvs[x]

        x++
        y++

        a.kvs[x], a.kvs[y] = a.kvs[y], a.kvs[x] // Buf index.
}

func (a *segment) Less(i, j int) bool {
        x := i * 2
        y := j * 2

        kxLength := int((maskKeyLength & a.kvs[x]) >> 32)
        kxStart := int(a.kvs[x+1])
        kx := a.buf[kxStart : kxStart+kxLength]

        kyLength := int((maskKeyLength & a.kvs[y]) >> 32)
        kyStart := int(a.kvs[y+1])
        ky := a.buf[kyStart : kyStart+kyLength]

        return bytes.Compare(kx, ky) < 0
}

// ------------------------------------------------------

// FindStartKeyInclusivePos() returns the logical entry position for
// the given (inclusive) start key.  With segment keys of [b, d, f],
// looking for 'c' will return 1.  Looking for 'd' will return 1.
// Looking for 'g' will return 3.  Looking for 'a' will return 0.
func (a *segment) FindStartKeyInclusivePos(startKeyInclusive []byte) int {
        kvs := a.kvs
        buf := a.buf

        i, j := 0, a.Len()
        for i < j {
                h := i + (j-i)/2 // Keep i <= h < j.
                x := h * 2
                klen := int((maskKeyLength & kvs[x]) >> 32)
                kbeg := int(kvs[x+1])
                if bytes.Compare(buf[kbeg:kbeg+klen], startKeyInclusive) < 0 {
                        i = h + 1
                } else {
                        j = h
                }
        }

        return i

        // TODO: Do better than binary search?
        // TODO: Consider a perfectly balanced btree?
}

// GetOperationKeyVal() returns the operation, key, val for a given
// logical entry position in the segment.
func (a *segment) GetOperationKeyVal(pos int) (uint64, []byte, []byte) {
        x := pos * 2
        if x < len(a.kvs) {
                opklvl := a.kvs[x]
                kstart := int(a.kvs[x+1])
                operation, keyLen, valLen := decodeOpKeyLenValLen(opklvl)
                vstart := kstart + keyLen

                return operation, a.buf[kstart:vstart], a.buf[vstart : vstart+valLen]
        }

        return 0, nil, nil
}

// ------------------------------------------------------

func encodeOpKeyLenValLen(operation uint64, keyLen, valLen int) uint64 {
        return (maskOperation & operation) |
                (maskKeyLength & (uint64(keyLen) << 32)) |
                (maskValLength & (uint64(valLen)))
}

func decodeOpKeyLenValLen(opklvl uint64) (uint64, int, int) {
        operation := maskOperation & opklvl
        keyLen := int((maskKeyLength & opklvl) >> 32)
        valLen := int(maskValLength & opklvl)
        return operation, keyLen, valLen
}

// ------------------------------------------------------
// readyDeferredSort() will create a ticket for the future sorter and
// a channel to wait for its completion
func (a *segment) readyDeferredSort() {
        a.needSorterCh = make(chan bool, 1)
        a.needSorterCh <- true // A ticket for the future sorter.
        close(a.needSorterCh)

        a.waitSortedCh = make(chan struct{})
}

// RequestSort() will either perform the previously deferred sorting,
// if the goroutine can acquire the 1 ticket from the needSorterCh.
// Or, requestSort() will ensure that a sorter is working on this
// segment.  Returns true if the segment is sorted, and returns false
// if the sorting is only asynchronously scheduled.
func (a *segment) RequestSort(synchronous bool) bool {
        if a.needSorterCh == nil {
                return true
        }

        iAmTheSorter := <-a.needSorterCh
        if iAmTheSorter {
                sort.Sort(a)
                close(a.waitSortedCh) // Signal any waiters.
                return true
        }

        if synchronous {
                <-a.waitSortedCh // Wait for the sorter to be done.
                return true
        }

        return false
}

// doSort() will immediately sort this segment.
func (a *segment) doSort() {
        sort.Sort(a)
}

// ------------------------------------------------------

// Persist persists a basic segment, and allows a segment to meet the
// SegmentPersister interface.
func (seg *segment) Persist(file File) (rv SegmentLoc, err error) {
        finfo, err := file.Stat()
        if err != nil {
                return rv, err
        }
        pos := finfo.Size()

        kvsBuf, err := Uint64SliceToByteSlice(seg.kvs)
        if err != nil {
                return rv, err
        }

        kvsPos := pageAlign(pos)
        bufPos := pageAlign(kvsPos + int64(len(kvsBuf)))

        ioCh := make(chan ioResult)

        go func() {
                kvsWritten, err := file.WriteAt(kvsBuf, kvsPos)
                ioCh <- ioResult{kind: "kvs", want: len(kvsBuf), got: kvsWritten, err: err}
        }()

        go func() {
                bufWritten, err := file.WriteAt(seg.buf, bufPos)
                ioCh <- ioResult{kind: "buf", want: len(seg.buf), got: bufWritten, err: err}
        }()

        resMap := map[string]ioResult{}
        for len(resMap) < 2 {
                res := <-ioCh
                if res.err != nil {
                        return rv, res.err
                }
                if res.want != res.got {
                        return rv, fmt.Errorf("store: persistSegment error writing,"+
                                " res: %+v, err: %v", res, res.err)
                }
                resMap[res.kind] = res
        }

        close(ioCh)

        return SegmentLoc{
                Kind:       seg.Kind(),
                KvsOffset:  uint64(kvsPos),
                KvsBytes:   uint64(resMap["kvs"].got),
                BufOffset:  uint64(bufPos),
                BufBytes:   uint64(resMap["buf"].got),
                TotOpsSet:  seg.totOperationSet,
                TotOpsDel:  seg.totOperationDel,
                TotKeyByte: seg.totKeyByte,
                TotValByte: seg.totValByte,
        }, nil
}

// ------------------------------------------------------

// loadBasicSegment loads a basic segment.
func loadBasicSegment(sloc *SegmentLoc) (Segment, error) {
        var kvs []uint64
        var buf []byte
        var err error

        if sloc.KvsBytes > 0 {
                if sloc.KvsBytes > uint64(len(sloc.mref.buf)) {
                        return nil, fmt.Errorf("store_footer: KvsOffset/KvsBytes too big,"+
                                " len(mref.buf): %d, sloc: %+v", len(sloc.mref.buf), sloc)
                }

                kvsBytes := sloc.mref.buf[0:sloc.KvsBytes]
                kvs, err = ByteSliceToUint64Slice(kvsBytes)
                if err != nil {
                        return nil, err
                }
        }

        if sloc.BufBytes > 0 {
                bufStart := sloc.BufOffset - sloc.KvsOffset
                if bufStart+sloc.BufBytes > uint64(len(sloc.mref.buf)) {
                        return nil, fmt.Errorf("store_footer: BufOffset/BufBytes too big,"+
                                " len(mref.buf): %d, sloc: %+v", len(sloc.mref.buf), sloc)
                }

                buf = sloc.mref.buf[bufStart : bufStart+sloc.BufBytes]
        }
        return &segment{
                kvs:             kvs,
                buf:             buf,
                totOperationSet: sloc.TotOpsSet,
                totOperationDel: sloc.TotOpsDel,
                totKeyByte:      sloc.TotKeyByte,
                totValByte:      sloc.TotValByte,
        }, nil
}

// ------------------------------------------------------
type batch struct {
        // Compose batch as a type of segment extending it with childCollections
        *segment

        // childBatches track the segments of child collections indexed by their
        // unique collection names.
        childBatches map[string]*batch
}

// deletedChildBatchMarker is used as a conduit to convey the delete
// request from DelChildCollection() to ExecuteBatch()
var deletedChildBatchMarker = &batch{}

// newBatch() allocates a segment with hinted amount of resources.
func newBatch(collectionName string, options BatchOptions) (
        *batch, error) {
        return &batch{
                segment: &segment{
                        kvs: make([]uint64, 0, options.TotalOps*2),
                        buf: make([]byte, 0, options.TotalKeyValBytes),
                },
                childBatches: nil, // created later on demand
        }, nil
}

func (seg *batch) NewChildCollectionBatch(collectionName string,
        options BatchOptions) (b Batch, err error) {
        if len(collectionName) == 0 {
                return nil, ErrBadCollectionName
        }

        childBatch, err := newBatch(collectionName, options)

        if seg.childBatches == nil { // First creation of child batch.
                seg.childBatches = make(map[string]*batch)
        }

        seg.childBatches[collectionName] = childBatch

        // Directly return the child batch, allowing recursive recreations.
        return childBatch, err
}

func (seg *batch) DelChildCollection(collectionName string) error {
        if len(collectionName) == 0 {
                return ErrNoSuchCollection
        }
        if seg.childBatches == nil { // No previous child batches seen.
                seg.childBatches = make(map[string]*batch)
        }
        // Load the parent batch with this batch having special signature.
        seg.childBatches[collectionName] = deletedChildBatchMarker

        return nil
}

func (b *batch) readyDeferredSort() {
        if b == deletedChildBatchMarker {
                return
        }
        for _, childBatch := range b.childBatches {
                childBatch.readyDeferredSort()
        }

        b.segment.readyDeferredSort()
}

// RequestSort() returns true if all child batches are sorted and
// false if sorting has been asynchronously scheduled.
func (b *batch) RequestSort() bool {
        if b == deletedChildBatchMarker {
                return true
        }
        // false because we must never wait for sorter else it can deadlock.
        sorted := b.segment.RequestSort(false)

        for _, childBatch := range b.childBatches {
                sorted = childBatch.RequestSort() && sorted
        }
        return sorted
}

func (b *batch) doSort() {
        if b == deletedChildBatchMarker {
                return
        }
        sort.Sort(b.segment)
        for _, childBatch := range b.childBatches {
                childBatch.doSort()
        }
}

func (b *batch) isEmpty() bool {
        if len(b.childBatches) != 0 {
                // Very presence of child batches indicates a non-empty batch
                // even if the child batches themselves are empty. This is so that
                // collection creation/deletions can still work.
                return false
        }
        return b.Len() <= 0
}

1	// Copyright (c) 2016 Couchbase, Inc.
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the
4	// License. You may obtain a copy of the License at
5	// http://www.apache.org/licenses/LICENSE-2.0
6	// Unless required by applicable law or agreed to in writing,
7	// software distributed under the License is distributed on an "AS
8	// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
9	// express or implied. See the License for the specific language
10	// governing permissions and limitations under the License.
11
12	package moss
13
14	import (
15	"bytes"
16	"fmt"
17	"sort"
18	)
19
20	// BASIC_SEGMENT_KIND is the code for a basic, persistable segment
21	// implementation, which represents a segment as two arrays: an array
22	// of contiguous key-val bytes [key0, val0, key1, val1, ... keyN,
23	// valN], and an array of offsets plus lengths into the first array.
24	var BASIC_SEGMENT_KIND = "a"
25
26	func init() {	2✔
27	SegmentLoaders[BASIC_SEGMENT_KIND] = loadBasicSegment	2✔
28	}	2✔
29
30	// A Segment represents the read-oriented interface for a segment.
31	type Segment interface {
32	// Returns the kind of segment, used for persistence.
33	Kind() string
34
35	// Len returns the number of ops in the segment.
36	Len() int
37
38	// NumKeyValBytes returns the number of bytes used for key-val data.
39	NumKeyValBytes() (uint64, uint64)
40
41	// FindStartKeyInclusivePos() returns the logical entry position for
42	// the given (inclusive) start key. With segment keys of [b, d, f],
43	// looking for 'c' will return 1. Looking for 'd' will return 1.
44	// Looking for 'g' will return 3. Looking for 'a' will return 0.
45	FindStartKeyInclusivePos(startKeyInclusive []byte) int
46
47	// GetOperationKeyVal() returns the operation, key, val for a given
48	// logical entry position in the segment.
49	GetOperationKeyVal(pos int) (operation uint64, key []byte, val []byte)
50
51	// Returns true if the segment is already sorted, and returns
52	// false if the sorting is only asynchronously scheduled.
53	RequestSort(synchronous bool) bool
54	}
55
56	// A SegmentMutator represents the mutation methods of a segment.
57	type SegmentMutator interface {
58	Mutate(operation uint64, key, val []byte) error
59	}
60
61	// A SegmentPersister represents a segment that can be persisted.
62	type SegmentPersister interface {
63	Persist(file File) (SegmentLoc, error)
64	}
65
66	// A segment is a basic implementation of the segment related
67	// interfaces and represents a sequence of key-val entries or
68	// operations. A segment's kvs will be sorted by key when the segment
69	// is pushed into the collection. A segment implements the Batch
70	// interface.
71	type segment struct {
72	// Each key-val operation is encoded as 2 uint64's...
73	// - operation (see: maskOperation) \|
74	// key length (see: maskKeyLength) \|
75	// val length (see: maskValLength).
76	// - start index into buf for key-val bytes.
77	kvs []uint64
78
79	// Contiguous backing memory for the keys and vals of the segment.
80	buf []byte
81
82	// If this segment needs sorting, then needSorterCh will be
83	// non-nil and also the first goroutine that reads successfully
84	// from needSorterCh becomes the sorter of this segment. All
85	// other goroutines must instead wait on the waitSortedCh.
86	needSorterCh chan bool
87
88	// Once the sorter of this segment is done sorting the kvs, it
89	// close()'s the waitSortedCh, treating waitSortedCh like a
90	// one-way latch. The needSorterCh and waitSortedCh will either
91	// be nil or non-nil together. A segment that was "born
92	// sorted" will have needSorterCh and waitSortedCh as both nil.
93	waitSortedCh chan struct{}
94
95	totOperationSet uint64
96	totOperationDel uint64
97	totOperationMerge uint64
98	totKeyByte uint64
99	totValByte uint64
100	}
101
102	// See the OperationXxx consts.
103	const maskOperation = uint64(0x0F00000000000000)
104
105	// Max key length is 2^24, from 24 bits key length.
106	const maskKeyLength = uint64(0x00FFFFFF00000000)
107
108	const maxKeyLength = 1<<24 - 1
109
110	// Max val length is 2^28, from 28 bits val length.
111	const maskValLength = uint64(0x000000000FFFFFFF)
112
113	const maxValLength = 1<<28 - 1
114
115	const maskRESERVED = uint64(0xF0000000F0000000)
116
117	// newSegment() allocates a segment with hinted amount of resources.
118	func newSegment(totalOps, totalKeyValBytes int) (*segment, error) {	599✔
119	return &segment{	599✔
120	kvs: make([]uint64, 0, totalOps*2),	599✔
121	buf: make([]byte, 0, totalKeyValBytes),	599✔
122	}, nil	599✔
123	}	599✔
124
125	func (a *segment) Kind() string { return BASIC_SEGMENT_KIND }	100✔
126
127	// Close releases resources associated with the segment.
128	func (a *segment) Close() error {	627✔
129	return nil	627✔
130	}	627✔
131
132	// Set copies the key and val bytes into the segment as a "set"
133	// mutation. The key must be unique (not repeated) within the
134	// segment.
135	func (a *segment) Set(key, val []byte) error {	68,154✔
136	return a.mutate(OperationSet, key, val)	68,154✔
137	}	68,154✔
138
139	// Del copies the key bytes into the segment as a "deletion" mutation.
140	// The key must be unique (not repeated) within the segment.
141	func (a *segment) Del(key []byte) error {	4,585✔
142	return a.mutate(OperationDel, key, nil)	4,585✔
143	}	4,585✔
144
145	// Merge creates or updates a key-val entry in the Collection via the
146	// MergeOperator defined in the CollectionOptions. The key must be
147	// unique (not repeated) within the segment.
148	func (a *segment) Merge(key, val []byte) error {	25✔
149	return a.mutate(OperationMerge, key, val)	25✔
150	}	25✔
151
152	// ------------------------------------------------------
153
154	// Alloc provides a slice of bytes "owned" by the segment, to reduce
155	// extra copying of memory. See the Collection.NewBatch() method.
156	func (a *segment) Alloc(numBytes int) ([]byte, error) {	45✔
157	bufLen := len(a.buf)	45✔
158	bufCap := cap(a.buf)	45✔
159		45✔
160	if numBytes > bufCap-bufLen {	46✔
161	return nil, ErrAllocTooLarge	1✔
162	}	1✔
163
164	rv := a.buf[bufLen : bufLen+numBytes]	44✔
165		44✔
166	a.buf = a.buf[0 : bufLen+numBytes]	44✔
167		44✔
168	return rv, nil	44✔
169	}
170
171	// AllocSet is like Set(), but the caller must provide []byte
172	// parameters that came from Alloc(), for less buffer copying.
173	func (a *segment) AllocSet(keyFromAlloc, valFromAlloc []byte) error {	22✔
174	bufCap := cap(a.buf)	22✔
175		22✔
176	keyStart := bufCap - cap(keyFromAlloc)	22✔
177		22✔
178	return a.mutateEx(OperationSet,	22✔
179	keyStart, len(keyFromAlloc), len(valFromAlloc))	22✔
180	}	22✔
181
182	// AllocDel is like Del(), but the caller must provide []byte
183	// parameters that came from Alloc(), for less buffer copying.
184	func (a *segment) AllocDel(keyFromAlloc []byte) error {	1✔
185	bufCap := cap(a.buf)	1✔
186		1✔
187	keyStart := bufCap - cap(keyFromAlloc)	1✔
188		1✔
189	return a.mutateEx(OperationDel,	1✔
190	keyStart, len(keyFromAlloc), 0)	1✔
191	}	1✔
192
193	// AllocMerge is like Merge(), but the caller must provide []byte
194	// parameters that came from Alloc(), for less buffer copying.
195	func (a *segment) AllocMerge(keyFromAlloc, valFromAlloc []byte) error {	1✔
196	bufCap := cap(a.buf)	1✔
197		1✔
198	keyStart := bufCap - cap(keyFromAlloc)	1✔
199		1✔
200	return a.mutateEx(OperationMerge,	1✔
201	keyStart, len(keyFromAlloc), len(valFromAlloc))	1✔
202	}	1✔
203
204	// ------------------------------------------------------
205
206	func (a *segment) Mutate(operation uint64, key, val []byte) error {	89,642✔
207	return a.mutate(operation, key, val)	89,642✔
208	}	89,642✔
209
210	func (a *segment) mutate(operation uint64, key, val []byte) error {	160,265✔
211	keyStart := len(a.buf)	160,265✔
212	a.buf = append(a.buf, key...)	160,265✔
213	keyLength := len(a.buf) - keyStart	160,265✔
214		160,265✔
215	valStart := len(a.buf)	160,265✔
216	a.buf = append(a.buf, val...)	160,265✔
217	valLength := len(a.buf) - valStart	160,265✔
218		160,265✔
219	return a.mutateEx(operation, keyStart, keyLength, valLength)	160,265✔
220	}	160,265✔
221
222	func (a *segment) mutateEx(operation uint64,
223	keyStart, keyLength, valLength int) error {	160,480✔
224	if keyLength > maxKeyLength {	160,482✔
225	return ErrKeyTooLarge	2✔
226	}	2✔
227	if valLength > maxValLength {	160,555✔
228	return ErrValueTooLarge	1✔
229	}	1✔
230
231	if keyLength <= 0 && valLength <= 0 {	160,586✔
232	keyStart = 0	×
233	}	×
234
235	opKlVl := encodeOpKeyLenValLen(operation, keyLength, valLength)	160,618✔
236		160,618✔
237	a.kvs = append(a.kvs, opKlVl, uint64(keyStart))	160,618✔
238		160,618✔
239	switch operation {	160,618✔
240	case OperationSet:	150,557✔
241	a.totOperationSet++	150,557✔
242	case OperationDel:	10,456✔
243	a.totOperationDel++	10,456✔
244	case OperationMerge:	30✔
245	a.totOperationMerge++	30✔
246	default:	2✔
247	}
248
249	a.totKeyByte += uint64(keyLength)	160,827✔
250	a.totValByte += uint64(valLength)	160,827✔
251		160,827✔
252	return nil	160,827✔
253	}
254
255	// ------------------------------------------------------
256
257	// NumKeyValBytes returns the number of bytes used for key-val data.
258	func (a *segment) NumKeyValBytes() (uint64, uint64) {	1,760✔
259	return a.totKeyByte, a.totValByte	1,760✔
260	}	1,760✔
261
262	// ------------------------------------------------------
263
264	// Len returns the number of ops in the segment.
265	func (a *segment) Len() int {	62,451✔
266	return len(a.kvs) / 2	62,451✔
267	}	62,451✔
268
269	func (a *segment) Swap(i, j int) {	173,775✔
270	x := i * 2	173,775✔
271	y := j * 2	173,775✔
272		173,775✔
273	// Operation + key length + val length.	173,775✔
274	a.kvs[x], a.kvs[y] = a.kvs[y], a.kvs[x]	173,775✔
275		173,775✔
276	x++	173,775✔
277	y++	173,775✔
278		173,775✔
279	a.kvs[x], a.kvs[y] = a.kvs[y], a.kvs[x] // Buf index.	173,775✔
280	}	173,775✔
281
282	func (a *segment) Less(i, j int) bool {	586,276✔
283	x := i * 2	586,276✔
284	y := j * 2	586,276✔
285		586,276✔
286	kxLength := int((maskKeyLength & a.kvs[x]) >> 32)	586,276✔
287	kxStart := int(a.kvs[x+1])	586,276✔
288	kx := a.buf[kxStart : kxStart+kxLength]	586,276✔
289		586,276✔
290	kyLength := int((maskKeyLength & a.kvs[y]) >> 32)	586,276✔
291	kyStart := int(a.kvs[y+1])	586,276✔
292	ky := a.buf[kyStart : kyStart+kyLength]	586,276✔
293		586,276✔
294	return bytes.Compare(kx, ky) < 0	586,276✔
295	}	586,276✔
296
297	// ------------------------------------------------------
298
299	// FindStartKeyInclusivePos() returns the logical entry position for
300	// the given (inclusive) start key. With segment keys of [b, d, f],
301	// looking for 'c' will return 1. Looking for 'd' will return 1.
302	// Looking for 'g' will return 3. Looking for 'a' will return 0.
303	func (a *segment) FindStartKeyInclusivePos(startKeyInclusive []byte) int {	50,783✔
304	kvs := a.kvs	50,783✔
305	buf := a.buf	50,783✔
306		50,783✔
307	i, j := 0, a.Len()	50,783✔
308	for i < j {	295,124✔
309	h := i + (j-i)/2 // Keep i <= h < j.	244,341✔
310	x := h * 2	244,341✔
311	klen := int((maskKeyLength & kvs[x]) >> 32)	244,341✔
312	kbeg := int(kvs[x+1])	244,341✔
313	if bytes.Compare(buf[kbeg:kbeg+klen], startKeyInclusive) < 0 {	425,963✔
314	i = h + 1	181,622✔
315	} else {	245,447✔
316	j = h	63,825✔
317	}	63,825✔
318	}
319
320	return i	51,078✔
321
322	// TODO: Do better than binary search?
323	// TODO: Consider a perfectly balanced btree?
324	}
325
326	// GetOperationKeyVal() returns the operation, key, val for a given
327	// logical entry position in the segment.
328	func (a *segment) GetOperationKeyVal(pos int) (uint64, []byte, []byte) {	267,308✔
329	x := pos * 2	267,308✔
330	if x < len(a.kvs) {	500,045✔
331	opklvl := a.kvs[x]	232,737✔
332	kstart := int(a.kvs[x+1])	232,737✔
333	operation, keyLen, valLen := decodeOpKeyLenValLen(opklvl)	232,737✔
334	vstart := kstart + keyLen	232,737✔
335		232,737✔
336	return operation, a.buf[kstart:vstart], a.buf[vstart : vstart+valLen]	232,737✔
337	}	232,737✔
338
339	return 0, nil, nil	35,384✔
340	}
341
342	// ------------------------------------------------------
343
344	func encodeOpKeyLenValLen(operation uint64, keyLen, valLen int) uint64 {	217,523✔
345	return (maskOperation & operation) \|	217,523✔
346	(maskKeyLength & (uint64(keyLen) << 32)) \|	217,523✔
347	(maskValLength & (uint64(valLen)))	217,523✔
348	}	217,523✔
349
350	func decodeOpKeyLenValLen(opklvl uint64) (uint64, int, int) {	294,262✔
351	operation := maskOperation & opklvl	294,262✔
352	keyLen := int((maskKeyLength & opklvl) >> 32)	294,262✔
353	valLen := int(maskValLength & opklvl)	294,262✔
354	return operation, keyLen, valLen	294,262✔
355	}	294,262✔
356
357	// ------------------------------------------------------
358	// readyDeferredSort() will create a ticket for the future sorter and
359	// a channel to wait for its completion
360	func (a *segment) readyDeferredSort() {	42✔
361	a.needSorterCh = make(chan bool, 1)	42✔
362	a.needSorterCh <- true // A ticket for the future sorter.	42✔
363	close(a.needSorterCh)	42✔
364		42✔
365	a.waitSortedCh = make(chan struct{})	42✔
366	}	42✔
367
368	// RequestSort() will either perform the previously deferred sorting,
369	// if the goroutine can acquire the 1 ticket from the needSorterCh.
370	// Or, requestSort() will ensure that a sorter is working on this
371	// segment. Returns true if the segment is sorted, and returns false
372	// if the sorting is only asynchronously scheduled.
373	func (a *segment) RequestSort(synchronous bool) bool {	764✔
374	if a.needSorterCh == nil {	976✔
375	return true	212✔
376	}	212✔
377
378	iAmTheSorter := <-a.needSorterCh	552✔
379	if iAmTheSorter {	594✔
380	sort.Sort(a)	42✔
381	close(a.waitSortedCh) // Signal any waiters.	42✔
382	return true	42✔
383	}	42✔
384
385	if synchronous {	864✔
386	<-a.waitSortedCh // Wait for the sorter to be done.	354✔
387	return true	354✔
388	}	354✔
389
390	return false	156✔
391	}
392
393	// doSort() will immediately sort this segment.
394	func (a *segment) doSort() {	×
395	sort.Sort(a)	×
396	}	×
397
398	// ------------------------------------------------------
399
400	// Persist persists a basic segment, and allows a segment to meet the
401	// SegmentPersister interface.
402	func (seg *segment) Persist(file File) (rv SegmentLoc, err error) {	100✔
403	finfo, err := file.Stat()	100✔
404	if err != nil {	100✔
405	return rv, err	×
406	}	×
407	pos := finfo.Size()	100✔
408		100✔
409	kvsBuf, err := Uint64SliceToByteSlice(seg.kvs)	100✔
410	if err != nil {	100✔
411	return rv, err	×
412	}	×
413
414	kvsPos := pageAlign(pos)	100✔
415	bufPos := pageAlign(kvsPos + int64(len(kvsBuf)))	100✔
416		100✔
417	ioCh := make(chan ioResult)	100✔
418		100✔
419	go func() {	200✔
420	kvsWritten, err := file.WriteAt(kvsBuf, kvsPos)	100✔
421	ioCh <- ioResult{kind: "kvs", want: len(kvsBuf), got: kvsWritten, err: err}	100✔
422	}()	100✔
423
424	go func() {	200✔
425	bufWritten, err := file.WriteAt(seg.buf, bufPos)	100✔
426	ioCh <- ioResult{kind: "buf", want: len(seg.buf), got: bufWritten, err: err}	100✔
427	}()	100✔
428
429	resMap := map[string]ioResult{}	100✔
430	for len(resMap) < 2 {	300✔
431	res := <-ioCh	200✔
432	if res.err != nil {	200✔
433	return rv, res.err	×
434	}	×
435	if res.want != res.got {	200✔
436	return rv, fmt.Errorf("store: persistSegment error writing,"+	×
437	" res: %+v, err: %v", res, res.err)	×
438	}	×
439	resMap[res.kind] = res	200✔
440	}
441
442	close(ioCh)	100✔
443		100✔
444	return SegmentLoc{	100✔
445	Kind: seg.Kind(),	100✔
446	KvsOffset: uint64(kvsPos),	100✔
447	KvsBytes: uint64(resMap["kvs"].got),	100✔
448	BufOffset: uint64(bufPos),	100✔
449	BufBytes: uint64(resMap["buf"].got),	100✔
450	TotOpsSet: seg.totOperationSet,	100✔
451	TotOpsDel: seg.totOperationDel,	100✔
452	TotKeyByte: seg.totKeyByte,	100✔
453	TotValByte: seg.totValByte,	100✔
454	}, nil	100✔
455	}
456
457	// ------------------------------------------------------
458
459	// loadBasicSegment loads a basic segment.
460	func loadBasicSegment(sloc *SegmentLoc) (Segment, error) {	444✔
461	var kvs []uint64	444✔
462	var buf []byte	444✔
463	var err error	444✔
464		444✔
465	if sloc.KvsBytes > 0 {	881✔
466	if sloc.KvsBytes > uint64(len(sloc.mref.buf)) {	437✔
467	return nil, fmt.Errorf("store_footer: KvsOffset/KvsBytes too big,"+	×
468	" len(mref.buf): %d, sloc: %+v", len(sloc.mref.buf), sloc)	×
469	}	×
470
471	kvsBytes := sloc.mref.buf[0:sloc.KvsBytes]	437✔
472	kvs, err = ByteSliceToUint64Slice(kvsBytes)	437✔
473	if err != nil {	437✔
474	return nil, err	×
475	}	×
476	}
477
478	if sloc.BufBytes > 0 {	881✔
479	bufStart := sloc.BufOffset - sloc.KvsOffset	437✔
480	if bufStart+sloc.BufBytes > uint64(len(sloc.mref.buf)) {	437✔
481	return nil, fmt.Errorf("store_footer: BufOffset/BufBytes too big,"+	×
482	" len(mref.buf): %d, sloc: %+v", len(sloc.mref.buf), sloc)	×
483	}	×
484
485	buf = sloc.mref.buf[bufStart : bufStart+sloc.BufBytes]	437✔
486	}
487	return &segment{	444✔
488	kvs: kvs,	444✔
489	buf: buf,	444✔
490	totOperationSet: sloc.TotOpsSet,	444✔
491	totOperationDel: sloc.TotOpsDel,	444✔
492	totKeyByte: sloc.TotKeyByte,	444✔
493	totValByte: sloc.TotValByte,	444✔
494	}, nil	444✔
495	}
496
497	// ------------------------------------------------------
498	type batch struct {
499	// Compose batch as a type of segment extending it with childCollections
500	*segment
501
502	// childBatches track the segments of child collections indexed by their
503	// unique collection names.
504	childBatches map[string]*batch
505	}
506
507	// deletedChildBatchMarker is used as a conduit to convey the delete
508	// request from DelChildCollection() to ExecuteBatch()
509	var deletedChildBatchMarker = &batch{}
510
511	// newBatch() allocates a segment with hinted amount of resources.
512	func newBatch(collectionName string, options BatchOptions) (
513	*batch, error) {	1,195✔
514	return &batch{	1,195✔
515	segment: &segment{	1,195✔
516	kvs: make([]uint64, 0, options.TotalOps*2),	1,195✔
517	buf: make([]byte, 0, options.TotalKeyValBytes),	1,195✔
518	},	1,195✔
519	childBatches: nil, // created later on demand	1,195✔
520	}, nil	1,195✔
521	}	1,195✔
522
523	func (seg *batch) NewChildCollectionBatch(collectionName string,
524	options BatchOptions) (b Batch, err error) {	423✔
525	if len(collectionName) == 0 {	423✔
526	return nil, ErrBadCollectionName	×
527	}	×
528
529	childBatch, err := newBatch(collectionName, options)	423✔
530		423✔
531	if seg.childBatches == nil { // First creation of child batch.	844✔
532	seg.childBatches = make(map[string]*batch)	421✔
533	}	421✔
534
535	seg.childBatches[collectionName] = childBatch	423✔
536		423✔
537	// Directly return the child batch, allowing recursive recreations.	423✔
538	return childBatch, err	423✔
539	}
540
541	func (seg *batch) DelChildCollection(collectionName string) error {	3✔
542	if len(collectionName) == 0 {	3✔
543	return ErrNoSuchCollection	×
544	}	×
545	if seg.childBatches == nil { // No previous child batches seen.	6✔
546	seg.childBatches = make(map[string]*batch)	3✔
547	}	3✔
548	// Load the parent batch with this batch having special signature.
549	seg.childBatches[collectionName] = deletedChildBatchMarker	3✔
550		3✔
551	return nil	3✔
552	}
553
554	func (b *batch) readyDeferredSort() {	42✔
555	if b == deletedChildBatchMarker {	42✔
556	return	×
557	}	×
558	for _, childBatch := range b.childBatches {	42✔
559	childBatch.readyDeferredSort()	×
560	}	×
561
562	b.segment.readyDeferredSort()	42✔
563	}
564
565	// RequestSort() returns true if all child batches are sorted and
566	// false if sorting has been asynchronously scheduled.
567	func (b *batch) RequestSort() bool {	×
568	if b == deletedChildBatchMarker {	×
569	return true	×
570	}	×
571	// false because we must never wait for sorter else it can deadlock.
572	sorted := b.segment.RequestSort(false)	×
573		×
574	for _, childBatch := range b.childBatches {	×
575	sorted = childBatch.RequestSort() && sorted	×
576	}	×
577	return sorted	×
578	}
579
580	func (b *batch) doSort() {	1,154✔
581	if b == deletedChildBatchMarker {	1,157✔
582	return	3✔
583	}	3✔
584	sort.Sort(b.segment)	1,151✔
585	for _, childBatch := range b.childBatches {	1,577✔
586	childBatch.doSort()	426✔
587	}	426✔
588	}
589
590	func (b *batch) isEmpty() bool {	771✔
591	if len(b.childBatches) != 0 {	1,193✔
592	// Very presence of child batches indicates a non-empty batch	422✔
593	// even if the child batches themselves are empty. This is so that	422✔
594	// collection creation/deletions can still work.	422✔
595	return false	422✔
596	}	422✔
597	return b.Len() <= 0	349✔
598	}

couchbase / moss / 199

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