feat: bt improve

This commit is contained in:
levi committed 2022-10-26 17:18:13 +08:00
1 parent c6aa83194f
commit 0fc133e46d
28 files changed
+1537 -260

No files matched your search

+305 -69
View File
@@ -10,9 +10,16 @@ import (
"github.com/monkeyWie/gopeed-core/pkg/util"
"strings"
"sync"
"sync/atomic"
"time"
)
const (
bucketTask = "task"
bucketSave = "save"
bucketConfig = "config"
)
type Listener func(event *Event)
type Progress struct {
@@ -25,46 +32,133 @@ type Progress struct {
}
type Downloader struct {
*controller.DefaultController
controller *controller.DefaultController
fetchBuilders map[string]fetcher.FetcherBuilder
storage Storage
tasks []*Task
listener Listener
refreshInterval int
lock *sync.Mutex
closed atomic.Bool
}
func NewDownloader(fbs ...fetcher.FetcherBuilder) *Downloader {
d := &Downloader{DefaultController: controller.NewController()}
d.fetchBuilders = make(map[string]fetcher.FetcherBuilder)
for _, f := range fbs {
type DownloaderConfig struct {
Controller *controller.DefaultController
FetchBuilders []fetcher.FetcherBuilder
Storage Storage
// RefreshInterval time duration to refresh task progress(ms)
RefreshInterval int
}
func (cfg *DownloaderConfig) Init() *DownloaderConfig {
if cfg.Controller == nil {
cfg.Controller = controller.NewController()
}
if len(cfg.FetchBuilders) == 0 {
cfg.FetchBuilders = []fetcher.FetcherBuilder{
new(http.FetcherBuilder),
new(bt.FetcherBuilder),
}
}
if cfg.Storage == nil {
cfg.Storage = NewMemStorage()
}
if cfg.RefreshInterval == 0 {
cfg.RefreshInterval = 1000
}
return cfg
}
func NewDownloader(cfg *DownloaderConfig) *Downloader {
if cfg == nil {
cfg = &DownloaderConfig{}
}
cfg.Init()
d := &Downloader{
controller: cfg.Controller,
fetchBuilders: make(map[string]fetcher.FetcherBuilder),
refreshInterval: cfg.RefreshInterval,
storage: cfg.Storage,
lock: &sync.Mutex{},
}
for _, f := range cfg.FetchBuilders {
for _, p := range f.Schemes() {
d.fetchBuilders[strings.ToUpper(p)] = f
}
}
d.tasks = make([]*Task, 0)
// 每秒统计一次下载速度
go func() {
for {
if len(d.tasks) > 0 {
for _, task := range d.tasks {
if task.Status == base.DownloadStatusDone ||
task.Status == base.DownloadStatusError ||
task.Status == base.DownloadStatusPause {
continue
}
current := task.fetcher.Progress().TotalDownloaded()
task.Progress.Used = task.timer.Used()
task.Progress.Speed = current - task.Progress.Downloaded
task.Progress.Downloaded = current
d.emit(EventKeyProgress, task)
}
}
time.Sleep(time.Second)
}
}()
return d
}
func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) {
func (d *Downloader) Setup() error {
// setup storage
if err := d.storage.Setup([]string{bucketTask, bucketSave, bucketConfig}); err != nil {
return err
}
// load tasks from storage
var tasks []*Task
if err := d.storage.List(bucketTask, &tasks); err != nil {
return err
}
if tasks == nil {
tasks = make([]*Task, 0)
} else {
for _, task := range tasks {
initTask(task)
if task.Status != base.DownloadStatusDone && task.Status != base.DownloadStatusError {
task.Status = base.DownloadStatusPause
}
}
}
d.tasks = tasks
// 每秒统计一次下载速度
go func() {
for !d.closed.Load() {
if len(d.tasks) > 0 {
for _, task := range d.tasks {
func() {
task.lock.Lock()
defer task.lock.Unlock()
if task.Status == base.DownloadStatusDone ||
task.Status == base.DownloadStatusError ||
task.Status == base.DownloadStatusPause {
return
}
// check if task is deleted
if d.GetTask(task.ID) == nil {
return
}
current := task.fetcher.Progress().TotalDownloaded()
task.Progress.Used = task.timer.Used()
task.Progress.Speed = current - task.Progress.Downloaded
task.Progress.Downloaded = current
d.emit(EventKeyProgress, task)
// store fetcher progress
data, err := task.fetcherBuilder.Store(task.fetcher)
if err != nil {
return // TODO log
}
if err := d.storage.Put(bucketSave, task.ID, data); err != nil {
return // TODO log
}
if err := d.storage.Put(bucketTask, task.ID, task); err != nil {
return // TODO log
}
}()
}
}
time.Sleep(time.Millisecond * time.Duration(d.refreshInterval))
}
}()
return nil
}
func (d *Downloader) buildFetcher(url string) (fetcher.FetcherBuilder, fetcher.Fetcher, error) {
schema := util.ParseSchema(url)
fetchBuilder, ok := d.fetchBuilders[schema]
if !ok {
@@ -72,14 +166,14 @@ func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) {
}
if ok {
fetcher := fetchBuilder.Build()
fetcher.Setup(d.DefaultController)
return fetcher, nil
fetcher.Setup(d.controller)
return fetchBuilder, fetcher, nil
}
return nil, errors.New("unsupported protocol")
return nil, nil, errors.New("unsupported protocol")
}
func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) {
fetcher, err := d.buildFetcher(req.URL)
_, fetcher, err := d.buildFetcher(req.URL)
if err != nil {
return nil, err
}
@@ -99,13 +193,13 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) (taskId stri
opts.Connections = 1
}
if len(opts.SelectFiles) == 0 {
opts.SelectFiles = make([]int, len(res.Files))
for i := range opts.SelectFiles {
opts.SelectFiles[i] = i
opts.SelectFiles = make([]int, 0)
for i := range res.Files {
opts.SelectFiles = append(opts.SelectFiles, i)
}
}
fetcher, err := d.buildFetcher(res.Req.URL)
fetcherBuilder, fetcher, err := d.buildFetcher(res.Req.URL)
if err != nil {
return
}
@@ -115,74 +209,144 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) (taskId stri
}
task := NewTask()
task.fetcherBuilder = fetcherBuilder
task.fetcher = fetcher
task.Res = res
task.Opts = opts
task.Progress = &Progress{}
task.timer = &util.Timer{}
task.locker = new(sync.Mutex)
task.timer.Start()
task.Status = base.DownloadStatusRunning
d.tasks = append(d.tasks, task)
d.emit(EventKeyStart, task)
taskId = task.ID
err = fetcher.Start()
// 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
}()
if err != nil {
return
}
d.emit(EventKeyStart, task)
taskId = task.ID
go func() {
err = fetcher.Wait()
err := fetcher.Start()
if err != nil {
task.Status = base.DownloadStatusError
d.emit(EventKeyError, task, err)
} else {
task.Progress.Used = task.timer.Used()
if task.Res.Size == 0 {
task.Res.Size = task.fetcher.Progress().TotalDownloaded()
}
used := task.Progress.Used / int64(time.Second)
if used == 0 {
used = 1
}
task.Progress.Speed = task.Res.Size / used
task.Progress.Downloaded = task.Res.Size
task.Status = base.DownloadStatusDone
d.emit(EventKeyDone, task)
return
}
d.emit(EventKeyFinally, task, err)
d.watch(task)
}()
return
}
func (d *Downloader) Pause(id string) {
func (d *Downloader) Pause(id string) (err error) {
task := d.GetTask(id)
if task == nil {
return
}
task.locker.Lock()
defer task.locker.Unlock()
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)
}
return
}
func (d *Downloader) Continue(id string) {
func (d *Downloader) Continue(id string) (err error) {
task := d.GetTask(id)
if task == nil {
return
}
task.locker.Lock()
defer task.locker.Unlock()
task.lock.Lock()
defer task.lock.Unlock()
if task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusError {
task.Status = base.DownloadStatusRunning
task.timer.Continue()
task.fetcher.Continue()
err = d.restoreFetcher(task)
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)
}
return
}
func (d *Downloader) Delete(id string, force bool) (err error) {
d.lock.Lock()
defer d.lock.Unlock()
task := d.GetTask(id)
if task == nil {
return
}
task.lock.Lock()
defer task.lock.Unlock()
if task.fetcher != nil {
err = task.fetcher.Close()
if err != nil {
return
}
}
for i, t := range d.tasks {
if t.ID == id {
d.tasks = append(d.tasks[:i], d.tasks[i+1:]...)
break
}
}
err = d.storage.Delete(bucketTask, id)
if err != nil {
return
}
err = d.storage.Delete(bucketSave, id)
if err != nil {
return
}
if force {
names := make([]string, 0)
for _, file := range task.Res.Files {
names = append(names, util.Filepath(file.Path, file.Name, task.Opts.Name))
}
util.SafeRemoveAll(task.Opts.Path, names)
}
d.emit(EventKeyDelete, task)
task = nil
return
}
func (d *Downloader) Close() error {
d.lock.Lock()
defer d.lock.Unlock()
d.closed.Store(true)
for _, task := range d.tasks {
d.Pause(task.ID)
}
return d.storage.Close()
}
func (d *Downloader) Clear() error {
if err := d.Close(); err != nil {
return err
}
if err := d.storage.Clear(); err != nil {
return err
}
return nil
}
func (d *Downloader) Listener(fn Listener) {
@@ -216,7 +380,75 @@ func (d *Downloader) GetTasks() []*Task {
return d.tasks
}
var defaultDownloader = NewDownloader(new(http.FetcherBuilder), new(bt.FetcherBuilder))
func (d *Downloader) GetConfig(v any) error {
return d.storage.Get(bucketConfig, "config", v)
}
func (d *Downloader) PutConfig(v any) error {
return d.storage.Put(bucketConfig, "config", v)
}
func (d *Downloader) watch(task *Task) {
err := task.fetcher.Wait()
if err != nil {
task.Status = base.DownloadStatusError
d.emit(EventKeyError, task, err)
} else {
task.Progress.Used = task.timer.Used()
if task.Size == 0 {
task.Size = task.fetcher.Progress().TotalDownloaded()
task.Res.Size = task.Size
}
used := task.Progress.Used / int64(time.Second)
if used == 0 {
used = 1
}
task.Progress.Speed = task.Size / used
task.Progress.Downloaded = task.Size
task.Status = base.DownloadStatusDone
d.storage.Put(bucketTask, task.ID, task.clone())
d.emit(EventKeyDone, task)
}
d.emit(EventKeyFinally, task, err)
}
func (d *Downloader) restoreFetcher(task *Task) error {
if task.fetcher == nil {
fetcherBuilder, _, err := d.buildFetcher(task.Res.Req.URL)
if err != nil {
return err
}
task.fetcherBuilder = fetcherBuilder
err = func() error {
v, f := fetcherBuilder.Restore()
if v != nil {
err := d.storage.Pop(bucketSave, task.ID, v)
if err != nil {
return err
}
task.fetcher = f(task.Res, task.Opts, v)
}
return nil
}()
if err != nil {
// TODO log
}
if task.fetcher == nil {
task.fetcher = fetcherBuilder.Build()
task.fetcher.Create(task.Res, task.Opts)
}
task.fetcher.Setup(d.controller)
go d.watch(task)
}
return nil
}
func initTask(task *Task) {
task.timer = &util.Timer{}
task.lock = &sync.Mutex{}
}
var defaultDownloader = NewDownloader(nil)
type boot struct {
url string
@@ -256,5 +488,9 @@ func (b *boot) Create(opts *base.Options) (string, error) {
}
func Boot() *boot {
err := defaultDownloader.Setup()
if err != nil {
panic(err)
}
return &boot{}
}
+79 -4
View File
@@ -1,19 +1,23 @@
package download
import (
"github.com/monkeyWie/gopeed-core/internal/protocol/http"
"github.com/monkeyWie/gopeed-core/internal/test"
"github.com/monkeyWie/gopeed-core/pkg/base"
"github.com/monkeyWie/gopeed-core/pkg/test"
"reflect"
"sync"
"testing"
"time"
)
func TestDownloader_Resolve(t *testing.T) {
listener := test.StartTestFileServer()
defer listener.Close()
downloader := NewDownloader(new(http.FetcherBuilder))
downloader := NewDownloader(nil)
if err := downloader.Setup(); err != nil {
t.Fatal(err)
}
defer downloader.Clear()
req := &base.Request{
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
}
@@ -22,6 +26,7 @@ func TestDownloader_Resolve(t *testing.T) {
t.Fatal(err)
}
want := &base.Resource{
Name: test.BuildName,
Req: req,
Size: test.BuildSize,
Range: true,
@@ -42,7 +47,11 @@ func TestDownloader_Create(t *testing.T) {
listener := test.StartTestFileServer()
defer listener.Close()
downloader := NewDownloader(new(http.FetcherBuilder))
downloader := NewDownloader(nil)
if err := downloader.Setup(); err != nil {
t.Fatal(err)
}
defer downloader.Clear()
req := &base.Request{
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
}
@@ -73,3 +82,69 @@ func TestDownloader_Create(t *testing.T) {
t.Errorf("Download() got = %v, want %v", got, want)
}
}
func TestDownloader_StoreAndRestore(t *testing.T) {
listener := test.StartTestSlowFileServer(time.Millisecond * 2000)
defer listener.Close()
downloader := NewDownloader(&DownloaderConfig{
Storage: NewBoltStorage("./"),
})
if err := downloader.Setup(); err != nil {
t.Fatal(err)
}
defer downloader.Clear()
req := &base.Request{
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
}
res, err := downloader.Resolve(req)
if err != nil {
t.Fatal(err)
}
id, err := downloader.Create(res, &base.Options{
Path: test.Dir,
Name: test.DownloadName,
Connections: 4,
})
if err != nil {
t.Fatal(err)
}
time.Sleep(time.Millisecond * 1001)
err = downloader.Pause(id)
if err != nil {
t.Fatal(err)
}
downloader.Close()
downloader = NewDownloader(&DownloaderConfig{
Storage: NewBoltStorage("./"),
})
if err := downloader.Setup(); err != nil {
t.Fatal(err)
}
task := downloader.GetTask(id)
if task == nil {
t.Fatal("task is nil")
}
var wg sync.WaitGroup
wg.Add(1)
downloader.Listener(func(event *Event) {
if event.Key == EventKeyDone {
wg.Done()
}
})
err = downloader.Continue(id)
wg.Wait()
if err != nil {
t.Fatal(err)
}
want := test.FileMd5(test.BuildFile)
got := test.FileMd5(test.DownloadFile)
if want != got {
t.Errorf("StoreAndResume() got = %v, want %v", got, want)
}
downloader.Clear()
}
+1
View File
@@ -8,6 +8,7 @@ const (
EventKeyContinue = "continue"
EventKeyProgress = "progress"
EventKeyError = "error"
EventKeyDelete = "delete"
EventKeyDone = "done"
EventKeyFinally = "finally"
)
+219
View File
@@ -0,0 +1,219 @@
package download
import (
"encoding/json"
"go.etcd.io/bbolt"
"os"
"path/filepath"
"reflect"
)
type Storage interface {
Setup(buckets []string) error
Put(bucket string, key string, v any) error
Get(bucket string, key string, v any) error
List(bucket string, v any) error
Pop(bucket string, key string, v any) error
Delete(bucket string, key string) error
Close() error
Clear() error
}
func changeValue(p any, v any) {
if v == nil {
return
}
rp := reflect.ValueOf(p)
rv := reflect.ValueOf(v)
if rv.Kind() == reflect.Slice {
if rv.Len() == 0 {
return
}
// get underlying type
tp := reflect.TypeOf(p).Elem().Elem()
for i := 0; i < rv.Len(); i++ {
// convert to underlying type
vv := rv.Index(i).Elem().Convert(tp)
rp.Elem().Set(reflect.Append(rp.Elem(), vv))
}
} else if rv.Kind() == reflect.Ptr {
rp.Elem().Set(rv.Elem())
} else {
rp.Elem().Set(rv)
}
}
var memData = make(map[string]map[string]any)
type MemStorage struct {
}
func NewMemStorage() *MemStorage {
return &MemStorage{}
}
func (n *MemStorage) Setup(buckets []string) error {
for _, bucket := range buckets {
if _, ok := memData[bucket]; !ok {
memData[bucket] = make(map[string]any)
}
}
return nil
}
func (n *MemStorage) Put(bucket string, key string, v any) error {
memData[bucket][key] = v
return nil
}
func (n *MemStorage) Get(bucket string, key string, v any) error {
changeValue(v, memData[bucket][key])
return nil
}
func (n *MemStorage) List(bucket string, v any) error {
data := memData[bucket]
list := make([]any, 0)
for _, v := range data {
list = append(list, v)
}
changeValue(v, list)
return nil
}
func (n *MemStorage) Pop(bucket string, key string, v any) error {
data := memData[bucket]
changeValue(v, data[key])
delete(data, key)
return nil
}
func (n *MemStorage) Delete(bucket string, key string) error {
delete(memData[bucket], key)
return nil
}
func (n *MemStorage) Close() error {
return nil
}
func (n *MemStorage) Clear() error {
memData = make(map[string]map[string]any)
return nil
}
const (
dbFile = "gopeed.db"
)
type BoltStorage struct {
db *bbolt.DB
path string
}
func NewBoltStorage(dir string) *BoltStorage {
path := filepath.Join(dir, dbFile)
db, err := bbolt.Open(path, 0600, nil)
if err != nil {
panic(err)
}
return &BoltStorage{
db: db,
path: path,
}
}
func (b *BoltStorage) Setup(buckets []string) error {
return b.db.Update(func(tx *bbolt.Tx) error {
for _, bucket := range buckets {
_, err := tx.CreateBucketIfNotExists([]byte(bucket))
if err != nil {
return err
}
}
return nil
})
}
func (b *BoltStorage) Put(bucket string, key string, v any) error {
buf, err := json.Marshal(v)
if err != nil {
return err
}
return b.db.Update(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(bucket))
return b.Put([]byte(key), buf)
})
}
func (b *BoltStorage) Get(bucket string, key string, v any) error {
var data []byte
err := b.db.View(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(bucket))
data = b.Get([]byte(key))
return nil
})
if err != nil {
return err
}
return json.Unmarshal(data, v)
}
func (b *BoltStorage) List(bucket string, v any) error {
list := make([]any, 0)
tv := reflect.TypeOf(v).Elem().Elem()
if err := b.db.View(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(bucket))
if err := b.ForEach(func(k, v []byte) error {
data := reflect.New(tv.Elem()).Interface()
if err := json.Unmarshal(v, &data); err != nil {
return err
}
list = append(list, data)
return nil
}); err != nil {
return err
}
return nil
}); err != nil {
return err
}
changeValue(v, list)
return nil
}
func (b *BoltStorage) Pop(bucket string, key string, v any) error {
var data []byte
err := b.db.Update(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(bucket))
kb := []byte(key)
data = b.Get(kb)
return b.Delete(kb)
})
if err != nil {
return err
}
return json.Unmarshal(data, v)
}
func (b *BoltStorage) Delete(bucket string, key string) error {
return b.db.Update(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(bucket))
return b.Delete([]byte(key))
})
}
func (b *BoltStorage) Close() error {
return b.db.Close()
}
func (b *BoltStorage) Clear() error {
if err := b.Close(); err != nil {
return err
}
if err := os.Remove(b.path); err != nil {
return err
}
return nil
}
+31 -14
View File
@@ -1,7 +1,7 @@
package download
import (
"github.com/google/uuid"
gonanoid "github.com/matoous/go-nanoid/v2"
"github.com/monkeyWie/gopeed-core/internal/fetcher"
"github.com/monkeyWie/gopeed-core/pkg/base"
"github.com/monkeyWie/gopeed-core/pkg/util"
@@ -10,23 +10,40 @@ import (
)
type Task struct {
ID string `json:"id"`
Res *base.Resource `json:"res"`
Opts *base.Options `json:"opts"`
Status base.Status `json:"status"`
Files map[string]*base.FileInfo `json:"files"`
Progress *Progress `json:"progress"`
CreateTime time.Time `json:"create_time"`
ID string `json:"id"`
Res *base.Resource `json:"res"`
Opts *base.Options `json:"opts"`
Status base.Status `json:"status"`
Progress *Progress `json:"progress"`
Size int64 `json:"size"`
CreatedAt time.Time `json:"createdAt"`
fetcher fetcher.Fetcher
timer *util.Timer
locker *sync.Mutex
fetcherBuilder fetcher.FetcherBuilder
fetcher fetcher.Fetcher
timer *util.Timer
lock *sync.Mutex
}
func NewTask() *Task {
id, err := gonanoid.New()
if err != nil {
panic(err)
}
return &Task{
ID: uuid.New().String(),
Status: base.DownloadStatusReady,
CreateTime: time.Now(),
ID: id,
Status: base.DownloadStatusReady,
CreatedAt: time.Now(),
}
}
func (t *Task) clone() *Task {
return &Task{
ID: t.ID,
Res: t.Res,
Opts: t.Opts,
Status: t.Status,
Progress: t.Progress,
Size: t.Size,
CreatedAt: t.CreatedAt,
}
}