M4B audiobook parsing improvements

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,38 +189,37 @@ 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)
 		}
 		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

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

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")
+}

pkg/parser/audio/mp4.go

@@ -9,21 +9,31 @@ 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
 // still read in a single request. It keeps over-reading small while collapsing a
 // 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
 }
 

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 {