Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
6.3 kB · 202 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203package s3
import ( "container/list" "errors" "fmt" "io" "sync")
// CachingReaderAt wraps a backend io.ReaderAt with a bounded LRU cache of// fixed-size blocks. It exists for the demuxer-over-S3 access pattern:// qtdemux seeks all over a large MP4, and the bare S3 ReaderAt turns every// non-sequential read into a fresh ranged GetObject. Serving reads out of// cached 16 MB blocks collapses those thousands of round-trips into a// handful of full-block fetches, while keeping memory bounded (maxBlocks *// blockSize) so we never have to download a whole (potentially many-GB)// upload up front.//// Reads are serialized by a single mutex, matching the bare ReaderAt and// the single-threaded gstreamer streaming thread that drives it. The// backend only ever sees aligned, full-block ReadAt calls.type CachingReaderAt struct { backend io.ReaderAt size int64 blockSize int64 maxBlocks int
mu sync.Mutex blocks map[int64]*list.Element // block index -> LRU element lru *list.List // front = most-recently-used everFetched map[int64]bool // blocks fetched at least once (redownload accounting) stats CacheStats}
// Defaults chosen from replaying real demuxer traces against the// simulation (see TestCachingReaderAtSim): 16 MB blocks minimize round// trips (a full 1.4 GB read went from ~120k ranged GETs to ~91), and 10// cached blocks (160 MB) leaves headroom for files whose tracks live in// separate regions, where the demuxer keeps multiple read-fronts alive.const ( DefaultCacheBlockSize = 16 * 1024 * 1024 DefaultCacheBlocks = 10)
type cacheEntry struct { index int64 data []byte}
// CacheStats is a snapshot of the cache's behavior, used both for the// offline simulation and (eventually) live metrics.type CacheStats struct { Reads int64 // ReadAt calls served BytesRequested int64 // sum of len(p) across ReadAt calls (clamped to size) BlockTouches int64 // block-level accesses (a read may touch several) Hits int64 // block touches served from cache Misses int64 // block touches that required a backend fetch ColdMisses int64 // misses for a block never fetched before Redownloads int64 // misses for a block that was fetched then evicted Evictions int64 // blocks dropped from the cache BackendReads int64 // ReadAt calls issued to the backend (== Misses) BackendBytes int64 // bytes pulled from the backend}
// NewCachingReaderAt wraps backend (whose total length is size) with an LRU// block cache of maxBlocks blocks of blockSize bytes each.func NewCachingReaderAt(backend io.ReaderAt, size, blockSize int64, maxBlocks int) (*CachingReaderAt, error) { if blockSize <= 0 { return nil, fmt.Errorf("blockSize must be positive, got %d", blockSize) } if maxBlocks <= 0 { return nil, fmt.Errorf("maxBlocks must be positive, got %d", maxBlocks) } return &CachingReaderAt{ backend: backend, size: size, blockSize: blockSize, maxBlocks: maxBlocks, blocks: make(map[int64]*list.Element), lru: list.New(), everFetched: make(map[int64]bool), }, nil}
// ReadAt implements io.ReaderAt, serving from cached blocks and fetching// (full, aligned) blocks from the backend on a miss. Returns io.EOF when a// read runs past the end of the object, matching io.ReaderAt semantics.func (c *CachingReaderAt) ReadAt(p []byte, off int64) (int, error) { if off < 0 { return 0, fmt.Errorf("negative offset %d", off) } if off >= c.size { return 0, io.EOF }
c.mu.Lock() defer c.mu.Unlock()
c.stats.Reads++ want := len(p) if int64(off)+int64(want) > c.size { want = int(c.size - off) } c.stats.BytesRequested += int64(want)
copied := 0 for copied < want { readOff := off + int64(copied) idx := readOff / c.blockSize block, err := c.getBlock(idx) if err != nil { return copied, err } within := int(readOff - idx*c.blockSize) n := copy(p[copied:want], block[within:]) copied += n } if copied < len(p) { // Caller asked for more than the object holds. return copied, io.EOF } return copied, nil}
// getBlock returns block idx, fetching it from the backend on a miss and// updating LRU/stats. Caller must hold c.mu.func (c *CachingReaderAt) getBlock(idx int64) ([]byte, error) { c.stats.BlockTouches++ if el, ok := c.blocks[idx]; ok { c.lru.MoveToFront(el) c.stats.Hits++ return el.Value.(*cacheEntry).data, nil }
c.stats.Misses++ if c.everFetched[idx] { c.stats.Redownloads++ } else { c.stats.ColdMisses++ c.everFetched[idx] = true }
start := idx * c.blockSize n := c.blockSize if start+n > c.size { n = c.size - start } buf := make([]byte, n) got, err := readAtFull(c.backend, buf, start) c.stats.BackendReads++ c.stats.BackendBytes += int64(got) if err != nil && !errors.Is(err, io.EOF) { return nil, fmt.Errorf("cache backend read block %d (offset %d): %w", idx, start, err) } buf = buf[:got]
el := c.lru.PushFront(&cacheEntry{index: idx, data: buf}) c.blocks[idx] = el if c.lru.Len() > c.maxBlocks { back := c.lru.Back() evicted := back.Value.(*cacheEntry) c.lru.Remove(back) delete(c.blocks, evicted.index) c.stats.Evictions++ } return buf, nil}
// Size returns the underlying object's size, satisfying blob.Reader.func (c *CachingReaderAt) Size() int64 { return c.size }
// Close closes the backend if it owns resources (e.g. the S3 ReaderAt's// open GetObject body), satisfying io.Closer / blob.Reader.func (c *CachingReaderAt) Close() error { if closer, ok := c.backend.(io.Closer); ok { return closer.Close() } return nil}
// Stats returns a copy of the current cache statistics.func (c *CachingReaderAt) Stats() CacheStats { c.mu.Lock() defer c.mu.Unlock() return c.stats}
// readAtFull reads len(buf) bytes via repeated ReadAt, tolerating short// reads from backends that don't fill the buffer in one call. Returns the// number of bytes read; io.EOF if the object ended first.func readAtFull(r io.ReaderAt, buf []byte, off int64) (int, error) { total := 0 for total < len(buf) { n, err := r.ReadAt(buf[total:], off+int64(total)) total += n if err != nil { return total, err } } return total, nil}