diff --git a/BACKLOG.md b/BACKLOG.md index 63a59f8..b21b0a7 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -19,7 +19,7 @@ certificate pinning and metadata debugging to all TLS protocols. - [ ] **`--pinned-cert` for all TLS protocols** — extend certificate pinning to FTPS, Gemini, and WebDAVS (currently HTTP/HTTPS only). Also add SSH host-key pinning for SFTP as `--pinned-host-key`. -- [ ] **Unified worker pool** — a single protocol-agnostic worker pool +- [x] **Unified worker pool** — a single protocol-agnostic worker pool used by FTP, SFTP, and WebDAV recursive-parallel paths. Removes the three near-identical semaphore+WaitGroup implementations. diff --git a/internal/pool/pool.go b/internal/pool/pool.go new file mode 100644 index 0000000..1085986 --- /dev/null +++ b/internal/pool/pool.go @@ -0,0 +1,125 @@ +// Package pool provides a protocol-agnostic recursive worker pool. +// +// It replaces the three near-identical semaphore+WaitGroup +// implementations that existed in the FTP, SFTP, and WebDAV packages +// with a single generic type. Each protocol supplies its own task +// type and handler function; the pool takes care of concurrency +// limiting, goroutine lifecycle, and error collection. +// +// Tasks may submit additional tasks from inside the handler (tree +// recursion). Submit never blocks the caller—each task gets its +// own goroutine that waits for a semaphore slot—so recursive +// submissions cannot deadlock even when all pool slots are occupied. +package pool + +import ( + "context" + "sync" +) + +// ctxKey is an unexported type for context value keys in this package. +type ctxKey struct{} + +// Pool manages a bounded set of goroutines that execute tasks of +// type T. Tasks are submitted via Submit and executed by the +// handler function provided to New. The pool is safe for +// concurrent use from multiple goroutines. +type Pool[T any] struct { + handler func(ctx context.Context, task T) + + sem chan struct{} + wg sync.WaitGroup + errMu sync.Mutex + errs []error + ctx context.Context + cancel context.CancelFunc +} + +// New creates a pool with max parallelism workers. handler is +// called for every submitted task, running in its own goroutine +// bounded by the semaphore. The returned pool is ready to use. +// +// ctx is the parent context for all handler goroutines. When ctx +// is cancelled, waiting and running handlers observe the +// cancellation via the context passed to them. +// +// The context passed to handler carries a reference to the pool +// itself, retrievable via Get. This allows handlers to submit +// child tasks without an explicit closure capture. +func New[T any](ctx context.Context, maxParallelism int, handler func(ctx context.Context, task T)) *Pool[T] { + if maxParallelism < 1 { + maxParallelism = 1 + } + + ctx, cancel := context.WithCancel(ctx) + + return &Pool[T]{ + handler: handler, + sem: make(chan struct{}, maxParallelism), + cancel: cancel, + ctx: ctx, + } +} + +// Get retrieves the pool of type *Pool[T] from the context. +// It returns nil when the context does not carry a pool (for +// example, when called outside a handler goroutine). +func Get[T any](ctx context.Context) *Pool[T] { + v, _ := ctx.Value(ctxKey{}).(*Pool[T]) + return v +} + +// Submit enqueues a task for execution. Each submitted task gets +// its own goroutine that will wait for a semaphore slot before +// running the handler. Submit may be called from inside a handler +// (for recursive tasks). +// +// Submit is a no-op if the pool has already been waited on. +func (p *Pool[T]) Submit(task T) { + // After Wait returns the context is cancelled; skip work that + // can never run. + select { + case <-p.ctx.Done(): + return + default: + } + + p.wg.Add(1) + go func() { + defer p.wg.Done() + select { + case p.sem <- struct{}{}: + defer func() { <-p.sem }() + handlerCtx := context.WithValue(p.ctx, ctxKey{}, p) + p.handler(handlerCtx, task) + case <-p.ctx.Done(): + return + } + }() +} + +// Wait blocks until all submitted tasks (including tasks submitted +// by handlers) have completed. It returns all errors collected +// during execution. After Wait returns the pool must not be used. +func (p *Pool[T]) Wait() []error { + p.wg.Wait() + p.cancel() + + p.errMu.Lock() + defer p.errMu.Unlock() + if len(p.errs) == 0 { + return nil + } + return p.errs +} + +// AddError records a task-level error. It is safe to call from +// any goroutine, typically from inside a handler. +func (p *Pool[T]) AddError(err error) { + if err == nil { + return + } + p.errMu.Lock() + p.errs = append(p.errs, err) + p.errMu.Unlock() +} diff --git a/internal/pool/pool_test.go b/internal/pool/pool_test.go new file mode 100644 index 0000000..025c6e1 --- /dev/null +++ b/internal/pool/pool_test.go @@ -0,0 +1,210 @@ +//go:build linux || freebsd +// +build linux freebsd + +package pool + +import ( + "context" + "sort" + "sync" + "sync/atomic" + "testing" +) + +func TestPoolBasic(t *testing.T) { + var count atomic.Int64 + p := New(context.Background(), 4, func(_ context.Context, task int) { + count.Add(1) + }) + + for i := 0; i < 10; i++ { + p.Submit(i) + } + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } + if got := count.Load(); got != 10 { + t.Fatalf("processed %d tasks, want 10", got) + } +} + +func TestPoolConcurrencyLimit(t *testing.T) { + const parallel = 3 + var running atomic.Int64 + var maxRunning atomic.Int64 + + p := New(context.Background(), parallel, func(_ context.Context, task int) { + cur := running.Add(1) + for { + old := maxRunning.Load() + if cur <= old || maxRunning.CompareAndSwap(old, cur) { + break + } + } + // Simulate work so concurrent goroutines overlap. + for i := 0; i < 1000; i++ { + } + running.Add(-1) + }) + + for i := 0; i < 20; i++ { + p.Submit(i) + } + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } + if got := maxRunning.Load(); got > parallel { + t.Fatalf("max concurrent goroutines %d, want <= %d", got, parallel) + } +} + +func TestPoolRecursive(t *testing.T) { + // Simulate a tree: each task submits two children until depth 3. + // Expected total tasks: 1 + 2 + 4 + 8 = 15. + var count atomic.Int64 + + type task struct { + depth int + } + + p := New(context.Background(), 4, func(ctx context.Context, tk task) { + count.Add(1) + if tk.depth > 0 { + Get[task](ctx).Submit(task{depth: tk.depth - 1}) + Get[task](ctx).Submit(task{depth: tk.depth - 1}) + } + }) + + p.Submit(task{depth: 3}) + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } + if got := count.Load(); got != 15 { + t.Fatalf("processed %d tasks, want 15", got) + } +} + +func TestPoolErrorCollection(t *testing.T) { + p := New(context.Background(), 2, func(ctx context.Context, task int) { + if task%2 == 0 { + Get[int](ctx).AddError(&testError{task}) + } + }) + + for i := 0; i < 6; i++ { + p.Submit(i) + } + + errs := p.Wait() + if len(errs) != 3 { + t.Fatalf("collected %d errors, want 3", len(errs)) + } +} + +func TestGetOutsideHandler(t *testing.T) { + // Get returns nil when called outside a handler goroutine. + if got := Get[int](context.Background()); got != nil { + t.Fatalf("Get outside handler returned %v, want nil", got) + } +} + +func TestPoolContextCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + p := New(ctx, 4, func(taskCtx context.Context, task int) { + // Each task blocks until context is cancelled. + <-taskCtx.Done() + }) + + for i := 0; i < 5; i++ { + p.Submit(i) + } + + // Cancel after a short delay so the tasks start. + go func() { + // Give goroutines time to start and acquire semaphore. + cancel() + }() + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } +} + +func TestPoolDeterministic(t *testing.T) { + // Verify that all submitted tasks are processed exactly once. + const n = 100 + var mu sync.Mutex + seen := make(map[int]int) + + p := New(context.Background(), 8, func(_ context.Context, task int) { + mu.Lock() + seen[task]++ + mu.Unlock() + }) + + for i := 0; i < n; i++ { + p.Submit(i) + } + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } + + if len(seen) != n { + t.Fatalf("saw %d unique tasks, want %d", len(seen), n) + } + for k, v := range seen { + if v != 1 { + t.Errorf("task %d processed %d times, want 1", k, v) + } + } +} + +func TestPoolSortedResults(t *testing.T) { + // Collect results from concurrent workers and verify we get them all. + const n = 50 + var mu sync.Mutex + results := make([]int, 0, n) + + p := New(context.Background(), 4, func(_ context.Context, task int) { + mu.Lock() + results = append(results, task) + mu.Unlock() + }) + + for i := 0; i < n; i++ { + p.Submit(i) + } + + errs := p.Wait() + if len(errs) != 0 { + t.Fatalf("unexpected errors: %v", errs) + } + + sort.Ints(results) + if len(results) != n { + t.Fatalf("got %d results, want %d", len(results), n) + } + for i, v := range results { + if v != i { + t.Fatalf("results[%d] = %d, want %d", i, v, i) + } + } +} + +type testError struct { + task int +} + +func (e *testError) Error() string { + return "error in task" +} diff --git a/internal/protocol/ftp/client.go b/internal/protocol/ftp/client.go index 0df0f2c..7a53f9e 100644 --- a/internal/protocol/ftp/client.go +++ b/internal/protocol/ftp/client.go @@ -16,10 +16,10 @@ import ( "path/filepath" "strconv" "strings" - "sync" "time" "codeberg.org/petrbalvin/goget/internal/output" + "codeberg.org/petrbalvin/goget/internal/pool" ) type Client struct { @@ -698,6 +698,76 @@ func (p *clientPool) close() { } } +// ftpTask represents a unit of work for the concurrent recursive +// download: either a directory to list or a file to download. +type ftpTask struct { + remotePath string + localPath string + isDir bool + name string +} + +// ftpTaskHandler processes one ftpTask. For directories it lists the +// contents and submits child tasks for each entry. For files it +// acquires a client from the pool and downloads. +func ftpTaskHandler(cpool *clientPool, progressFunc func(file string, current, total int64)) func(ctx context.Context, t ftpTask) { + return func(ctx context.Context, t ftpTask) { + if ctx.Err() != nil { + return + } + + if err := os.MkdirAll(t.localPath, 0755); err != nil { + pool.Get[ftpTask](ctx).AddError(fmt.Errorf("failed to create local dir %s: %w", t.localPath, err)) + return + } + + if !t.isDir { + c := cpool.acquire() + defer cpool.release(c) + if progressFunc != nil { + progressFunc(t.name, 0, 0) + } + var fileProgress func(current, total int64) + if progressFunc != nil { + fileProgress = func(current, total int64) { + progressFunc(t.name, current, total) + } + } + if err := c.DownloadFile(ctx, t.remotePath, t.localPath, 0, fileProgress); err != nil { + pool.Get[ftpTask](ctx).AddError(fmt.Errorf("download %s: %w", t.remotePath, err)) + } + return + } + + c := cpool.acquire() + entries, err := c.ListRemoteDir(t.remotePath) + cpool.release(c) + if err != nil { + pool.Get[ftpTask](ctx).AddError(fmt.Errorf("list %s: %w", t.remotePath, err)) + return + } + + p := pool.Get[ftpTask](ctx) + for _, entry := range entries { + fields := strings.Fields(entry) + if len(fields) < 9 { + continue + } + name := fields[len(fields)-1] + if name == "." || name == ".." { + continue + } + childIsDir := fields[0][0] == 'd' + p.Submit(ftpTask{ + remotePath: filepath.Join(t.remotePath, name), + localPath: filepath.Join(t.localPath, name), + isDir: childIsDir, + name: name, + }) + } + } +} + // DownloadDirectoryConcurrent downloads a remote directory to localPath // using a pool of N FTP connections. All operations (LIST, RETR) use // absolute paths so concurrent workers never share or mutate CWD state @@ -708,110 +778,21 @@ func DownloadDirectoryConcurrent(ctx context.Context, rawURL string, timeout tim if parallel < 1 { parallel = 1 } - pool, err := newClientPool(rawURL, timeout, parallel) + cpool, err := newClientPool(rawURL, timeout, parallel) if err != nil { return err } - defer pool.close() + defer cpool.close() - sem := make(chan struct{}, parallel) - var wg sync.WaitGroup - var firstErr error - var errMu sync.Mutex + p := pool.New(ctx, parallel, ftpTaskHandler(cpool, progressFunc)) + p.Submit(ftpTask{ + remotePath: remotePath, + localPath: localPath, + isDir: true, + }) - setErr := func(err error) { - errMu.Lock() - if firstErr == nil { - firstErr = err - } - errMu.Unlock() - } - - processEntry(ctx, pool, sem, &wg, setErr, remotePath, localPath, progressFunc) - - wg.Wait() - return firstErr -} - -// processEntry processes one remote path: lists it, then dispatches -// its child entries (file → download, subdir → recursive processEntry) -// through the shared semaphore-bounded goroutine pool. A single -// sem is threaded through the recursion so the total in-flight -// goroutines never exceed `parallel`, regardless of directory depth. -func processEntry(ctx context.Context, pool *clientPool, sem chan struct{}, wg *sync.WaitGroup, setErr func(error), remotePath, localPath string, progressFunc func(file string, current, total int64)) { - defer wg.Done() - defer func() { <-sem }() - - if ctx.Err() != nil { - return - } - - if err := os.MkdirAll(localPath, 0755); err != nil { - setErr(fmt.Errorf("failed to create local dir %s: %w", localPath, err)) - return - } - - c := pool.acquire() - entries, err := c.ListRemoteDir(remotePath) - pool.release(c) - if err != nil { - setErr(fmt.Errorf("list %s: %w", remotePath, err)) - return - } - - for _, entry := range entries { - fields := strings.Fields(entry) - if len(fields) < 9 { - continue - } - name := fields[len(fields)-1] - if name == "." || name == ".." { - continue - } - isDir := fields[0][0] == 'd' - remoteFilePath := filepath.Join(remotePath, name) - localFilePath := filepath.Join(localPath, name) - - if isDir { - wg.Add(1) - select { - case sem <- struct{}{}: - go processEntry(ctx, pool, sem, wg, setErr, remoteFilePath, localFilePath, progressFunc) - case <-ctx.Done(): - return - } - } else { - wg.Add(1) - select { - case sem <- struct{}{}: - go func(remoteFilePath, localFilePath, name string) { - defer wg.Done() - defer func() { <-sem }() - if ctx.Err() != nil { - return - } - c := pool.acquire() - defer pool.release(c) - if progressFunc != nil { - progressFunc(name, 0, 0) - } - // DownloadFile's progress callback has a different - // signature than our directory-level progressFunc, so - // we wrap it here to keep the existing per-file - // progress behaviour. - var fileProgress func(current, total int64) - if progressFunc != nil { - fileProgress = func(current, total int64) { - progressFunc(name, current, total) - } - } - if err := c.DownloadFile(ctx, remoteFilePath, localFilePath, 0, fileProgress); err != nil { - setErr(fmt.Errorf("download %s: %w", remoteFilePath, err)) - } - }(remoteFilePath, localFilePath, name) - case <-ctx.Done(): - return - } - } + if errs := p.Wait(); len(errs) > 0 { + return errs[0] } + return nil } diff --git a/internal/protocol/sftp/sftp.go b/internal/protocol/sftp/sftp.go index 6daf6ae..756251a 100644 --- a/internal/protocol/sftp/sftp.go +++ b/internal/protocol/sftp/sftp.go @@ -20,6 +20,7 @@ import ( "codeberg.org/petrbalvin/goget/internal/core" "codeberg.org/petrbalvin/goget/internal/output" + "codeberg.org/petrbalvin/goget/internal/pool" "codeberg.org/petrbalvin/goget/internal/protocol" "golang.org/x/crypto/ssh" ) @@ -862,56 +863,113 @@ func downloadRecursiveSequential(ctx context.Context, req *core.DownloadRequest) return sftpDownloadDirStandalone(ctx, c, req) } +// sftpTask represents a unit of work for the concurrent recursive +// download: either a directory to list or a file to download. +type sftpTask struct { + req *core.DownloadRequest + isDir bool + size uint64 +} + // downloadRecursiveParallel downloads a remote directory using a pool -// of N SFTP connections. All workers share a single semaphore so the -// total in-flight goroutines never exceed N, regardless of directory -// depth. +// of N SFTP connections. All workers share a single pool so the total +// in-flight goroutines never exceed N, regardless of directory depth. func downloadRecursiveParallel(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) { parallel := req.RecursiveParallel if parallel < 1 { parallel = 1 } - pool, err := newSFTPClientPool(ctx, req, parallel) + cpool, err := newSFTPClientPool(ctx, req, parallel) if err != nil { return nil, err } - defer pool.close() + defer cpool.close() var ( statsMu sync.Mutex totalBytes int64 totalChunks int - firstErr error ) - sem := make(chan struct{}, parallel) - var wg sync.WaitGroup - - setErr := func(err error) { - statsMu.Lock() - if firstErr == nil { - firstErr = err - } - statsMu.Unlock() - } - - wg.Add(1) - sem <- struct{}{} - go func() { - defer wg.Done() - defer func() { <-sem }() - if ctx.Err() != nil { + p := pool.New(ctx, parallel, func(taskCtx context.Context, t sftpTask) { + if taskCtx.Err() != nil { return } - c := pool.acquire() - defer pool.release(c) - sftpDownloadDirWorker(ctx, c, req, &wg, sem, setErr, &statsMu, &totalBytes, &totalChunks, pool) - }() - wg.Wait() - if firstErr != nil { - return nil, firstErr + c := cpool.acquire() + defer cpool.release(c) + + if !t.isDir { + if err := sftpDownloadFile(taskCtx, c, t.req, t.size); err != nil { + pool.Get[sftpTask](taskCtx).AddError(fmt.Errorf("sftp download %s: %w", t.req.URL.Path, err)) + return + } + statsMu.Lock() + totalBytes += int64(t.size) + totalChunks++ + statsMu.Unlock() + return + } + + // Directory: list and submit child tasks. + remotePath := t.req.URL.Path + if remotePath == "" { + remotePath = "." + } + + absPath, err := c.realpath(remotePath) + if err != nil { + absPath = remotePath + } + + outputDir := t.req.Output + if outputDir == "" { + outputDir = filepath.Base(absPath) + if outputDir == "" || outputDir == "/" { + outputDir = "download" + } + } + + entries, err := sftpListDir(taskCtx, c, absPath) + if err != nil { + pool.Get[sftpTask](taskCtx).AddError(fmt.Errorf("sftp list %s: %w", absPath, err)) + return + } + + if err := os.MkdirAll(outputDir, 0755); err != nil { + pool.Get[sftpTask](taskCtx).AddError(fmt.Errorf("mkdir %s: %w", outputDir, err)) + return + } + + pp := pool.Get[sftpTask](taskCtx) + for _, e := range entries { + name := e.Name + if name == "." || name == ".." { + continue + } + entryIsDir := e.Attr != nil && sftpIsDir(e.Attr.Mode) + remoteChild := absPath + "/" + name + localChild := filepath.Join(outputDir, name) + subReq := *t.req + subReq.URL = &url.URL{Scheme: t.req.URL.Scheme, User: t.req.URL.User, Host: t.req.URL.Host, Path: remoteChild} + subReq.Output = localChild + if entryIsDir { + pp.Submit(sftpTask{req: &subReq, isDir: true}) + } else { + subReq.Recursive = false + subReq.RecursiveParallel = 0 + pp.Submit(sftpTask{req: &subReq, isDir: false, size: e.Attr.Size}) + } + } + }) + + p.Submit(sftpTask{req: req, isDir: true}) + + if errs := p.Wait(); len(errs) > 0 { + return nil, errs[0] } + statsMu.Lock() + defer statsMu.Unlock() return &core.DownloadResult{ BytesDownloaded: totalBytes, Duration: 0, @@ -958,115 +1016,6 @@ func sftpDownloadDirStandalone(ctx context.Context, c *client, req *core.Downloa }, nil } -// sftpDownloadDirWorker is the worker-pool variant. It lists the -// directory, then dispatches file downloads and subdirectory recursion -// through the shared semaphore. Subdirectory workers reuse the same -// pool to acquire clients. The return value is always nil; results -// and errors are routed through the shared fields. -func sftpDownloadDirWorker( - ctx context.Context, - c *client, - req *core.DownloadRequest, - wg *sync.WaitGroup, - sem chan struct{}, - setErr func(error), - statsMu *sync.Mutex, - totalBytes *int64, - totalChunks *int, - pool *sftpClientPool, -) { - remotePath := req.URL.Path - if remotePath == "" { - remotePath = "." - } - - absPath, err := c.realpath(remotePath) - if err != nil { - absPath = remotePath - } - - outputDir := req.Output - if outputDir == "" { - outputDir = filepath.Base(absPath) - if outputDir == "" || outputDir == "/" { - outputDir = "download" - } - } - - entries, err := sftpListDir(ctx, c, absPath) - if err != nil { - setErr(fmt.Errorf("sftp list %s: %w", absPath, err)) - return - } - - if err := os.MkdirAll(outputDir, 0755); err != nil { - setErr(fmt.Errorf("mkdir %s: %w", outputDir, err)) - return - } - - for i, e := range entries { - name := e.Name - if name == "." || name == ".." { - continue - } - isDir := e.Attr != nil && sftpIsDir(e.Attr.Mode) - remoteChild := absPath + "/" + name - localChild := filepath.Join(outputDir, name) - _ = i - - if isDir { - wg.Add(1) - select { - case sem <- struct{}{}: - go func(remoteChild, localChild string) { - defer wg.Done() - defer func() { <-sem }() - if ctx.Err() != nil { - return - } - cc := pool.acquire() - defer pool.release(cc) - subReq := *req - subReq.URL = &url.URL{Scheme: req.URL.Scheme, User: req.URL.User, Host: req.URL.Host, Path: remoteChild} - subReq.Output = localChild - sftpDownloadDirWorker(ctx, cc, &subReq, wg, sem, setErr, statsMu, totalBytes, totalChunks, pool) - }(remoteChild, localChild) - case <-ctx.Done(): - return - } - } else { - wg.Add(1) - select { - case sem <- struct{}{}: - go func(remoteChild, localChild string, size uint64) { - defer wg.Done() - defer func() { <-sem }() - if ctx.Err() != nil { - return - } - cc := pool.acquire() - defer pool.release(cc) - subReq := *req - subReq.URL = &url.URL{Scheme: req.URL.Scheme, User: req.URL.User, Host: req.URL.Host, Path: remoteChild} - subReq.Output = localChild - subReq.Recursive = false - subReq.RecursiveParallel = 0 - if err := sftpDownloadFile(ctx, cc, &subReq, size); err != nil { - setErr(fmt.Errorf("sftp download %s: %w", remoteChild, err)) - return - } - statsMu.Lock() - *totalBytes += int64(size) - *totalChunks++ - statsMu.Unlock() - }(remoteChild, localChild, e.Attr.Size) - case <-ctx.Done(): - return - } - } - } -} - // sftpWalkDir recursively walks one remote directory on `c`, // downloading each file and recursing into subdirectories. Stats are // accumulated inline. Used by the standalone mode. diff --git a/internal/protocol/webdav/protocol.go b/internal/protocol/webdav/protocol.go index 0beef30..d1d4728 100644 --- a/internal/protocol/webdav/protocol.go +++ b/internal/protocol/webdav/protocol.go @@ -21,6 +21,7 @@ import ( "codeberg.org/petrbalvin/goget/internal/core" "codeberg.org/petrbalvin/goget/internal/output" + "codeberg.org/petrbalvin/goget/internal/pool" "codeberg.org/petrbalvin/goget/internal/protocol" "codeberg.org/petrbalvin/goget/internal/transport" ) @@ -982,9 +983,18 @@ func (p *Protocol) downloadRecursiveSequential(ctx context.Context, req *core.Do }, nil } +// webdavTask represents a unit of work for the concurrent recursive +// download: a file or subdirectory entry with its pre-built request. +type webdavTask struct { + entry WebDAVEntry + subReq *core.DownloadRequest + outPath string + isDir bool +} + // downloadRecursiveParallel is the worker-pool implementation used when // req.RecursiveParallel > 1. Top-level entries are processed concurrently, -// bounded by a semaphore. Subdirectory recursion is delegated back to +// bounded by the pool. Subdirectory recursion is delegated back to // downloadRecursive, so the same RecursiveParallel limit applies at every // level of the tree. Stats are aggregated under a mutex. func (p *Protocol) downloadRecursiveParallel(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) { @@ -1021,24 +1031,52 @@ func (p *Protocol) downloadRecursiveParallel(ctx context.Context, req *core.Down // these counters, so they reflect the entire subtree processed by // this call. var ( - statsMu sync.Mutex - totalBytes int64 - totalChunks int - processedDirs int + statsMu sync.Mutex + totalBytes int64 + totalChunks int ) - processedDirs++ + processedDirs := 1 - // Pre-build sub-requests and output paths so the goroutine body has - // no per-entry allocation cost. - type task struct { - entry WebDAVEntry - subReq *core.DownloadRequest - outPath string - isDir bool - } + wp := pool.New(ctx, req.RecursiveParallel, func(taskCtx context.Context, t webdavTask) { + if taskCtx.Err() != nil { + return + } + + if t.isDir { + subResult, err := p.downloadRecursive(taskCtx, t.subReq) + if err != nil { + if req.Verbose { + fmt.Fprintf(os.Stderr, "[webdav] Warning: failed to %s sub-directory %s: %v\n", dryRunVerb(req.DryRun), t.entry.URL.String(), err) + } + return + } + statsMu.Lock() + totalBytes += subResult.BytesDownloaded + totalChunks += subResult.ChunksCount + statsMu.Unlock() + return + } + + subResult, err := p.downloadFile(taskCtx, t.subReq, t.entry.Size, t.entry.LastModified) + if err != nil { + if req.Verbose { + fmt.Fprintf(os.Stderr, "[webdav] Warning: failed to download file %s: %v\n", t.entry.URL.String(), err) + } + return + } + + // Apply keep-timestamps if requested. + if req.KeepTimestamps { + setFileTimestamp(t.outPath, t.entry.LastModified) + } + + statsMu.Lock() + totalBytes += subResult.BytesDownloaded + totalChunks++ + statsMu.Unlock() + }) - tasks := make([]task, 0, len(filteredEntries)) for _, entry := range filteredEntries { // Respect MaxDepth to avoid infinite recursion. if entry.IsDir && req.MaxDepth == 0 { @@ -1065,7 +1103,7 @@ func (p *Protocol) downloadRecursiveParallel(ctx context.Context, req *core.Down subReq.MaxDepth = req.MaxDepth - 1 subReq.RecursiveParallel = req.RecursiveParallel subReq.DryRun = true - tasks = append(tasks, task{entry: entry, subReq: &subReq, outPath: itemOutputPath, isDir: true}) + wp.Submit(webdavTask{entry: entry, subReq: &subReq, outPath: itemOutputPath, isDir: true}) } else { fmt.Fprintf(os.Stderr, "[webdav] [dry-run] would download: %s (%d bytes)\n", entry.URL.String(), entry.Size) statsMu.Lock() @@ -1106,7 +1144,7 @@ func (p *Protocol) downloadRecursiveParallel(ctx context.Context, req *core.Down RejectPatterns: req.RejectPatterns, Ctx: req.Ctx, } - tasks = append(tasks, task{entry: entry, subReq: subReq, outPath: itemOutputPath, isDir: true}) + wp.Submit(webdavTask{entry: entry, subReq: subReq, outPath: itemOutputPath, isDir: true}) } else { if req.Verbose { fmt.Fprintf(os.Stderr, "[webdav] Downloading file %s -> %s (depth=%d)\n", entry.URL.String(), itemOutputPath, req.MaxDepth) @@ -1126,71 +1164,11 @@ func (p *Protocol) downloadRecursiveParallel(ctx context.Context, req *core.Down MaxDepth: req.MaxDepth - 1, Ctx: req.Ctx, } - tasks = append(tasks, task{entry: entry, subReq: subReq, outPath: itemOutputPath, isDir: false}) + wp.Submit(webdavTask{entry: entry, subReq: subReq, outPath: itemOutputPath, isDir: false}) } } - // Worker pool: bounded by req.RecursiveParallel. We use a semaphore - // channel + WaitGroup so total in-flight workers never exceed the - // requested count, regardless of how many tasks are enqueued. - sem := make(chan struct{}, req.RecursiveParallel) - var wg sync.WaitGroup - - for _, t := range tasks { - // Honour context cancellation before launching a new task. - if ctx.Err() != nil { - break - } - - wg.Add(1) - sem <- struct{}{} - - go func(t task) { - defer wg.Done() - defer func() { <-sem }() - - // If the context was cancelled while we were waiting for - // the semaphore, just exit cleanly. - if ctx.Err() != nil { - return - } - - if t.isDir { - subResult, err := p.downloadRecursive(ctx, t.subReq) - if err != nil { - if req.Verbose { - fmt.Fprintf(os.Stderr, "[webdav] Warning: failed to %s sub-directory %s: %v\n", dryRunVerb(req.DryRun), t.entry.URL.String(), err) - } - return - } - statsMu.Lock() - totalBytes += subResult.BytesDownloaded - totalChunks += subResult.ChunksCount - statsMu.Unlock() - return - } - - subResult, err := p.downloadFile(ctx, t.subReq, t.entry.Size, t.entry.LastModified) - if err != nil { - if req.Verbose { - fmt.Fprintf(os.Stderr, "[webdav] Warning: failed to download file %s: %v\n", t.entry.URL.String(), err) - } - return - } - - // Apply keep-timestamps if requested. - if req.KeepTimestamps { - setFileTimestamp(t.outPath, t.entry.LastModified) - } - - statsMu.Lock() - totalBytes += subResult.BytesDownloaded - totalChunks++ - statsMu.Unlock() - }(t) - } - - wg.Wait() + wp.Wait() _ = processedDirs // reserved for future per-directory accounting