s2randomaccess

README

Random Access over an S2 compressed stream

S2 is an extension of Snappy. This package builds on top of the exellent github.com/klauspost/compress/s2.

This package is similar to s2.ReadSeeker, but optimized for concurrent requests and caching decompression of blocks.

This package requires the entire stream as a []byte, which should not be modified during the lifetime of the *Seeker.

var compressedStream []byte

func baseLibrary() {
	seeker, err := s2.NewReader(bytes.NewReader(compressedStream)).ReadSeeker(true, nil)
	_, err := seeker.ReadAt(buf, uncompressedOffset)
}

func withS2RandomAccess() {
	seeker := s2randomaccess.Open(compressedStream)
	buf, deref, err := seeker.Get(uncompressedOffset, uncompressedLength)
	use(buf)
	deref()
}

Allocators

The default allocator simply uses make([]byte, n) and never reuses memory. WithAllocator(&SyncPoolAllocator{}) can be used to reuse buffers using sync.Pools. WithAllocator(s2ramalloc.Allocator{}) can be used to use malloc(3) and free(3), but requires CGO.

When using the default allocator, the returned deref functions can be ignored. With other allocators they must be called exactly once after being done with the returned slice.

Returned slices might point into memory allocated by the chosen Allocator, or straight into the input data if it was in an uncompressed data block in the stream.

Caching

This library caches the last 100 decoded blocks globally. The number of globally cached blocks can be changed with SetGlobalLRUSize(1000) at any time.

Documentation

Index

• Constants
• func PurgeGlobalCache()
• func SetGlobalLRUSize(n int)
• type Allocator
• type Option
  • func WithAllocator(a Allocator) Option
  • func WithAllowBuildIndex() Option
  • func WithIndex(idx s2.Index) Option
• type Seeker
  • func New(data []byte, options ...Option) (*Seeker, error)
  • func (s *Seeker) Get(offset, length int64) ([]byte, func(), error)
  • func (s *Seeker) ReadAt(dst []byte, offset int64) (int, error)
• type SyncPoolAllocator
  • func (a *SyncPoolAllocator) Alloc(n int) []byte
  • func (a *SyncPoolAllocator) Free(b []byte)

Constants

const (
	SyncPoolAllocatorSkipBuckets   = 6
	SyncPoolAllocatorLargestBucket = 33
)

Functions

func PurgeGlobalCache

func PurgeGlobalCache()

PurgeGlobalCache purges the global LRU cache that holds decompressed blocks. It is safe to call this function at any time.

func SetGlobalLRUSize

func SetGlobalLRUSize(n int)

SetGlobalLRUSize configures the number of blocks that can be in the global compressed blocks LRU at any time.

Types

type Allocator

type Allocator interface {
	Alloc(n int) []byte
	Free([]byte)
}

Allocator defines the methods a custom allocator needs to have.

type Option

type Option func(*Seeker) error

func WithAllocator

func WithAllocator(a Allocator) Option

WithAllocator can be passed to New() to use an alternative allocator.

func WithAllowBuildIndex

func WithAllowBuildIndex() Option

WithAllowBuildIndex fall back to indexing the stream ourselves if it isn't present in the stream.

func WithIndex

func WithIndex(idx s2.Index) Option

WithIndex passes the index rather than having it loaded from the stream.

type Seeker

type Seeker struct {
	// contains filtered or unexported fields
}

func New

func New(data []byte, options ...Option) (*Seeker, error)

New creates a new Seeker from a slice of compressed S2 (or Snappy) stream. The given slice is never written to. The index must be either: 1. At the end of the stream 2. Built during new (if WithAllowBuildIndex() is passed) 3. Given with WithIndex()

func (*Seeker) Get

func (s *Seeker) Get(offset, length int64) ([]byte, func(), error)

Get the uncompressed data at the given (uncompressed) offset. Will return a slice with length $length and a deref function that should be called exactly once. The returned slice is valid until a call to deref and should not be modified. If an error is returned, no deref function will be returned.

func (*Seeker) ReadAt

func (s *Seeker) ReadAt(dst []byte, offset int64) (int, error)

ReadAt reads uncompressed data from the compressed stream at the given (uncompressed) offset. Currently reads where offset+len(dst) is past the end of the stream fail with 0, io.ErrUnexpectedEOF, but that might be changed in the future to give a partial result together with io.EOF.

type SyncPoolAllocator

type SyncPoolAllocator struct {
	Pools [SyncPoolAllocatorLargestBucket - SyncPoolAllocatorSkipBuckets]sync.Pool
}

SyncPoolAllocator is an Allocator that uses one `sync.Pool` for each n**2 for n in (SyncPoolAllocatorSkipBuckets, SyncPoolAllocatorLargestBucket].

func (*SyncPoolAllocator) Alloc

func (a *SyncPoolAllocator) Alloc(n int) []byte

func (*SyncPoolAllocator) Free

func (a *SyncPoolAllocator) Free(b []byte)