prometheus/tsdb/chunkenc/bstream.go
Bryan Boreham d166da7b59
tsdb: stop saving a copy of last 4 samples in memSeries (#11296)
* TSDB chunks: remove race between writing and reading

Because the data is stored as a bit-stream, the last byte in the stream
could change if the stream is appended to after an Iterator is obtained.
Copy the last byte when the Iterator is created, so we don't have to
read it later.

Clarify in comments that concurrent Iterator and Appender are allowed,
but the chunk must not be modified while an Iterator is created.
(This was already the case, in order to copy the bstream slice header.)

* TSDB: stop saving last 4 samples in memSeries

This extra copy of the last 4 samples was introduced to avoid a race
condition between reading the last byte of the chunk and writing to it.

But now we have fixed that by having `bstreamReader` copy the last byte,
we don't need to copy the last 4 samples.

This change saves 56 bytes per series, which is very worthwhile when
you have millions or tens of millions of series.

* TSDB: tidy up stopIterator re-use

Previous changes have left this code duplicating some lines; pull
them out to a separate function and tidy up.

* TSDB head_test: stop checking when iterators are wrapped

The behaviour has changed so chunk iterators are only wrapped when
transaction isolation requires them to stop short of the end.
This makes tests fail which are checking the type.

Tests should check the observable behaviour, not the type.

Signed-off-by: Bryan Boreham <bjboreham@gmail.com>
Signed-off-by: Ganesh Vernekar <ganeshvern@gmail.com>
Co-authored-by: Ganesh Vernekar <ganeshvern@gmail.com>
2022-09-27 19:32:05 +05:30

260 lines
7.4 KiB
Go

// Copyright 2017 The Prometheus Authors
// 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.
// The code in this file was largely written by Damian Gryski as part of
// https://github.com/dgryski/go-tsz and published under the license below.
// It received minor modifications to suit Prometheus's needs.
// Copyright (c) 2015,2016 Damian Gryski <damian@gryski.com>
// All rights reserved.
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// * Redistributions of source code must retain the above copyright notice,
// this list of conditions and the following disclaimer.
//
// * Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
package chunkenc
import (
"encoding/binary"
"io"
)
// bstream is a stream of bits.
type bstream struct {
stream []byte // The data stream.
count uint8 // How many right-most bits are available for writing in the current byte (the last byte of the stream).
}
func (b *bstream) bytes() []byte {
return b.stream
}
type bit bool
const (
zero bit = false
one bit = true
)
func (b *bstream) writeBit(bit bit) {
if b.count == 0 {
b.stream = append(b.stream, 0)
b.count = 8
}
i := len(b.stream) - 1
if bit {
b.stream[i] |= 1 << (b.count - 1)
}
b.count--
}
func (b *bstream) writeByte(byt byte) {
if b.count == 0 {
b.stream = append(b.stream, 0)
b.count = 8
}
i := len(b.stream) - 1
// Complete the last byte with the leftmost b.count bits from byt.
b.stream[i] |= byt >> (8 - b.count)
b.stream = append(b.stream, 0)
i++
// Write the remainder, if any.
b.stream[i] = byt << b.count
}
// writeBits writes the nbits right-most bits of u to the stream
// in left-to-right order.
func (b *bstream) writeBits(u uint64, nbits int) {
u <<= 64 - uint(nbits)
for nbits >= 8 {
byt := byte(u >> 56)
b.writeByte(byt)
u <<= 8
nbits -= 8
}
for nbits > 0 {
b.writeBit((u >> 63) == 1)
u <<= 1
nbits--
}
}
type bstreamReader struct {
stream []byte
streamOffset int // The offset from which read the next byte from the stream.
buffer uint64 // The current buffer, filled from the stream, containing up to 8 bytes from which read bits.
valid uint8 // The number of right-most bits valid to read (from left) in the current 8 byte buffer.
last byte // A copy of the last byte of the stream.
}
func newBReader(b []byte) bstreamReader {
// The last byte of the stream can be updated later, so we take a copy.
var last byte
if len(b) > 0 {
last = b[len(b)-1]
}
return bstreamReader{
stream: b,
last: last,
}
}
func (b *bstreamReader) readBit() (bit, error) {
if b.valid == 0 {
if !b.loadNextBuffer(1) {
return false, io.EOF
}
}
return b.readBitFast()
}
// readBitFast is like readBit but can return io.EOF if the internal buffer is empty.
// If it returns io.EOF, the caller should retry reading bits calling readBit().
// This function must be kept small and a leaf in order to help the compiler inlining it
// and further improve performances.
func (b *bstreamReader) readBitFast() (bit, error) {
if b.valid == 0 {
return false, io.EOF
}
b.valid--
bitmask := uint64(1) << b.valid
return (b.buffer & bitmask) != 0, nil
}
// readBits constructs a uint64 with the nbits right-most bits
// read from the stream, and any other bits 0.
func (b *bstreamReader) readBits(nbits uint8) (uint64, error) {
if b.valid == 0 {
if !b.loadNextBuffer(nbits) {
return 0, io.EOF
}
}
if nbits <= b.valid {
return b.readBitsFast(nbits)
}
// We have to read all remaining valid bits from the current buffer and a part from the next one.
bitmask := (uint64(1) << b.valid) - 1
nbits -= b.valid
v := (b.buffer & bitmask) << nbits
b.valid = 0
if !b.loadNextBuffer(nbits) {
return 0, io.EOF
}
bitmask = (uint64(1) << nbits) - 1
v = v | ((b.buffer >> (b.valid - nbits)) & bitmask)
b.valid -= nbits
return v, nil
}
// readBitsFast is like readBits but can return io.EOF if the internal buffer is empty.
// If it returns io.EOF, the caller should retry reading bits calling readBits().
// This function must be kept small and a leaf in order to help the compiler inlining it
// and further improve performances.
func (b *bstreamReader) readBitsFast(nbits uint8) (uint64, error) {
if nbits > b.valid {
return 0, io.EOF
}
bitmask := (uint64(1) << nbits) - 1
b.valid -= nbits
return (b.buffer >> b.valid) & bitmask, nil
}
func (b *bstreamReader) ReadByte() (byte, error) {
v, err := b.readBits(8)
if err != nil {
return 0, err
}
return byte(v), nil
}
// loadNextBuffer loads the next bytes from the stream into the internal buffer.
// The input nbits is the minimum number of bits that must be read, but the implementation
// can read more (if possible) to improve performances.
func (b *bstreamReader) loadNextBuffer(nbits uint8) bool {
if b.streamOffset >= len(b.stream) {
return false
}
// Handle the case there are more then 8 bytes in the buffer (most common case)
// in a optimized way. It's guaranteed that this branch will never read from the
// very last byte of the stream (which suffers race conditions due to concurrent
// writes).
if b.streamOffset+8 < len(b.stream) {
b.buffer = binary.BigEndian.Uint64(b.stream[b.streamOffset:])
b.streamOffset += 8
b.valid = 64
return true
}
// We're here if there are 8 or less bytes left in the stream.
// The following code is slower but called less frequently.
nbytes := int((nbits / 8) + 1)
if b.streamOffset+nbytes > len(b.stream) {
nbytes = len(b.stream) - b.streamOffset
}
buffer := uint64(0)
skip := 0
if b.streamOffset+nbytes == len(b.stream) {
// There can be concurrent writes happening on the very last byte
// of the stream, so use the copy we took at initialization time.
buffer = buffer | uint64(b.last)
// Read up to the byte before
skip = 1
}
for i := 0; i < nbytes-skip; i++ {
buffer = buffer | (uint64(b.stream[b.streamOffset+i]) << uint(8*(nbytes-i-1)))
}
b.buffer = buffer
b.streamOffset += nbytes
b.valid = uint8(nbytes * 8)
return true
}