diff --git a/internal/protocol/bt/fetcher.go b/internal/protocol/bt/fetcher.go index e599df56..db1d8df4 100644 --- a/internal/protocol/bt/fetcher.go +++ b/internal/protocol/bt/fetcher.go @@ -33,6 +33,7 @@ type Fetcher struct { meta *fetcher.FetcherMeta torrentReady atomic.Bool + torrentDrop atomic.Bool create atomic.Bool progress fetcher.Progress } @@ -143,6 +144,7 @@ func (f *Fetcher) Pause() (err error) { } func (f *Fetcher) Close() (err error) { + f.torrentDrop.Store(false) f.safeDrop() return nil } @@ -158,6 +160,9 @@ func (f *Fetcher) safeDrop() { func (f *Fetcher) Wait() (err error) { for { + if f.torrentDrop.Load() { + break + } if f.torrentReady.Load() && len(f.meta.Opts.SelectFiles) > 0 { done := true for _, selectIndex := range f.meta.Opts.SelectFiles { diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index adce3b01..5d7faeb9 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -172,8 +172,8 @@ func (d *Downloader) Setup() error { if len(d.tasks) > 0 { for _, task := range d.tasks { func() { - task.lock.Lock() - defer task.lock.Unlock() + task.statusLock.Lock() + defer task.statusLock.Unlock() if task.Status != base.DownloadStatusRunning { return } @@ -329,8 +329,7 @@ func (d *Downloader) Pause(id string) (err error) { if task.Status == base.DownloadStatusPause { return } - err = d.doPause(task) - if err != nil { + if err = d.doPause(task); err != nil { return } d.notifyRunning() @@ -363,19 +362,7 @@ func (d *Downloader) Continue(id string) (err error) { } }() - func() { - task.lock.Lock() - defer task.lock.Unlock() - - if task.Status == base.DownloadStatusRunning { - return - } - - if err = d.doStart(task); err != nil { - return - } - }() - return + return d.doStart(task) } func (d *Downloader) PauseAll() (err error) { @@ -388,13 +375,7 @@ func (d *Downloader) PauseAll() (err error) { }() for _, task := range d.tasks { - err = func() error { - task.lock.Lock() - defer task.lock.Unlock() - - return d.doPause(task) - }() - if err != nil { + if err = d.doPause(task); err != nil { return } } @@ -424,13 +405,7 @@ func (d *Downloader) ContinueAll() (err error) { for _, task := range continuedTasks { tt := task - err = func() error { - tt.lock.Lock() - defer tt.lock.Unlock() - - return d.doStart(tt) - }() - if err != nil { + if err = d.doStart(tt); err != nil { return } } @@ -522,17 +497,21 @@ func (d *Downloader) Stats(id string) (sr any, err error) { func (d *Downloader) doDelete(task *Task, force bool) (err error) { err = func() error { - if task.fetcher != nil { - if err := task.fetcher.Close(); err != nil { - return err - } - } + d.lock.Lock() + defer d.lock.Unlock() + if err := d.storage.Delete(bucketTask, task.ID); err != nil { return err } if err := d.storage.Delete(bucketSave, task.ID); err != nil { return err } + + if task.fetcher != nil { + if err := task.fetcher.Close(); err != nil { + return err + } + } if force && task.Meta.Res != nil { if task.Meta.Res.Name != "" { if err := os.RemoveAll(task.Meta.FolderPath()); err != nil { @@ -794,31 +773,52 @@ func (d *Downloader) doCreate(fetcher fetcher.Fetcher, opts *base.Options) (task return } - d.doStart(task) + err = d.doStart(task) }() go d.watch(task) return } +func (d *Downloader) statusMut(task *Task, fn func() (bool, error)) (bool, error) { + task.statusLock.Lock() + defer task.statusLock.Unlock() + + return fn() +} + func (d *Downloader) doStart(task *Task) (err error) { - if task.Status != base.DownloadStatusRunning && task.Status != base.DownloadStatusDone { - err := d.restoreFetcher(task) + var isCreate bool + isReturn, err := d.statusMut(task, func() (isReturn bool, err error) { + if task.Status == base.DownloadStatusRunning || task.Status == base.DownloadStatusDone { + isReturn = true + return + } + + err = d.restoreFetcher(task) if err != nil { d.Logger.Error().Stack().Err(err).Msgf("restore fetcher failed, task id: %s", task.ID) - return err + return } - } - - isCreate := task.Status == base.DownloadStatusReady - - doStart := func() error { - task.lock.Lock() - defer task.lock.Unlock() + isCreate = task.Status == base.DownloadStatusReady d.triggerOnStart(task) task.updateStatus(base.DownloadStatusRunning) + return + }) + if err != nil { + d.Logger.Error().Stack().Err(err).Msgf("start task failed, task id: %s", task.ID) + return + } + if isReturn { + return + } + + handler := func() error { + task.lock.Lock() + defer task.lock.Unlock() + if task.Meta.Res == nil { err := task.fetcher.Resolve(task.Meta.Req) if err != nil { @@ -865,8 +865,7 @@ func (d *Downloader) doStart(task *Task) (err error) { return nil } go func() { - err := doStart() - if err != nil { + if err := handler(); err != nil { d.doOnError(task, err) } }() @@ -875,27 +874,44 @@ func (d *Downloader) doStart(task *Task) (err error) { } func (d *Downloader) doPause(task *Task) (err error) { - err = func() error { - if task.Status != base.DownloadStatusDone { - task.updateStatus(base.DownloadStatusPause) - task.timer.Pause() - if task.fetcher != nil { - if err := task.fetcher.Pause(); err != nil { - return err - } - } - if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil { - return err - } - d.emit(EventKeyPause, task) + isReturn, err := d.statusMut(task, func() (isReturn bool, err error) { + if task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusDone { + isReturn = true + return } - return nil - }() + + task.updateStatus(base.DownloadStatusPause) + task.timer.Pause() + return + }) if err != nil { d.Logger.Error().Stack().Err(err).Msgf("pause task failed, task id: %s", task.ID) return } + if isReturn { + return + } + handler := func() error { + task.lock.Lock() + defer task.lock.Unlock() + + if task.fetcher != nil { + if err := task.fetcher.Pause(); err != nil { + return err + } + } + if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil { + return err + } + d.emit(EventKeyPause, task) + return nil + } + go func() { + if err := handler(); err != nil { + d.Logger.Error().Stack().Err(err).Msgf("pause task handle failed, task id: %s", task.ID) + } + }() return } @@ -923,6 +939,8 @@ func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) { func initTask(task *Task) { task.timer = util.NewTimer(task.Progress.Used) + + task.statusLock = &sync.Mutex{} task.lock = &sync.Mutex{} task.speedArr = make([]int64, 0) } diff --git a/pkg/download/model.go b/pkg/download/model.go index 4d83a221..c78521ce 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -31,6 +31,7 @@ type Task struct { fetcherBuilder fetcher.FetcherBuilder fetcher fetcher.Fetcher timer *util.Timer + statusLock *sync.Mutex lock *sync.Mutex speedArr []int64 } diff --git a/ui/flutter/lib/api/model/options.dart b/ui/flutter/lib/api/model/options.dart index 52342059..fdce5ac4 100644 --- a/ui/flutter/lib/api/model/options.dart +++ b/ui/flutter/lib/api/model/options.dart @@ -12,7 +12,7 @@ class Options { Options({ required this.name, required this.path, - required this.selectFiles, + this.selectFiles = const [], this.extra, }); diff --git a/ui/flutter/lib/api/model/options.g.dart b/ui/flutter/lib/api/model/options.g.dart index 3d8394b9..b496e9fc 100644 --- a/ui/flutter/lib/api/model/options.g.dart +++ b/ui/flutter/lib/api/model/options.g.dart @@ -9,8 +9,10 @@ part of 'options.dart'; Options _$OptionsFromJson(Map json) => Options( name: json['name'] as String, path: json['path'] as String, - selectFiles: - (json['selectFiles'] as List).map((e) => e as int).toList(), + selectFiles: (json['selectFiles'] as List?) + ?.map((e) => e as int) + .toList() ?? + const [], extra: json['extra'], );