feat: support max running tasks config (#112)

pkg/base/constants.go

@@ -3,9 +3,10 @@ package base
 type Status string
 
 const (
-	DownloadStatusReady   Status = "ready"
+	DownloadStatusReady   Status = "ready" // task create but not start
 	DownloadStatusRunning Status = "running"
 	DownloadStatusPause   Status = "pause"
+	DownloadStatusWait    Status = "wait" // task is wait for running
 	DownloadStatusError   Status = "error"
 	DownloadStatusDone    Status = "done"
 )

pkg/download/downloader.go

@@ -9,6 +9,7 @@ import (
 	gonanoid "github.com/matoous/go-nanoid/v2"
 	"os"
 	"path"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -36,15 +37,15 @@ type Progress struct {
 }
 
 type Downloader struct {
+	cfg           *DownloaderConfig
 	fetchBuilders map[string]fetcher.FetcherBuilder
 	fetcherMap    map[string]fetcher.Fetcher
 	storage       Storage
 	tasks         []*Task
 	listener      Listener
 
-	refreshInterval int
-	lock            *sync.Mutex
-	closed          atomic.Bool
+	lock   *sync.Mutex
+	closed atomic.Bool
 }
 
 func NewDownloader(cfg *DownloaderConfig) *Downloader {
@@ -54,12 +55,13 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader {
 	cfg.Init()
 
 	d := &Downloader{
+		cfg: cfg,
+
 		fetchBuilders: make(map[string]fetcher.FetcherBuilder),
 		fetcherMap:    make(map[string]fetcher.Fetcher),
 
-		refreshInterval: cfg.RefreshInterval,
-		storage:         cfg.Storage,
-		lock:            &sync.Mutex{},
+		storage: cfg.Storage,
+		lock:    &sync.Mutex{},
 	}
 	for _, f := range cfg.FetchBuilders {
 		for _, p := range f.Schemes() {
@@ -74,6 +76,21 @@ func (d *Downloader) Setup() error {
 	if err := d.storage.Setup([]string{bucketTask, bucketSave, bucketConfig}); err != nil {
 		return err
 	}
+	// load config from storage
+	var cfg DownloaderStoreConfig
+	exist, err := d.storage.Get(bucketConfig, "config", &cfg)
+	if err != nil {
+		return err
+	}
+	if exist {
+		d.cfg.DownloaderStoreConfig = &cfg
+	} else {
+		d.cfg.DownloaderStoreConfig = &DownloaderStoreConfig{
+			FirstLoad: true,
+		}
+	}
+	// init default config
+	d.cfg.DownloaderStoreConfig.Init()
 	// load tasks from storage
 	var tasks []*Task
 	if err := d.storage.List(bucketTask, &tasks); err != nil {
@@ -94,6 +111,10 @@ func (d *Downloader) Setup() error {
 				task.Status = base.DownloadStatusPause
 			}
 		}
+		// sort by create time
+		sort.Slice(tasks, func(i, j int) bool {
+			return tasks[i].CreatedAt.Before(tasks[j].CreatedAt)
+		})
 	}
 	d.tasks = tasks
 
@@ -106,9 +127,7 @@ func (d *Downloader) Setup() error {
 					func() {
 						task.lock.Lock()
 						defer task.lock.Unlock()
-						if task.Status == base.DownloadStatusDone ||
-							task.Status == base.DownloadStatusError ||
-							task.Status == base.DownloadStatusPause {
+						if task.Status != base.DownloadStatusRunning {
 							return
 						}
 						// check if task is deleted
@@ -118,7 +137,7 @@ func (d *Downloader) Setup() error {
 
 						current := task.fetcher.Progress().TotalDownloaded()
 						task.Progress.Used = task.timer.Used()
-						task.Progress.Speed = task.calcSpeed(current-task.Progress.Downloaded, float64(d.refreshInterval)/1000)
+						task.Progress.Speed = task.calcSpeed(current-task.Progress.Downloaded, float64(d.cfg.RefreshInterval)/1000)
 						task.Progress.Downloaded = current
 						d.emit(EventKeyProgress, task)
 
@@ -136,7 +155,7 @@ func (d *Downloader) Setup() error {
 					}()
 				}
 			}
-			time.Sleep(time.Millisecond * time.Duration(d.refreshInterval))
+			time.Sleep(time.Millisecond * time.Duration(d.cfg.RefreshInterval))
 		}
 	}()
 	return nil
@@ -185,7 +204,7 @@ func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) {
 	return
 }
 
-func (d *Downloader) DirectCreate(req *base.Request, opts *base.Options) (taskId string, err error) {
+func (d *Downloader) CreateDirect(req *base.Request, opts *base.Options) (taskId string, err error) {
 	rr, err := d.Resolve(req)
 	if err != nil {
 		return
@@ -201,6 +220,8 @@ func (d *Downloader) Create(rrId string, opts *base.Options) (taskId string, err
 	if !ok {
 		return "", errors.New("invalid resource id")
 	}
+	defer delete(d.fetcherMap, rrId)
+
 	meta := fetcher.Meta()
 	meta.Opts = opts
 	res := meta.Res
@@ -211,13 +232,11 @@ func (d *Downloader) Create(rrId string, opts *base.Options) (taskId string, err
 		}
 	}
 	if opts.Path == "" {
-		exist, storeConfig, err := d.GetConfig()
+		storeConfig, err := d.GetConfig()
 		if err != nil {
 			return "", err
 		}
-		if exist {
-			opts.Path = storeConfig.DownloadDir
-		}
+		opts.Path = storeConfig.DownloadDir
 	}
 
 	fb, err := d.parseFb(fetcher.Meta().Req.URL)
@@ -253,36 +272,27 @@ func (d *Downloader) Create(rrId string, opts *base.Options) (taskId string, err
 	task.fetcher = fetcher
 	task.Meta = fetcher.Meta()
 	task.Progress = &Progress{}
-	task.Status = base.DownloadStatusRunning
+	task.Status = base.DownloadStatusReady
 	// calculate select files size
 	for _, selectIndex := range opts.SelectFiles {
 		task.Size += res.Files[selectIndex].Size
 	}
 	initTask(task)
-	task.timer.Start()
 	err = func() error {
 		d.lock.Lock()
 		defer d.lock.Unlock()
-		if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
-			return err
-		}
 		d.tasks = append(d.tasks, task)
-		return nil
+		return d.storage.Put(bucketTask, task.ID, task.clone())
 	}()
 	if err != nil {
 		return
 	}
-	d.emit(EventKeyStart, task)
 	taskId = task.ID
-	go func() {
-		err := fetcher.Start()
-		if err != nil {
-			d.emit(EventKeyError, task, err)
-			return
+	d.tryRunning(func(remain int) {
+		if remain > 0 {
+			err = d.start(task)
 		}
-		d.watch(task)
-	}()
-	delete(d.fetcherMap, rrId)
+	})
 	return
 }
 
@@ -291,16 +301,10 @@ func (d *Downloader) Pause(id string) (err error) {
 	if task == nil {
 		return
 	}
-	task.lock.Lock()
-	defer task.lock.Unlock()
-	if task.Status == base.DownloadStatusRunning {
-		task.Status = base.DownloadStatusPause
-		task.timer.Pause()
-		task.fetcher.Pause()
-		d.storage.Put(bucketTask, task.ID, task.clone())
-		d.emit(EventKeyPause, task)
+	if err = d.doPause(task, base.DownloadStatusPause); err != nil {
+		return err
 	}
-	return
+	return d.notifyRunning()
 }
 
 func (d *Downloader) Continue(id string) (err error) {
@@ -308,24 +312,24 @@ func (d *Downloader) Continue(id string) (err error) {
 	if task == nil {
 		return
 	}
-	task.lock.Lock()
-	defer task.lock.Unlock()
-	if task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusError {
-		task.Status = base.DownloadStatusRunning
-		task.Progress.Speed = 0
-		err = d.restoreFetcher(task)
+	d.tryRunning(func(remain int) {
+		if remain == 0 {
+			// no more task can be running, need to doPause a running task
+			for _, t := range d.tasks {
+				if t.Status == base.DownloadStatusRunning {
+					err = d.doPause(t, base.DownloadStatusWait)
+					break
+				}
+			}
+		}
 		if err != nil {
 			return
 		}
-		task.timer.Start()
-		err = task.fetcher.Continue()
-		if err != nil {
-			return
-		}
-		d.storage.Put(bucketTask, task.ID, task.clone())
-		d.emit(EventKeyContinue, task)
+	})
+	if err != nil {
+		return
 	}
-	return
+	return d.doContinue(task)
 }
 
 func (d *Downloader) Delete(id string, force bool) (err error) {
@@ -371,6 +375,11 @@ func (d *Downloader) Delete(id string, force bool) (err error) {
 		}
 	}
 	d.emit(EventKeyDelete, task)
+	if task.Status == base.DownloadStatusRunning {
+		if err = d.doNotifyRunning(); err != nil {
+			return err
+		}
+	}
 	task = nil
 	return
 }
@@ -380,7 +389,7 @@ func (d *Downloader) Close() error {
 	defer d.lock.Unlock()
 	d.closed.Store(true)
 	for _, task := range d.tasks {
-		d.Pause(task.ID)
+		d.doPause(task, base.DownloadStatusPause)
 	}
 	return d.storage.Close()
 }
@@ -426,27 +435,20 @@ func (d *Downloader) GetTasks() []*Task {
 	return d.tasks
 }
 
-func (d *Downloader) GetConfig() (bool, *DownloaderStoreConfig, error) {
-	var cfg DownloaderStoreConfig
-	exist, err := d.storage.Get(bucketConfig, "config", &cfg)
-	if err != nil {
-		return false, nil, err
-	}
-	return exist, &cfg, nil
+func (d *Downloader) GetConfig() (*DownloaderStoreConfig, error) {
+	return d.cfg.DownloaderStoreConfig, nil
 }
 
 func (d *Downloader) PutConfig(v *DownloaderStoreConfig) error {
+	d.cfg.DownloaderStoreConfig = v
 	return d.storage.Put(bucketConfig, "config", v)
 }
 
 func (d *Downloader) getProtocolConfig(name string, v any) (bool, error) {
-	exist, cfg, err := d.GetConfig()
+	cfg, err := d.GetConfig()
 	if err != nil {
 		return false, err
 	}
-	if !exist {
-		return false, nil
-	}
 	if cfg.ProtocolConfig == nil || cfg.ProtocolConfig[name] == nil {
 		return false, nil
 	}
@@ -479,6 +481,7 @@ func (d *Downloader) watch(task *Task) {
 		d.emit(EventKeyDone, task)
 	}
 	d.emit(EventKeyFinally, task, err)
+	d.notifyRunning()
 }
 
 func (d *Downloader) restoreFetcher(task *Task) error {
@@ -507,13 +510,96 @@ func (d *Downloader) restoreFetcher(task *Task) error {
 		}
 		d.setupFetcher(task.fetcher)
 		task.fetcher.Create(task.Meta.Opts)
-		go d.watch(task)
 	}
 	return nil
 }
 
+func (d *Downloader) tryRunning(cb func(remain int)) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	runningCount := 0
+	for _, task := range d.tasks {
+		if task.Status == base.DownloadStatusRunning {
+			runningCount++
+		}
+	}
+	cb(d.cfg.MaxRunning - runningCount)
+}
+
+func (d *Downloader) notifyRunning() error {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	return d.doNotifyRunning()
+}
+
+func (d *Downloader) doNotifyRunning() error {
+	for _, task := range d.tasks {
+		if task.Status == base.DownloadStatusReady || task.Status == base.DownloadStatusWait {
+			return d.doContinue(task)
+		}
+	}
+	return nil
+}
+
+func (d *Downloader) start(task *Task) error {
+	var event EventKey
+	if task.Status == base.DownloadStatusReady {
+		event = EventKeyStart
+	} else {
+		event = EventKeyContinue
+	}
+
+	task.Status = base.DownloadStatusRunning
+	task.Progress.Speed = 0
+	task.timer.Start()
+	if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
+		return err
+	}
+	var err error
+	if event == EventKeyStart {
+		err = task.fetcher.Start()
+	}
+	if event == EventKeyContinue {
+		err = task.fetcher.Continue()
+	}
+	if err != nil {
+		return err
+	}
+	go d.watch(task)
+	d.emit(event, task)
+	return nil
+}
+
+func (d *Downloader) doPause(task *Task, status base.Status) (err error) {
+	task.lock.Lock()
+	defer task.lock.Unlock()
+	if task.Status == base.DownloadStatusRunning {
+		task.Status = status
+		task.timer.Pause()
+		task.fetcher.Pause()
+		d.storage.Put(bucketTask, task.ID, task.clone())
+		d.emit(EventKeyPause, task)
+	}
+	return
+}
+
+func (d *Downloader) doContinue(task *Task) (err error) {
+	task.lock.Lock()
+	defer task.lock.Unlock()
+	if task.Status != base.DownloadStatusRunning && task.Status != base.DownloadStatusDone {
+		err = d.restoreFetcher(task)
+		if err != nil {
+			return
+		}
+		err = d.start(task)
+	}
+	return
+}
+
 func initTask(task *Task) {
-	task.timer = &util.Timer{}
+	task.timer = util.NewTimer(task.Progress.Used)
 	task.lock = &sync.Mutex{}
 	task.speedArr = make([]int64, 0)
 }

pkg/download/downloader_test.go

@@ -170,8 +170,7 @@ func TestDownloader_Protocol_Config(t *testing.T) {
 	}
 
 	storeCfg := &DownloaderStoreConfig{
-		RefreshInterval: 500,
-		DownloadDir:     "./downloads",
+		DownloadDir: "./downloads",
 		ProtocolConfig: map[string]any{
 			"http": map[string]any{
 				"connections": 4,
@@ -191,17 +190,14 @@ func TestDownloader_Protocol_Config(t *testing.T) {
 		},
 	}
 
-	if err := downloader.PutConfig(storeCfg); err != nil {
+	if err = downloader.PutConfig(storeCfg); err != nil {
 		t.Fatal(err)
 	}
 
-	exits, newStoreCfg, err := downloader.GetConfig()
+	newStoreCfg, err := downloader.GetConfig()
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !exits {
-		t.Errorf("GetConfig() got = %v, want %v", exits, true)
-	}
 
 	if !test.JsonEqual(storeCfg, newStoreCfg) {
 		t.Errorf("GetConfig() got = %v, want %v", test.ToJson(storeCfg), test.ToJson(newStoreCfg))

pkg/download/model.go

@@ -70,7 +70,9 @@ func (t *Task) calcSpeed(downloaded int64, usedTime float64) int64 {
 type DownloaderConfig struct {
 	Controller    *controller.Controller
 	FetchBuilders []fetcher.FetcherBuilder
-	Storage       Storage
+
+	RefreshInterval int `json:"refreshInterval"` // RefreshInterval time duration to refresh task progress(ms)
+	Storage         Storage
 
 	*DownloaderStoreConfig
 }
@@ -85,24 +87,28 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
 			new(bt.FetcherBuilder),
 		}
 	}
-	if cfg.Storage == nil {
-		cfg.Storage = NewMemStorage()
-	}
-
-	if cfg.DownloaderStoreConfig == nil {
-		cfg.DownloaderStoreConfig = &DownloaderStoreConfig{}
-	}
-
 	if cfg.RefreshInterval == 0 {
 		cfg.RefreshInterval = 350
 	}
+	if cfg.Storage == nil {
+		cfg.Storage = NewMemStorage()
+	}
 	return cfg
 }
 
 // DownloaderStoreConfig is the config that can restore the downloader.
 type DownloaderStoreConfig struct {
-	RefreshInterval int            `json:"refreshInterval"` // RefreshInterval time duration to refresh task progress(ms)
-	DownloadDir     string         `json:"downloadDir"`     // DownloadDir is the directory to save the downloaded files
-	ProtocolConfig  map[string]any `json:"protocolConfig"`  // ProtocolConfig is special config for each protocol
-	Extra           map[string]any `json:"extra"`           // Extra is the extra config
+	FirstLoad bool `json:"-"` // fromNoStore is the flag that the config is first time init and not from store
+
+	DownloadDir    string         `json:"downloadDir"`    // DownloadDir is the default directory to save the downloaded files
+	MaxRunning     int            `json:"maxRunning"`     // MaxRunning is the max running download count
+	ProtocolConfig map[string]any `json:"protocolConfig"` // ProtocolConfig is special config for each protocol
+	Extra          map[string]any `json:"extra"`          // Extra is the extra config
+}
+
+func (cfg *DownloaderStoreConfig) Init() *DownloaderStoreConfig {
+	if cfg.MaxRunning == 0 {
+		cfg.MaxRunning = 3
+	}
+	return cfg
 }

pkg/rest/api.go

@@ -33,7 +33,7 @@ func CreateTask(w http.ResponseWriter, r *http.Request) {
 		if req.Rid != "" {
 			taskId, err = Downloader.Create(req.Rid, req.Opts)
 		} else if req.Req != nil {
-			taskId, err = Downloader.DirectCreate(req.Req, req.Opts)
+			taskId, err = Downloader.CreateDirect(req.Req, req.Opts)
 		} else {
 			WriteJson(w, model.NewErrorResult("param invalid: rid or req", model.CodeInvalidParam))
 			return
@@ -195,6 +195,6 @@ func writeError(w http.ResponseWriter, msg string) {
 }
 
 func getServerConfig() *download.DownloaderStoreConfig {
-	_, cfg, _ := Downloader.GetConfig()
+	cfg, _ := Downloader.GetConfig()
 	return cfg
 }

pkg/rest/model/server.go

@@ -12,9 +12,9 @@ const (
 type StartConfig struct {
 	Network         string  `json:"network"`
 	Address         string  `json:"address"`
+	RefreshInterval int     `json:"refreshInterval"`
 	Storage         Storage `json:"storage"`
 	StorageDir      string  `json:"storageDir"`
-	RefreshInterval int     `json:"refreshInterval"`
 	ApiToken        string  `json:"apiToken"`
 
 	WebEnable bool
@@ -28,6 +28,9 @@ func (cfg *StartConfig) Init() *StartConfig {
 	if cfg.Address == "" {
 		cfg.Address = "127.0.0.1:0"
 	}
+	if cfg.RefreshInterval == 0 {
+		cfg.RefreshInterval = 350
+	}
 	if cfg.Storage == "" {
 		cfg.Storage = StorageBolt
 	}

pkg/rest/server.go

@@ -57,14 +57,15 @@ func BuildServer(startCfg *model.StartConfig) (*http.Server, net.Listener, error
 	}
 	startCfg.Init()
 
-	downloadCfg := &download.DownloaderConfig{}
+	downloadCfg := &download.DownloaderConfig{
+		RefreshInterval: startCfg.RefreshInterval,
+	}
 	if startCfg.Storage == model.StorageBolt {
 		downloadCfg.Storage = download.NewBoltStorage(startCfg.StorageDir)
 	} else {
 		downloadCfg.Storage = download.NewMemStorage()
 	}
 	downloadCfg.Init()
-	downloadCfg.RefreshInterval = startCfg.RefreshInterval
 	Downloader = download.NewDownloader(downloadCfg)
 	if err := Downloader.Setup(); err != nil {
 		return nil, nil, err

pkg/util/timer.go

@@ -2,12 +2,17 @@ package util
 
 import "time"
 
-// 计时器
 type Timer struct {
 	t    int64
 	used int64
 }
 
+func NewTimer(used int64) *Timer {
+	return &Timer{
+		used: used,
+	}
+}
+
 func (t *Timer) Start() {
 	t.t = time.Now().UnixNano()
 }