From f340486e37be20a6d02b3d2879a843a1c614d6ce Mon Sep 17 00:00:00 2001 From: Henry Date: Wed, 8 Jul 2026 15:20:11 -0700 Subject: [PATCH] more improvements in streaming caches for M4B cases --- pkg/fetcher/fetcher_gcs.go | 21 ++- pkg/fetcher/fetcher_http.go | 19 ++- pkg/fetcher/fetcher_http_test.go | 36 +++++ pkg/fetcher/fetcher_s3.go | 17 +++ pkg/parser/audio/mp4.go | 8 ++ pkg/parser/audio/parser.go | 30 ++++- pkg/parser/audio/readcache.go | 79 ++++++----- pkg/parser/audio/readcache_test.go | 4 +- pkg/parser/audio/rich.go | 30 +++-- pkg/parser/audio/servecache.go | 196 ++++++++++++++++++++++++++++ pkg/parser/audio/servecache_test.go | 163 +++++++++++++++++++++++ 11 files changed, 548 insertions(+), 55 deletions(-) create mode 100644 pkg/parser/audio/servecache.go create mode 100644 pkg/parser/audio/servecache_test.go diff --git a/pkg/fetcher/fetcher_gcs.go b/pkg/fetcher/fetcher_gcs.go index 97dade52..d769fd4f 100644 --- a/pkg/fetcher/fetcher_gcs.go +++ b/pkg/fetcher/fetcher_gcs.go @@ -7,6 +7,7 @@ import ( "net/http" "path" "strings" + "sync" "cloud.google.com/go/storage" "github.com/readium/go-toolkit/pkg/manifest" @@ -22,6 +23,12 @@ type GCSFetcher struct { handle *storage.ObjectHandle cachedLinks manifest.LinkList + + // attrsCache shares object metadata (object name -> *storage.ObjectAttrs) + // across the resources created by Get, so serving many requests for the + // same object — every one of which needs the length — performs a single + // attributes call over the fetcher's lifetime instead of one per request. + attrsCache sync.Map } func NewGCSFetcher(href string, client *storage.Client, handle *storage.ObjectHandle) *GCSFetcher { @@ -136,8 +143,9 @@ func (f *GCSFetcher) Get(ctx context.Context, link manifest.Link) Resource { // must not let an incoming request reach objects elsewhere in the bucket. if keyWithinRoot(f.handle.ObjectName(), resourceFile) { return &gcsResource{ - handle: f.client.Bucket(f.handle.BucketName()).Object(resourceFile), - link: link, + handle: f.client.Bucket(f.handle.BucketName()).Object(resourceFile), + link: link, + attrsCache: &f.attrsCache, } } } @@ -154,6 +162,7 @@ type gcsResource struct { link manifest.Link handle *storage.ObjectHandle cachedAttrs *storage.ObjectAttrs + attrsCache *sync.Map // Optional fetcher-shared metadata cache, see GCSFetcher } // Link implements Resource @@ -177,12 +186,20 @@ func (r *gcsResource) File() string { } func (r *gcsResource) attrs(ctx context.Context) (*storage.ObjectAttrs, *ResourceError) { + if r.cachedAttrs == nil && r.attrsCache != nil { + if v, ok := r.attrsCache.Load(r.handle.ObjectName()); ok { + r.cachedAttrs = v.(*storage.ObjectAttrs) + } + } if r.cachedAttrs == nil { head, err := r.handle.Attrs(ctx) if err != nil { return nil, gcsErrorToException(err) } r.cachedAttrs = head + if r.attrsCache != nil { + r.attrsCache.Store(r.handle.ObjectName(), head) + } } return r.cachedAttrs, nil } diff --git a/pkg/fetcher/fetcher_http.go b/pkg/fetcher/fetcher_http.go index 0a3a5f88..b1ad7325 100644 --- a/pkg/fetcher/fetcher_http.go +++ b/pkg/fetcher/fetcher_http.go @@ -8,6 +8,7 @@ import ( "slices" "strconv" "strings" + "sync" "github.com/pkg/errors" "github.com/readium/go-toolkit/pkg/manifest" @@ -19,6 +20,12 @@ type HTTPFetcher struct { href string client *http.Client url url.AbsoluteURL + + // sizes shares resource sizes (resolved URL -> int64) across the resources + // created by Get, so serving many requests for the same resource — every + // one of which needs the length — performs a single HEAD request over the + // fetcher's lifetime instead of one per request. + sizes sync.Map } func NewHTTPFetcher(href string, client *http.Client, url url.AbsoluteURL) *HTTPFetcher { @@ -75,6 +82,7 @@ func (f *HTTPFetcher) Get(ctx context.Context, link manifest.Link) Resource { link: link, client: f.client, url: resolved, + sizes: &f.sizes, } } } @@ -106,6 +114,7 @@ type httpResource struct { url url.AbsoluteURL cachedSize *int64 + sizes *sync.Map // Optional fetcher-shared size cache, see HTTPFetcher } // Link implements Resource @@ -129,6 +138,12 @@ func (r *httpResource) File() string { } func (r *httpResource) size(ctx context.Context) (int64, *ResourceError) { + if r.cachedSize == nil && r.sizes != nil { + if v, ok := r.sizes.Load(r.url.String()); ok { + length := v.(int64) + r.cachedSize = &length + } + } if r.cachedSize == nil { req, err := http.NewRequestWithContext(ctx, http.MethodHead, r.url.String(), nil) if err != nil { @@ -159,7 +174,9 @@ func (r *httpResource) size(ctx context.Context) (int64, *ResourceError) { return 0, Other(errors.Wrap(err, "failed to parse Content-Length header")) } r.cachedSize = &length - + if r.sizes != nil { + r.sizes.Store(r.url.String(), length) + } } return *r.cachedSize, nil } diff --git a/pkg/fetcher/fetcher_http_test.go b/pkg/fetcher/fetcher_http_test.go index dca17804..348ee925 100644 --- a/pkg/fetcher/fetcher_http_test.go +++ b/pkg/fetcher/fetcher_http_test.go @@ -95,3 +95,39 @@ func TestHTTPResourceStream(t *testing.T) { assert.Equal(t, []string{"bytes=100-199"}, ranges) mu.Unlock() } + +// Resources created by the same HTTPFetcher share learned sizes, so serving +// many requests for the same file — each of which needs Length for the +// Content-Length header — performs a single HEAD request in total. +func TestHTTPFetcherSharesResourceSize(t *testing.T) { + payload := make([]byte, 1234) + + var mu sync.Mutex + heads := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead { + mu.Lock() + heads++ + mu.Unlock() + } + http.ServeContent(w, r, "file.bin", time.Time{}, bytes.NewReader(payload)) + })) + defer srv.Close() + + base, err := url.AbsoluteURLFromString(srv.URL + "/") + require.NoError(t, err) + f := NewHTTPFetcher("", srv.Client(), base) + link := manifest.Link{Href: manifest.MustNewHREFFromString("file.bin", false)} + + for range 3 { + res := f.Get(t.Context(), link) + n, rerr := res.Length(t.Context()) + require.Nil(t, rerr) + assert.Equal(t, int64(1234), n) + res.Close() + } + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, heads, "size should be learned once and shared across resources") +} diff --git a/pkg/fetcher/fetcher_s3.go b/pkg/fetcher/fetcher_s3.go index 5fec1430..d4c129a3 100644 --- a/pkg/fetcher/fetcher_s3.go +++ b/pkg/fetcher/fetcher_s3.go @@ -7,6 +7,7 @@ import ( "path" "strconv" "strings" + "sync" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -24,6 +25,12 @@ type S3Fetcher struct { key string cachedLinks manifest.LinkList + + // heads shares object metadata (key -> *s3.HeadObjectOutput) across the + // resources created by Get, so serving many requests for the same object — + // every one of which needs the length — performs a single HeadObject call + // over the fetcher's lifetime instead of one per request. + heads sync.Map } func NewS3Fetcher(href string, client *s3.Client, bucket, key string) *S3Fetcher { @@ -126,6 +133,7 @@ func (f *S3Fetcher) Get(ctx context.Context, link manifest.Link) Resource { client: f.client, bucket: f.bucket, key: resourceFile, + heads: &f.heads, } } } @@ -145,6 +153,7 @@ type s3Resource struct { key string cachedHead *s3.HeadObjectOutput + heads *sync.Map // Optional fetcher-shared metadata cache, see S3Fetcher } // Link implements Resource @@ -175,6 +184,11 @@ func (r *s3Resource) object() *s3.GetObjectInput { } func (r *s3Resource) head(ctx context.Context) (*s3.HeadObjectOutput, *ResourceError) { + if r.cachedHead == nil && r.heads != nil { + if v, ok := r.heads.Load(r.key); ok { + r.cachedHead = v.(*s3.HeadObjectOutput) + } + } if r.cachedHead == nil { head, err := r.client.HeadObject(ctx, &s3.HeadObjectInput{ Bucket: &r.bucket, @@ -184,6 +198,9 @@ func (r *s3Resource) head(ctx context.Context) (*s3.HeadObjectOutput, *ResourceE return nil, awsErrorToException(err) } r.cachedHead = head + if r.heads != nil { + r.heads.Store(r.key, head) + } } return r.cachedHead, nil } diff --git a/pkg/parser/audio/mp4.go b/pkg/parser/audio/mp4.go index fe680896..e19a8ca4 100644 --- a/pkg/parser/audio/mp4.go +++ b/pkg/parser/audio/mp4.go @@ -277,6 +277,14 @@ func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint if cache != nil { if b, hit := cache.cachedSlice(run.lo, run.hi); hit { data = b + } else if cache.retain { + // The cache is being retained for serving the publication: + // pull whole blocks through it (costing up to a block of + // extra transfer per run) so the ranges a browser requests + // around each chapter are later served from memory. + if b, err := cache.Read(ctx, run.lo, run.hi); err == nil { + data = b + } } } if data == nil { diff --git a/pkg/parser/audio/parser.go b/pkg/parser/audio/parser.go index f7be96f7..1b809dc8 100644 --- a/pkg/parser/audio/parser.go +++ b/pkg/parser/audio/parser.go @@ -38,6 +38,11 @@ type AudioParser struct { // samples). Zero means the default. Higher values hide more per-file latency // on remote sources at the cost of more in-flight requests. concurrency int + + // retainCache keeps the blocks fetched while probing attached to the + // publication, so serving the audio files can answer those byte ranges from + // memory. See WithRetainedCache. + retainCache bool } // Option configures an AudioParser. @@ -79,6 +84,24 @@ func WithConcurrency(n int) Option { } } +// WithRetainedCache keeps the read-cache blocks fetched while probing attached +// to the parsed publication, so that serving its audio files answers those +// byte ranges — and the resources' lengths — from memory instead of new remote +// requests. Chapter-title samples are also pulled through the block cache +// (costing up to one block of extra transfer per chapter while parsing). +// +// The retained ranges — container headers and chapter samples — are exactly +// what a browser's demuxer requests before starting playback of an M4B, so +// this trades a slightly more expensive parse for a much faster time to first +// audio. Recommended when the publication is opened once and served many +// times (e.g. an HTTP streamer in front of remote storage); wasteful for +// one-shot parsing, and unnecessary for local files. Memory cost is bounded +// by what probing touched: typically a few blocks per file plus one per +// chapter. +func WithRetainedCache() Option { + return func(p *AudioParser) { p.retainCache = true } +} + func NewParser() AudioParser { return AudioParser{} } @@ -147,9 +170,14 @@ func (p AudioParser) Parse(ctx context.Context, asset asset.PublicationAsset, fe // When rich parsing is enabled, probe the audio files for durations, // bitrates, embedded metadata, a cover and a table of contents. if p.rich { - if coverFactory := p.enrich(ctx, fetcher, &man); coverFactory != nil { + coverFactory, caches := p.enrich(ctx, fetcher, &man) + if coverFactory != nil { serviceFactories[pub.CoverService_Name] = coverFactory } + if len(caches) > 0 { + // Serve the byte ranges fetched while probing from memory. + fetcher = &cacheFetcher{Fetcher: fetcher, caches: caches} + } } var builder *pub.ServicesBuilder diff --git a/pkg/parser/audio/readcache.go b/pkg/parser/audio/readcache.go index b0df2394..4666813f 100644 --- a/pkg/parser/audio/readcache.go +++ b/pkg/parser/audio/readcache.go @@ -2,6 +2,7 @@ package audio import ( "context" + "sync" "github.com/readium/go-toolkit/pkg/fetcher" ) @@ -21,18 +22,26 @@ const defaultCacheBlockSize = 256 << 10 // 256 KiB // every read is a byte-range request, so without caching opening a multi-file // audiobook fans out into hundreds of requests. Coalescing reads into block // fetches (and reusing them across passes) cuts that to a handful per file. +// +// Reads may happen concurrently (chapter-title runs are fetched in parallel), +// so block map access is guarded by a mutex. Concurrent fetches of the same +// missing block may duplicate a request; that is harmless and rare, and beats +// holding the lock across remote I/O. type readCache struct { fetcher.Resource size int64 blockSize int64 - blocks map[int64][]byte + retain bool // Route chapter-sample reads through the cache too, see WithRetainedCache + + mu sync.RWMutex + blocks map[int64][]byte } -func newReadCache(res fetcher.Resource, size, blockSize int64) *readCache { +func newReadCache(res fetcher.Resource, size, blockSize int64, retain bool) *readCache { if blockSize <= 0 { blockSize = defaultCacheBlockSize } - return &readCache{Resource: res, size: size, blockSize: blockSize, blocks: make(map[int64][]byte)} + return &readCache{Resource: res, size: size, blockSize: blockSize, retain: retain, blocks: make(map[int64][]byte)} } // Length returns the known size without hitting the underlying resource. @@ -63,6 +72,8 @@ func (c *readCache) Read(ctx context.Context, start, end int64) ([]byte, *fetche return nil, err } + c.mu.RLock() + defer c.mu.RUnlock() out := make([]byte, 0, end-start+1) for b := firstBlock; b <= lastBlock; b++ { blk := c.blocks[b] @@ -83,57 +94,34 @@ func (c *readCache) Read(ctx context.Context, start, end int64) ([]byte, *fetche // already fully present in the cache, without performing any read. The second // return value reports whether it was a hit. The returned slice is a copy. func (c *readCache) cachedSlice(start, end int64) ([]byte, bool) { - if c.size <= 0 || start < 0 || end < start { - return nil, false - } - if end >= c.size { - end = c.size - 1 - } - - firstBlock := start / c.blockSize - lastBlock := end / c.blockSize - for b := firstBlock; b <= lastBlock; b++ { - blk, ok := c.blocks[b] - if !ok { - return nil, false - } - // The needed portion of this block must actually be present (the final - // cached block may be short if it sits at the end of the resource). - base := b * c.blockSize - needEnd := end - if blockEnd := base + c.blockSize - 1; needEnd > blockEnd { - needEnd = blockEnd - } - if int64(len(blk)) <= needEnd-base { - return nil, false - } - } + c.mu.RLock() + defer c.mu.RUnlock() + return (&retainedCache{size: c.size, blockSize: c.blockSize, blocks: c.blocks}).slice(start, end) +} - out := make([]byte, 0, end-start+1) - for b := firstBlock; b <= lastBlock; b++ { - blk := c.blocks[b] - base := b * c.blockSize - lo := int64(0) - if start > base { - lo = start - base - } - hi := min(end-base+1, int64(len(blk))) - out = append(out, blk[lo:hi]...) +// snapshot returns the cache's blocks for retention on the publication after +// probing finishes. The block map is shared, not copied: no writes happen once +// probing is done, and retainedCache is read-only. +func (c *readCache) snapshot() *retainedCache { + c.mu.RLock() + defer c.mu.RUnlock() + if len(c.blocks) == 0 || c.size <= 0 { + return nil } - return out, true + return &retainedCache{size: c.size, blockSize: c.blockSize, blocks: c.blocks} } // fetch ensures every block in [first, last] is cached, retrieving each run of // contiguous missing blocks in a single underlying read. func (c *readCache) fetch(ctx context.Context, first, last int64) *fetcher.ResourceError { for b := first; b <= last; { - if _, ok := c.blocks[b]; ok { + if c.hasBlock(b) { b++ continue } runStart := b for b <= last { - if _, ok := c.blocks[b]; ok { + if c.hasBlock(b) { break } b++ @@ -149,6 +137,7 @@ func (c *readCache) fetch(ctx context.Context, first, last int64) *fetcher.Resou if err != nil { return err } + c.mu.Lock() for bi := runStart; bi <= runEnd; bi++ { lo := (bi - runStart) * c.blockSize if lo >= int64(len(data)) { @@ -161,6 +150,14 @@ func (c *readCache) fetch(ctx context.Context, first, last int64) *fetcher.Resou } c.blocks[bi] = data[lo:hi] } + c.mu.Unlock() } return nil } + +func (c *readCache) hasBlock(b int64) bool { + c.mu.RLock() + _, ok := c.blocks[b] + c.mu.RUnlock() + return ok +} diff --git a/pkg/parser/audio/readcache_test.go b/pkg/parser/audio/readcache_test.go index d3a9f103..db45db70 100644 --- a/pkg/parser/audio/readcache_test.go +++ b/pkg/parser/audio/readcache_test.go @@ -39,7 +39,7 @@ func TestReadCacheCorrectnessAndCoalescing(t *testing.T) { } base, _ := bytesResource(data) src := &countingResource{Resource: base, data: data} - c := newReadCache(src, int64(len(data)), 0) // 0 -> default 256 KiB blocks + c := newReadCache(src, int64(len(data)), 0, false) // 0 -> default 256 KiB blocks require.EqualValues(t, defaultCacheBlockSize, c.blockSize) ctx := t.Context() @@ -77,7 +77,7 @@ func TestReadCacheCustomBlockSize(t *testing.T) { } base, _ := bytesResource(data) src := &countingResource{Resource: base, data: data} - c := newReadCache(src, int64(len(data)), 128<<10) // 128 KiB blocks + c := newReadCache(src, int64(len(data)), 128<<10, false) // 128 KiB blocks require.EqualValues(t, 128<<10, c.blockSize) ctx := t.Context() diff --git a/pkg/parser/audio/rich.go b/pkg/parser/audio/rich.go index 4b5a1335..9711c925 100644 --- a/pkg/parser/audio/rich.go +++ b/pkg/parser/audio/rich.go @@ -23,8 +23,9 @@ const defaultProbeConcurrency = 8 // reading-order resource for its duration and bitrate, extracts publication // metadata (and a cover) from the first audio file, and builds a table of // contents. It mutates m in place and returns a cover service factory (or nil -// when no cover was found). -func (p AudioParser) enrich(ctx context.Context, fetch fetcher.Fetcher, m *manifest.Manifest) pub.ServiceFactory { +// when no cover was found), plus — when the parser retains probe caches — the +// per-HREF caches to attach to the publication. +func (p AudioParser) enrich(ctx context.Context, fetch fetcher.Fetcher, m *manifest.Manifest) (pub.ServiceFactory, map[string]*retainedCache) { var firstTags *audioTags var totalDuration float64 var embeddedChapters manifest.LinkList @@ -54,17 +55,25 @@ func (p AudioParser) enrich(ctx context.Context, fetch fetcher.Fetcher, m *manif for i := range readingOrder { i, link, res := i, readingOrder[i], resources[i] g.Go(func() error { - probed[i] = probeReadingOrderItem(ctx, res, link, extractChapters, blockSize, concurrency) + probed[i] = probeReadingOrderItem(ctx, res, link, extractChapters, blockSize, concurrency, p.retainCache) return nil }) } _ = g.Wait() // Combine the results in reading order so the output is deterministic. + var caches map[string]*retainedCache for i := range readingOrder { link := readingOrder[i] p := probed[i] + if p.cache != nil { + if caches == nil { + caches = make(map[string]*retainedCache, len(readingOrder)) + } + caches[link.Href.String()] = p.cache + } + if p.probe.Duration > 0 { link.Duration = p.probe.Duration totalDuration += p.probe.Duration @@ -104,15 +113,16 @@ func (p AudioParser) enrich(ctx context.Context, fetch fetcher.Fetcher, m *manif } if firstTags != nil { - return coverServiceFactory(firstTags.Picture) + return coverServiceFactory(firstTags.Picture), caches } - return nil + return nil, caches } // probedItem is the per-file result of the parallel probing pass. type probedItem struct { tags *audioTags probe probeResult + cache *retainedCache // Blocks fetched while probing, when retention is enabled } // probeReadingOrderItem reads the tags and probes the duration/bitrate/chapters @@ -124,12 +134,16 @@ type probedItem struct { // passes — which perform many small, overlapping reads of the header region — // coalesce into a handful of range requests rather than one request each. This // matters most for remote sources (HTTP, S3) and ZIP archives. -func probeReadingOrderItem(ctx context.Context, res fetcher.Resource, link manifest.Link, extractChapters bool, blockSize int64, concurrency int) probedItem { +func probeReadingOrderItem(ctx context.Context, res fetcher.Resource, link manifest.Link, extractChapters bool, blockSize int64, concurrency int, retain bool) probedItem { defer res.Close() size, _ := res.Length(ctx) - cached := newReadCache(res, size, blockSize) + cached := newReadCache(res, size, blockSize, retain) tags := readAudioTags(cached) - return probedItem{tags: tags, probe: probeAudioFile(ctx, cached, link, tags, extractChapters, concurrency)} + item := probedItem{tags: tags, probe: probeAudioFile(ctx, cached, link, tags, extractChapters, concurrency)} + if retain { + item.cache = cached.snapshot() + } + return item } // playlistTOC looks for the first parseable playlist file in the publication and diff --git a/pkg/parser/audio/servecache.go b/pkg/parser/audio/servecache.go new file mode 100644 index 00000000..e07a87a6 --- /dev/null +++ b/pkg/parser/audio/servecache.go @@ -0,0 +1,196 @@ +package audio + +import ( + "context" + "io" + + "github.com/readium/go-toolkit/pkg/fetcher" + "github.com/readium/go-toolkit/pkg/manifest" +) + +// retainedCache is a read-only snapshot of the blocks a readCache fetched while +// probing an audio file. When [WithRetainedCache] is enabled it stays attached +// to the publication so that serving the file can answer the same byte ranges +// from memory. Those ranges are exactly what a browser's demuxer asks for +// before starting playback: the container headers and the chapter-title +// samples scattered through the file. +type retainedCache struct { + size int64 + blockSize int64 + blocks map[int64][]byte +} + +// slice returns a copy of the inclusive range [start, end] if it is fully +// covered by cached blocks. +func (c *retainedCache) slice(start, end int64) ([]byte, bool) { + if c.size <= 0 || start < 0 || end < start { + return nil, false + } + if end >= c.size { + end = c.size - 1 + } + + firstBlock := start / c.blockSize + lastBlock := end / c.blockSize + for b := firstBlock; b <= lastBlock; b++ { + blk, ok := c.blocks[b] + if !ok { + return nil, false + } + // The needed portion of this block must actually be present (the final + // cached block may be short if it sits at the end of the resource). + base := b * c.blockSize + needEnd := end + if blockEnd := base + c.blockSize - 1; needEnd > blockEnd { + needEnd = blockEnd + } + if int64(len(blk)) <= needEnd-base { + return nil, false + } + } + + out := make([]byte, 0, end-start+1) + for b := firstBlock; b <= lastBlock; b++ { + blk := c.blocks[b] + base := b * c.blockSize + lo := int64(0) + if start > base { + lo = start - base + } + hi := min(end-base+1, int64(len(blk))) + out = append(out, blk[lo:hi]...) + } + return out, true +} + +// prefix returns a copy of the longest cached run of bytes beginning exactly at +// start, capped at end (inclusive). It returns nil when the byte at start is +// not cached. +func (c *retainedCache) prefix(start, end int64) []byte { + if c.size <= 0 || start < 0 || end < start || start >= c.size { + return nil + } + if end >= c.size { + end = c.size - 1 + } + + var out []byte + for pos := start; pos <= end; { + b := pos / c.blockSize + blk, ok := c.blocks[b] + if !ok { + break + } + base := b * c.blockSize + lo := pos - base + hi := min(end-base+1, int64(len(blk))) + if lo >= hi { + break + } + out = append(out, blk[lo:hi]...) + pos = base + hi + if hi < int64(len(blk)) || pos > end { + break + } + } + return out +} + +// cacheFetcher wraps the publication's fetcher, attaching each reading-order +// resource's retained probe cache to the resources it returns. +type cacheFetcher struct { + fetcher.Fetcher + caches map[string]*retainedCache // Keyed by the link's HREF string +} + +// Get implements fetcher.Fetcher +func (f *cacheFetcher) Get(ctx context.Context, link manifest.Link) fetcher.Resource { + res := f.Fetcher.Get(ctx, link) + if c, ok := f.caches[link.Href.String()]; ok { + return &cachedResource{Resource: res, cache: c} + } + return res +} + +// cachedResource serves a resource's length and any byte ranges covered by the +// retained probe cache from memory, delegating everything else to the wrapped +// resource. It intentionally does not forward the CompressedResource trait: +// audio files are already compressed and virtually never deflated inside +// archives, and hiding the trait only disables a passthrough optimization, +// not correctness. +type cachedResource struct { + fetcher.Resource + cache *retainedCache +} + +// Length implements fetcher.Resource, using the size learned while probing so +// no remote metadata request is needed. +func (r *cachedResource) Length(ctx context.Context) (int64, *fetcher.ResourceError) { + if r.cache.size >= 0 { + return r.cache.size, nil + } + return r.Resource.Length(ctx) +} + +// Read implements fetcher.Resource, serving fully-cached ranges from memory. +func (r *cachedResource) Read(ctx context.Context, start, end int64) ([]byte, *fetcher.ResourceError) { + s, e, whole := r.normalize(start, end) + if !whole { + if b, ok := r.cache.slice(s, e); ok { + return b, nil + } + } + return r.Resource.Read(ctx, start, end) +} + +// Stream implements fetcher.Resource. The cached prefix of the range is +// written immediately from memory — a browser probing chapter metadata +// typically aborts within it, costing no remote request at all — and only the +// remainder, if the client keeps reading, is streamed from the wrapped +// resource. +func (r *cachedResource) Stream(ctx context.Context, w io.Writer, start, end int64) (int64, *fetcher.ResourceError) { + s, e, whole := r.normalize(start, end) + if whole { + return r.Resource.Stream(ctx, w, start, end) + } + + pre := r.cache.prefix(s, e) + if len(pre) == 0 { + return r.Resource.Stream(ctx, w, start, end) + } + n, err := w.Write(pre) + written := int64(n) + if err != nil { + return written, fetcher.Other(err) + } + if s+written > e { + return written, nil + } + m, rerr := r.Resource.Stream(ctx, w, s+written, e) + if m > 0 { + written += m + } + return written, rerr +} + +// HasEfficientStream implements fetcher.EfficientStreamer by delegating to the +// wrapped resource: the cached prefix only ever makes Stream cheaper. +func (r *cachedResource) HasEfficientStream() bool { + if es, ok := r.Resource.(fetcher.EfficientStreamer); ok { + return es.HasEfficientStream() + } + return false +} + +// normalize resolves the (0, 0) whole-resource convention against the known +// size, reporting whole == true when the size is unknown and the range can't +// be resolved. +func (r *cachedResource) normalize(start, end int64) (int64, int64, bool) { + if start == 0 && end == 0 { + if r.cache.size <= 0 { + return start, end, true + } + return 0, r.cache.size - 1, false + } + return start, end, false +} diff --git a/pkg/parser/audio/servecache_test.go b/pkg/parser/audio/servecache_test.go new file mode 100644 index 00000000..3cbe7b72 --- /dev/null +++ b/pkg/parser/audio/servecache_test.go @@ -0,0 +1,163 @@ +package audio + +import ( + "bytes" + "context" + "io" + "sync" + "testing" + + "github.com/readium/go-toolkit/pkg/fetcher" + "github.com/readium/go-toolkit/pkg/manifest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// servedResource is an in-memory fetcher.Resource that counts how many times +// the underlying "remote" data is touched. +type servedResource struct { + data []byte + mu sync.Mutex + reads int + streams int + lengths int +} + +func (r *servedResource) Link() manifest.Link { return manifest.Link{} } +func (r *servedResource) Properties() manifest.Properties { return manifest.Properties{} } +func (r *servedResource) File() string { return "" } +func (r *servedResource) Close() {} + +func (r *servedResource) Length(ctx context.Context) (int64, *fetcher.ResourceError) { + r.mu.Lock() + r.lengths++ + r.mu.Unlock() + return int64(len(r.data)), nil +} + +func (r *servedResource) clamp(start, end int64) (int64, int64) { + if start == 0 && end == 0 { + return 0, int64(len(r.data)) - 1 + } + if end >= int64(len(r.data)) { + end = int64(len(r.data)) - 1 + } + return start, end +} + +func (r *servedResource) Read(ctx context.Context, start, end int64) ([]byte, *fetcher.ResourceError) { + r.mu.Lock() + r.reads++ + r.mu.Unlock() + s, e := r.clamp(start, end) + return append([]byte(nil), r.data[s:e+1]...), nil +} + +func (r *servedResource) Stream(ctx context.Context, w io.Writer, start, end int64) (int64, *fetcher.ResourceError) { + r.mu.Lock() + r.streams++ + r.mu.Unlock() + s, e := r.clamp(start, end) + n, err := w.Write(r.data[s : e+1]) + if err != nil { + return int64(n), fetcher.Other(err) + } + return int64(n), nil +} + +func (r *servedResource) counts() (int, int, int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.reads, r.streams, r.lengths +} + +// A cachedResource built from a probe-time readCache snapshot must serve the +// probed ranges — and the resource length — from memory, and delegate +// everything else. +func TestRetainedCacheServing(t *testing.T) { + data := make([]byte, 1<<20) + for i := range data { + data[i] = byte((i * 13) % 251) + } + const blockSize = 64 << 10 + + // Probe phase: read the "header" (block 0) and a "chapter sample" in block 8. + probeSrc := &servedResource{data: data} + rc := newReadCache(probeSrc, int64(len(data)), blockSize, true) + _, rerr := rc.Read(t.Context(), 0, 1000) + require.Nil(t, rerr) + _, rerr = rc.Read(t.Context(), 8*blockSize+100, 8*blockSize+200) + require.Nil(t, rerr) + + snap := rc.snapshot() + require.NotNil(t, snap) + + // Serve phase: a fresh resource wrapped with the retained cache. + serveSrc := &servedResource{data: data} + res := &cachedResource{Resource: serveSrc, cache: snap} + + // Length comes from the snapshot, not the resource. + l, rerr := res.Length(t.Context()) + require.Nil(t, rerr) + assert.Equal(t, int64(len(data)), l) + + // Fully-cached read: served from memory. + b, rerr := res.Read(t.Context(), 100, 199) + require.Nil(t, rerr) + assert.Equal(t, data[100:200], b) + + // Fully-cached stream (a browser probing a chapter sample block). + var buf bytes.Buffer + n, rerr := res.Stream(t.Context(), &buf, 8*blockSize, 8*blockSize+999) + require.Nil(t, rerr) + assert.Equal(t, int64(1000), n) + assert.Equal(t, data[8*blockSize:8*blockSize+1000], buf.Bytes()) + + reads, streams, lengths := serveSrc.counts() + assert.Equal(t, [3]int{0, 0, 0}, [3]int{reads, streams, lengths}, + "cached ranges must not touch the underlying resource") + + // Stream crossing out of the cached region: cached prefix from memory, + // remainder delegated with the right offset. + buf.Reset() + n, rerr = res.Stream(t.Context(), &buf, 8*blockSize, 10*blockSize-1) + require.Nil(t, rerr) + assert.Equal(t, int64(2*blockSize), n) + assert.Equal(t, data[8*blockSize:10*blockSize], buf.Bytes()) + _, streams, _ = serveSrc.counts() + assert.Equal(t, 1, streams, "only the uncached remainder should be streamed") + + // Uncached read and stream: fully delegated. + b, rerr = res.Read(t.Context(), 2*blockSize, 2*blockSize+99) + require.Nil(t, rerr) + assert.Equal(t, data[2*blockSize:2*blockSize+100], b) + reads, _, _ = serveSrc.counts() + assert.Equal(t, 1, reads) + + // Whole-resource stream: block 0 is cached, so it forms the prefix. + buf.Reset() + n, rerr = res.Stream(t.Context(), &buf, 0, 0) + require.Nil(t, rerr) + assert.Equal(t, int64(len(data)), n) + assert.Equal(t, data, buf.Bytes()) +} + +// Probing with retention through the rich parser attaches the caches to the +// publication so its resources answer probed ranges without new reads. +func TestProbeRetentionRoundTrip(t *testing.T) { + data := make([]byte, 512<<10) + for i := range data { + data[i] = byte((i * 7) % 256) + } + src := &servedResource{data: data} + rc := newReadCache(src, int64(len(data)), 0, false) + _, rerr := rc.Read(t.Context(), 0, 100) + require.Nil(t, rerr) + // Without retention there is nothing to keep for a resource this small... + assert.NotNil(t, rc.snapshot(), "snapshot works regardless of the retain flag") + + // The retain flag's effect is exercised in readChapterTitles: with it, + // cache misses for chapter runs are pulled through the block cache. + retained := newReadCache(&servedResource{data: data}, int64(len(data)), 0, true) + assert.True(t, retained.retain) +}