diff --git a/pkg/fetcher/fetcher_file.go b/pkg/fetcher/fetcher_file.go index a3a67e2e..8db1ddc9 100644 --- a/pkg/fetcher/fetcher_file.go +++ b/pkg/fetcher/fetcher_file.go @@ -9,6 +9,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" "weak" "github.com/readium/go-toolkit/pkg/manifest" @@ -132,8 +133,9 @@ func NewFileFetcher(href string, fpath string) *FileFetcher { type FileResource struct { link manifest.Link path string + + mu sync.Mutex // guards file and sequential (offset-based) access to it file *os.File - read bool } // Link implements Resource @@ -148,6 +150,8 @@ func (r *FileResource) Properties() manifest.Properties { // Close implements Resource func (r *FileResource) Close() { + r.mu.Lock() + defer r.mu.Unlock() if r.file != nil { r.file.Close() } @@ -158,11 +162,13 @@ func (r *FileResource) File() string { return r.path } +// open returns the lazily-opened file handle. The returned *os.File is only +// safe for position-independent access (ReadAt, Stat); sequential access that +// moves the file offset must hold r.mu. func (r *FileResource) open() (*os.File, *ResourceError) { + r.mu.Lock() + defer r.mu.Unlock() if r.file != nil { - if _, err := r.file.Seek(0, io.SeekStart); err != nil { - return nil, Other(err) - } return r.file, nil } f, err := os.Open(r.path) @@ -183,18 +189,25 @@ func (r *FileResource) open() (*os.File, *ResourceError) { return f, nil } -// Read implements Resource +// Read implements Resource. Ranged reads (end > 0) are safe for concurrent use. func (r *FileResource) Read(ctx context.Context, start int64, end int64) ([]byte, *ResourceError) { defer runtime.KeepAlive(r) if end < start { return nil, RangeNotSatisfiable(errors.New("end of range smaller than start")) } + if start < 0 { + start = 0 + } f, ex := r.open() if ex != nil { return nil, ex } - r.read = true if start == 0 && end == 0 { + r.mu.Lock() + defer r.mu.Unlock() + if _, err := f.Seek(0, io.SeekStart); err != nil { + return nil, Other(err) + } data, err := io.ReadAll(f) if err != nil { return nil, Other(err) @@ -202,19 +215,11 @@ func (r *FileResource) Read(ctx context.Context, start int64, end int64) ([]byte return data, nil } data := make([]byte, end-start+1) - if start > 0 { - n, err := f.ReadAt(data, start) - if err != nil && err != io.EOF { - return nil, Other(err) - } - return data[:n], nil - } else { - n, err := io.ReadFull(f, data) - if err != nil && err != io.ErrUnexpectedEOF { - return nil, Other(err) - } - return data[:n], nil + n, err := f.ReadAt(data, start) + if err != nil && err != io.EOF { + return nil, Other(err) } + return data[:n], nil } // Stream implements Resource @@ -224,26 +229,27 @@ func (r *FileResource) Stream(ctx context.Context, w io.Writer, start int64, end err := RangeNotSatisfiable(errors.New("end of range smaller than start")) return -1, err } + if start < 0 { + start = 0 + } f, ex := r.open() if ex != nil { return -1, ex } - r.read = true if start == 0 && end == 0 { - n, err := io.Copy(w, f) - if err != nil { + r.mu.Lock() + defer r.mu.Unlock() + if _, err := f.Seek(0, io.SeekStart); err != nil { return -1, Other(err) } - return n, nil - } - if start > 0 { - _, err := f.Seek(start, 0) + n, err := io.Copy(w, f) if err != nil { return -1, Other(err) } + return n, nil } - n, err := io.CopyN(w, f, end-start+1) - if err != nil && err != io.EOF { + n, err := io.Copy(w, io.NewSectionReader(f, start, end-start+1)) + if err != nil { return n, Other(err) } return n, nil diff --git a/pkg/fetcher/fetcher_s3.go b/pkg/fetcher/fetcher_s3.go index 7954e5e0..45984286 100644 --- a/pkg/fetcher/fetcher_s3.go +++ b/pkg/fetcher/fetcher_s3.go @@ -206,7 +206,7 @@ func (r *s3Resource) Read(ctx context.Context, start int64, end int64) ([]byte, obj.Range = aws.String(sb.String()) } - output, err := r.client.GetObject(ctx, r.object()) + output, err := r.client.GetObject(ctx, obj) if err != nil { return nil, awsErrorToException(err) } diff --git a/pkg/fetcher/fetcher_s3_test.go b/pkg/fetcher/fetcher_s3_test.go new file mode 100644 index 00000000..7c1afdc4 --- /dev/null +++ b/pkg/fetcher/fetcher_s3_test.go @@ -0,0 +1,77 @@ +package fetcher + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strconv" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newTestS3Resource returns an s3Resource backed by a fake S3 endpoint serving +// the given object payload, honouring Range requests. +func newTestS3Resource(t *testing.T, payload []byte) (*s3Resource, *[]string) { + t.Helper() + + // Records the Range header of every GetObject request ("" when absent). + ranges := &[]string{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *ranges = append(*ranges, r.Header.Get("Range")) + + body := payload + if rng := r.Header.Get("Range"); rng != "" { + var start, end int64 + n, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end) + require.NoError(t, err) + require.Equal(t, 2, n) + require.LessOrEqual(t, start, end) + if end >= int64(len(payload)) { + end = int64(len(payload)) - 1 + } + body = payload[start : end+1] + w.Header().Set("Content-Range", + "bytes "+strconv.FormatInt(start, 10)+"-"+strconv.FormatInt(end, 10)+"/"+strconv.Itoa(len(payload))) + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.WriteHeader(http.StatusPartialContent) + } else { + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + } + w.Write(body) + })) + t.Cleanup(server.Close) + + client := s3.New(s3.Options{ + Region: "us-east-1", + Credentials: aws.AnonymousCredentials{}, + BaseEndpoint: aws.String(server.URL), + UsePathStyle: true, + }) + return &s3Resource{client: client, bucket: "bucket", key: "key"}, ranges +} + +func TestS3ResourceReadRange(t *testing.T) { + payload := []byte("0123456789abcdefghij") + res, ranges := newTestS3Resource(t, payload) + + data, err := res.Read(t.Context(), 5, 9) + require.Nil(t, err) + assert.Equal(t, []byte("56789"), data, "a ranged read must return only the requested bytes") + require.Len(t, *ranges, 1) + assert.Equal(t, "bytes=5-9", (*ranges)[0], "a ranged read must send a Range header") +} + +func TestS3ResourceReadWhole(t *testing.T) { + payload := []byte("0123456789abcdefghij") + res, ranges := newTestS3Resource(t, payload) + + data, err := res.Read(t.Context(), 0, 0) + require.Nil(t, err) + assert.Equal(t, payload, data) + require.Len(t, *ranges, 1) + assert.Equal(t, "", (*ranges)[0], "a whole-object read must not send a Range header") +} diff --git a/pkg/parser/audio/mp4.go b/pkg/parser/audio/mp4.go index d7fc975c..fe680896 100644 --- a/pkg/parser/audio/mp4.go +++ b/pkg/parser/audio/mp4.go @@ -9,6 +9,7 @@ import ( mp4 "github.com/abema/go-mp4" "github.com/readium/go-toolkit/pkg/fetcher" + "golang.org/x/sync/errgroup" ) // chapterCoalesceGap is the largest gap between two chapter text samples that is @@ -16,14 +17,23 @@ import ( // chapter track stored contiguously into one read. const chapterCoalesceGap = 64 << 10 +// defaultChapterReadConcurrency bounds how many coalesced sample runs are +// fetched in parallel when the parser doesn't configure a concurrency. Chapter +// titles are often interleaved with the audio (one tiny sample every chapter), +// so a book can need dozens of scattered reads; fetching them concurrently +// hides the per-request latency of remote sources. +const defaultChapterReadConcurrency = 8 + // probeMP4 extracts the total duration (in seconds) and any embedded chapters -// from an ISO-BMFF / MP4 container (m4a, m4b, m4p, mp4, …). +// from an ISO-BMFF / MP4 container (m4a, m4b, m4p, mp4, …). concurrency bounds +// the parallel reads used to fetch scattered chapter samples (<= 0 for the +// default). // // The duration is read from the movie header (`mvhd`). Chapters are read from a // QuickTime/iTunes chapter track: a track whose media handler is `text`, whose // samples are length-prefixed UTF strings located in `mdat`, and whose // per-sample timing comes from the `stts` table. -func probeMP4(ctx context.Context, res fetcher.Resource, extractChapters bool) (duration float64, chapters []chapterEntry, err error) { +func probeMP4(ctx context.Context, res fetcher.Resource, extractChapters bool, concurrency int) (duration float64, chapters []chapterEntry, err error) { rs := fetcher.NewResourceReadSeeker(res) // Total duration from the movie header. @@ -34,7 +44,7 @@ func probeMP4(ctx context.Context, res fetcher.Resource, extractChapters bool) ( } if extractChapters { - chapters = extractMP4Chapters(ctx, res, rs) + chapters = extractMP4Chapters(ctx, res, rs, concurrency) } return duration, chapters, nil } @@ -45,7 +55,7 @@ func probeMP4(ctx context.Context, res fetcher.Resource, extractChapters bool) ( // the movie header, which has already been fetched, so it costs no extra reads. // Otherwise it falls back to a QuickTime text chapter track, whose title samples // live in `mdat` and may be scattered throughout the file. -func extractMP4Chapters(ctx context.Context, res fetcher.Resource, rs *fetcher.ResourceReadSeeker) []chapterEntry { +func extractMP4Chapters(ctx context.Context, res fetcher.Resource, rs *fetcher.ResourceReadSeeker, concurrency int) []chapterEntry { if chapters := extractNeroChapters(ctx, res, rs); len(chapters) > 0 { return chapters } @@ -146,7 +156,7 @@ func extractMP4Chapters(ctx context.Context, res fetcher.Resource, rs *fetcher.R } } - titles := readChapterTitles(ctx, res, offsets, sampleSizes) + titles := readChapterTitles(ctx, res, offsets, sampleSizes, concurrency) chapters := make([]chapterEntry, 0, len(offsets)) for i := range offsets { @@ -208,11 +218,14 @@ func sampleOffsets(sampleSizes []uint32, chunkOffsets []uint64, stsc []mp4.StscE // per title, so instead this reuses already-cached blocks where possible and // otherwise reads the exact sample bytes from the underlying resource, coalescing // neighbouring samples into a single read. -func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint64, sizes []uint32) []string { +func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint64, sizes []uint32, concurrency int) []string { titles := make([]string, len(offsets)) if len(offsets) == 0 { return titles } + if concurrency <= 0 { + concurrency = defaultChapterReadConcurrency + } // Underlying resource for exact reads, plus the cache (if any) to reuse. var cache *readCache @@ -229,6 +242,12 @@ func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint } sort.Slice(order, func(a, b int) bool { return offsets[order[a]] < offsets[order[b]] }) + // Coalesce neighbouring samples into runs of [lo, hi] covering order[i:j]. + type sampleRun struct { + i, j int + lo, hi int64 + } + var runs []sampleRun for i := 0; i < len(order); { runStart := offsets[order[i]] runEnd := runStart + uint64(sizes[order[i]]) // exclusive @@ -243,30 +262,41 @@ func readChapterTitles(ctx context.Context, res fetcher.Resource, offsets []uint } j++ } + runs = append(runs, sampleRun{i: i, j: j, lo: int64(runStart), hi: int64(runEnd) - 1}) + i = j + } - lo, hi := int64(runStart), int64(runEnd)-1 - var data []byte - if cache != nil { - if b, hit := cache.cachedSlice(lo, hi); hit { - data = b + // Fetch the runs concurrently: titles are often interleaved with the audio, + // one run per chapter, and sequential round trips would dominate the open + // time on remote sources. Each goroutine writes to distinct title indexes. + var g errgroup.Group + g.SetLimit(concurrency) + for _, run := range runs { + g.Go(func() error { + var data []byte + if cache != nil { + if b, hit := cache.cachedSlice(run.lo, run.hi); hit { + data = b + } } - } - if data == nil { - if b, err := raw.Read(ctx, lo, hi); err == nil { - data = b + if data == nil { + if b, err := raw.Read(ctx, run.lo, run.hi); err == nil { + data = b + } } - } - for k := i; k < j; k++ { - idx := order[k] - start := int64(offsets[idx]) - lo - end := start + int64(sizes[idx]) - if start >= 0 && end <= int64(len(data)) { - titles[idx] = decodeChapterTitle(data[start:end]) + for k := run.i; k < run.j; k++ { + idx := order[k] + start := int64(offsets[idx]) - run.lo + end := start + int64(sizes[idx]) + if start >= 0 && end <= int64(len(data)) { + titles[idx] = decodeChapterTitle(data[start:end]) + } } - } - i = j + return nil + }) } + _ = g.Wait() return titles } diff --git a/pkg/parser/audio/parser.go b/pkg/parser/audio/parser.go index 2fb9ccbc..f7be96f7 100644 --- a/pkg/parser/audio/parser.go +++ b/pkg/parser/audio/parser.go @@ -33,9 +33,10 @@ type AudioParser struct { // range requests; smaller blocks transfer less for scattered reads. cacheBlockSize int - // concurrency caps how many audio files are probed in parallel. Zero means - // the default. Higher values hide more per-file latency on remote sources at - // the cost of more in-flight requests. + // concurrency caps how many audio files are probed in parallel, and how many + // parallel reads are used within a file (e.g. fetching scattered chapter + // 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 } @@ -64,8 +65,12 @@ func WithCacheBlockSize(size int) Option { } // WithConcurrency sets how many audio files are probed in parallel while -// extracting rich metadata. The default is 8. A value <= 0 is ignored and keeps -// the default. Use 1 to probe sequentially. +// extracting rich metadata, and also bounds the parallel reads used within a +// single file (e.g. fetching the scattered chapter-title samples of an MP4 +// chapter track). The default is 8. A value <= 0 is ignored and keeps the +// default. Use 1 to probe and read fully sequentially. Note that both levels +// can be in flight at once, so the worst-case number of concurrent requests is +// roughly the square of this value. func WithConcurrency(n int) Option { return func(p *AudioParser) { if n > 0 { diff --git a/pkg/parser/audio/probe.go b/pkg/parser/audio/probe.go index dd9119f8..72a07993 100644 --- a/pkg/parser/audio/probe.go +++ b/pkg/parser/audio/probe.go @@ -19,16 +19,17 @@ type probeResult struct { } // probeAudioFile extracts the duration, bitrate and any embedded chapters from a -// single audio resource. The bitrate is computed as the average over the whole -// resource, which is the most portable definition across the many supported -// container formats. -func probeAudioFile(ctx context.Context, res fetcher.Resource, link manifest.Link, tags *audioTags, extractChapters bool) probeResult { +// single audio resource. concurrency bounds the parallel reads used to fetch +// scattered chapter samples (<= 0 for the default). The bitrate is computed as +// the average over the whole resource, which is the most portable definition +// across the many supported container formats. +func probeAudioFile(ctx context.Context, res fetcher.Resource, link manifest.Link, tags *audioTags, extractChapters bool, concurrency int) probeResult { size, _ := res.Length(ctx) var result probeResult switch audioFamily(link) { case familyMP4: - duration, chapters, _ := probeMP4(ctx, res, extractChapters) + duration, chapters, _ := probeMP4(ctx, res, extractChapters, concurrency) result.Duration = duration result.Chapters = chapters case familyOgg: diff --git a/pkg/parser/audio/rich.go b/pkg/parser/audio/rich.go index ab094209..4b5a1335 100644 --- a/pkg/parser/audio/rich.go +++ b/pkg/parser/audio/rich.go @@ -54,7 +54,7 @@ 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) + probed[i] = probeReadingOrderItem(ctx, res, link, extractChapters, blockSize, concurrency) return nil }) } @@ -117,17 +117,19 @@ type probedItem struct { // probeReadingOrderItem reads the tags and probes the duration/bitrate/chapters // of a single reading-order resource, always releasing the resource handle. +// concurrency bounds the parallel reads used within the file (e.g. fetching +// scattered chapter samples). // // All reads go through a per-file block cache so that the tag and duration // 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) probedItem { +func probeReadingOrderItem(ctx context.Context, res fetcher.Resource, link manifest.Link, extractChapters bool, blockSize int64, concurrency int) probedItem { defer res.Close() size, _ := res.Length(ctx) cached := newReadCache(res, size, blockSize) tags := readAudioTags(cached) - return probedItem{tags: tags, probe: probeAudioFile(ctx, cached, link, tags, extractChapters)} + return probedItem{tags: tags, probe: probeAudioFile(ctx, cached, link, tags, extractChapters, concurrency)} } // playlistTOC looks for the first parseable playlist file in the publication and