streamcache

Documentation

Overview

Package streamcache provides a cache for large blobs (in the order of gigabytes). Because storing gigabytes of data is slow, cache entries can be streamed on the read end before they have finished on the write end. Because storing gigabytes of data is expensive, cache entries have a back pressure mechanism: if the readers don't make progress reading the data, the writers will block. That way our disk can fill up no faster than our readers can read from the cache.

The cache has 3 main parts: Cache (in-memory index), filestore (files to store the cached data in because it does not fit in memory), and pipe (coordinated IO to one file between one writer and multiple readers). A cache entry consists of a key, an maximum age, a pipe and the error result of the thing writing to the pipe.

Eviction

There are two eviction goroutines: one for Cache and one for filestore. The Cache eviction goroutine evicts entries after a set amount of time, and deletes their underlying files too. This is safe because Unix file semantics guarantee that readers/writers that are still using those files can keep using them. In addition to evicting known cache entries, we also have a goroutine at the filestore level which performs a directory walk. This will clean up cache files left behind by other processes.

Index

• type Cache
  • func New(dir string, maxAge time.Duration, logger logrus.FieldLogger) Cache
• type NullCache
  • func (NullCache) FindOrCreate(key string, create func(io.Writer) error) (s *Stream, created bool, err error)
  • func (NullCache) Stop()
• type Stream
  • func (s *Stream) Close() error
  • func (s *Stream) Read(p []byte) (int, error)
  • func (s *Stream) Wait(ctx context.Context) error
• type TestLogEntry
• type TestLoggingCache
  • func (tlc *TestLoggingCache) Entries() []*TestLogEntry
  • func (tlc *TestLoggingCache) FindOrCreate(key string, create func(io.Writer) error) (s *Stream, created bool, err error)

Types

type Cache

type Cache interface {
	// FindOrCreate finds or creates a cache entry. If the create callback
	// runs, it will be asynchronous and created is set to true. Callers must
	// Close() the returned stream to free underlying resources.
	FindOrCreate(key string, create func(io.Writer) error) (s *Stream, created bool, err error)
	// Stop stops the cleanup goroutines of the cache.
	Stop()
}

Cache is a cache for large byte streams.

func New

func New(dir string, maxAge time.Duration, logger logrus.FieldLogger) Cache

New returns a new cache instance.

type NullCache

type NullCache struct{}

NullCache is a null implementation of Cache. Every lookup is a miss, and it uses no storage.

func (NullCache) FindOrCreate

func (NullCache) FindOrCreate(key string, create func(io.Writer) error) (s *Stream, created bool, err error)

FindOrCreate runs create in a goroutine and lets the caller consume the result via the returned stream. The created flag is always true.

func (NullCache) Stop

func (NullCache) Stop()

Stop is a no-op.

type Stream

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

Stream abstracts a stream of bytes (via Read()) plus an error (via Wait()). Callers must always call Close() to prevent resource leaks.

func (*Stream) Close

func (s *Stream) Close() error

Close releases the underlying resources of the stream.

func (*Stream) Read

func (s *Stream) Read(p []byte) (int, error)

Read reads from the underlying stream of the stream.

func (*Stream) Wait

func (s *Stream) Wait(ctx context.Context) error

Wait returns the error value of the Stream. If ctx is canceled, Wait unblocks and returns early.

type TestLogEntry

type TestLogEntry struct {
	Key     string
	Created bool
	Err     error
}

TestLogEntry records the result of a cache lookup for testing purposes.

type TestLoggingCache

type TestLoggingCache struct {
	Cache
	// contains filtered or unexported fields
}

TestLoggingCache wraps a real Cache and logs all its lookups. This is not suitable for production because the log will grow indefinitely. Use only for testing.

func (*TestLoggingCache) Entries

func (tlc *TestLoggingCache) Entries() []*TestLogEntry

Entries returns a reference to the log of entries observed so far. This is a reference so the caller should not modify the underlying array or its elements.

func (*TestLoggingCache) FindOrCreate

func (tlc *TestLoggingCache) FindOrCreate(key string, create func(io.Writer) error) (s *Stream, created bool, err error)

FindOrCreate calls the underlying FindOrCreate method and logs the result.