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(cfg config.StreamCacheConfig, logger logrus.FieldLogger) Cache
• type NullCache
  • func (NullCache) Fetch(ctx context.Context, key string, dst io.Writer, create func(io.Writer) error) (written int64, created bool, err error)
  • func (NullCache) Stop()
• type TestLogEntry
• type TestLoggingCache
  • func (tlc *TestLoggingCache) Entries() []*TestLogEntry
  • func (tlc *TestLoggingCache) Fetch(ctx context.Context, key string, dst io.Writer, create func(io.Writer) error) (written int64, created bool, err error)

Types

type Cache

type Cache interface {
	// Fetch finds or creates a cache entry and writes its contents into dst.
	// If the create callback is called the created return value is true. In
	// case of a non-nil error return, the create callback may still be
	// running in a goroutine for the benefit of another caller of Fetch with
	// the same key.
	Fetch(ctx context.Context, key string, dst io.Writer, create func(io.Writer) error) (written int64, 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(cfg config.StreamCacheConfig, 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) Fetch

func (NullCache) Fetch(ctx context.Context, key string, dst io.Writer, create func(io.Writer) error) (written int64, created bool, err error)

Fetch runs create(dst). The created flag is always true.

func (NullCache) Stop

func (NullCache) Stop()

Stop is a no-op.

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) Fetch

func (tlc *TestLoggingCache) Fetch(ctx context.Context, key string, dst io.Writer, create func(io.Writer) error) (written int64, created bool, err error)

Fetch calls the underlying Fetch method and logs the result.