github
/
alist
mirror of https://github.com/Xhofe/alist
1
0
Fork 0
Code Issues Releases Wiki Activity

refactor(task): remove `Data` field

refactor/fs
Noah Hsu 2022-06-22 19:28:41 +08:00
parent a6df492fff
commit 3fe0a7bf6b
8 changed files with 55 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/aria2/add.go
View File

@ -45,10 +45,10 @@ func AddURI(ctx context.Context, uri string, dstDirPath string) error {
return errors.Wrapf(err, "failed to add uri %s", uri) return errors.Wrapf(err, "failed to add uri %s", uri)
} }
// TODO add to task manager // TODO add to task manager
TaskManager.Submit(task.WithCancelCtx(&task.Task[string, interface{}]{ TaskManager.Submit(task.WithCancelCtx(&task.Task[string]{
ID: gid, ID: gid,
Name: fmt.Sprintf("download %s to [%s](%s)", uri, account.GetAccount().VirtualPath, dstDirActualPath), Name: fmt.Sprintf("download %s to [%s](%s)", uri, account.GetAccount().VirtualPath, dstDirActualPath),
Func: func(tsk *task.Task[string, interface{}]) error { Func: func(tsk *task.Task[string]) error {
m := &Monitor{ m := &Monitor{
tsk: tsk, tsk: tsk,
tempDir: tempDir, tempDir: tempDir,

2
internal/aria2/aria2.go
View File

@ -8,7 +8,7 @@ import (
"time" "time"
) )
var TaskManager = task.NewTaskManager[string, interface{}](3) var TaskManager = task.NewTaskManager[string](3)
var notify = NewNotify() var notify = NewNotify()
var client rpc.Client var client rpc.Client

8
internal/aria2/monitor.go
View File

@ -17,7 +17,7 @@ import (
) )
type Monitor struct { type Monitor struct {
tsk *task.Task[string, interface{}] tsk *task.Task[string]
tempDir string tempDir string
retried int retried int
c chan int c chan int
@ -92,7 +92,7 @@ func (m *Monitor) Update() (bool, error) {
} }
} }
var transferTaskManager = task.NewTaskManager[uint64, interface{}](3, func(k *uint64) { var transferTaskManager = task.NewTaskManager[uint64](3, func(k *uint64) {
atomic.AddUint64(k, 1) atomic.AddUint64(k, 1)
}) })
@ -118,9 +118,9 @@ func (m *Monitor) Complete() error {
} }
}() }()
for _, file := range files { for _, file := range files {
transferTaskManager.Submit(task.WithCancelCtx[uint64](&task.Task[uint64, interface{}]{ transferTaskManager.Submit(task.WithCancelCtx[uint64](&task.Task[uint64]{
Name: fmt.Sprintf("transfer %s to %s", file.Path, m.dstDirPath), Name: fmt.Sprintf("transfer %s to %s", file.Path, m.dstDirPath),
Func: func(tsk *task.Task[uint64, interface{}]) error { Func: func(tsk *task.Task[uint64]) error {
defer wg.Done() defer wg.Done()
size, _ := strconv.ParseUint(file.Length, 10, 64) size, _ := strconv.ParseUint(file.Length, 10, 64)
mimetype := mime.TypeByExtension(path.Ext(file.Path)) mimetype := mime.TypeByExtension(path.Ext(file.Path))

18
internal/fs/copy.go
View File

@ -15,7 +15,7 @@ import (
"github.com/pkg/errors" "github.com/pkg/errors"
) )
var CopyTaskManager = task.NewTaskManager[uint64, struct{}](3, func(tid *uint64) { var CopyTaskManager = task.NewTaskManager[uint64](3, func(tid *uint64) {
atomic.AddUint64(tid, 1) atomic.AddUint64(tid, 1)
}) })
@ -35,16 +35,16 @@ func Copy(ctx context.Context, account driver.Driver, srcObjPath, dstDirPath str
return false, operations.Copy(ctx, account, srcObjActualPath, dstDirActualPath) return false, operations.Copy(ctx, account, srcObjActualPath, dstDirActualPath)
} }
// not in an account // not in an account
CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64, struct{}]{ CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjActualPath, dstAccount.GetAccount().VirtualPath, dstDirActualPath), Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjActualPath, dstAccount.GetAccount().VirtualPath, dstDirActualPath),
Func: func(task *task.Task[uint64, struct{}]) error { Func: func(task *task.Task[uint64]) error {
return CopyBetween2Accounts(task, srcAccount, dstAccount, srcObjActualPath, dstDirActualPath) return CopyBetween2Accounts(task, srcAccount, dstAccount, srcObjActualPath, dstDirActualPath)
}, },
})) }))
return true, nil return true, nil
} }
func CopyBetween2Accounts(t *task.Task[uint64, struct{}], srcAccount, dstAccount driver.Driver, srcObjPath, dstDirPath string) error { func CopyBetween2Accounts(t *task.Task[uint64], srcAccount, dstAccount driver.Driver, srcObjPath, dstDirPath string) error {
t.SetStatus("getting src object") t.SetStatus("getting src object")
srcObj, err := operations.Get(t.Ctx, srcAccount, srcObjPath) srcObj, err := operations.Get(t.Ctx, srcAccount, srcObjPath)
if err != nil { if err != nil {
@ -62,17 +62,17 @@ func CopyBetween2Accounts(t *task.Task[uint64, struct{}], srcAccount, dstAccount
} }
srcObjPath := stdpath.Join(srcObjPath, obj.GetName()) srcObjPath := stdpath.Join(srcObjPath, obj.GetName())
dstObjPath := stdpath.Join(dstDirPath, obj.GetName()) dstObjPath := stdpath.Join(dstDirPath, obj.GetName())
CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64, struct{}]{ CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjPath, dstAccount.GetAccount().VirtualPath, dstObjPath), Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjPath, dstAccount.GetAccount().VirtualPath, dstObjPath),
Func: func(t *task.Task[uint64, struct{}]) error { Func: func(t *task.Task[uint64]) error {
return CopyBetween2Accounts(t, srcAccount, dstAccount, srcObjPath, dstObjPath) return CopyBetween2Accounts(t, srcAccount, dstAccount, srcObjPath, dstObjPath)
}, },
})) }))
} }
} else { } else {
CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64, struct{}]{ CopyTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjPath, dstAccount.GetAccount().VirtualPath, dstDirPath), Name: fmt.Sprintf("copy [%s](%s) to [%s](%s)", srcAccount.GetAccount().VirtualPath, srcObjPath, dstAccount.GetAccount().VirtualPath, dstDirPath),
Func: func(t *task.Task[uint64, struct{}]) error { Func: func(t *task.Task[uint64]) error {
return CopyFileBetween2Accounts(t, srcAccount, dstAccount, srcObjPath, dstDirPath) return CopyFileBetween2Accounts(t, srcAccount, dstAccount, srcObjPath, dstDirPath)
}, },
})) }))
@ -80,7 +80,7 @@ func CopyBetween2Accounts(t *task.Task[uint64, struct{}], srcAccount, dstAccount
return nil return nil
} }
func CopyFileBetween2Accounts(tsk *task.Task[uint64, struct{}], srcAccount, dstAccount driver.Driver, srcFilePath, dstDirPath string) error { func CopyFileBetween2Accounts(tsk *task.Task[uint64], srcAccount, dstAccount driver.Driver, srcFilePath, dstDirPath string) error {
srcFile, err := operations.Get(tsk.Ctx, srcAccount, srcFilePath) srcFile, err := operations.Get(tsk.Ctx, srcAccount, srcFilePath)
if err != nil { if err != nil {
return errors.WithMessagef(err, "failed get src [%s] file", srcFilePath) return errors.WithMessagef(err, "failed get src [%s] file", srcFilePath)

6
internal/fs/put.go
View File

@ -11,7 +11,7 @@ import (
"sync/atomic" "sync/atomic"
) )
var UploadTaskManager = task.NewTaskManager[uint64, struct{}](3, func(tid *uint64) { var UploadTaskManager = task.NewTaskManager[uint64](3, func(tid *uint64) {
atomic.AddUint64(tid, 1) atomic.AddUint64(tid, 1)
}) })
@ -24,9 +24,9 @@ func Put(ctx context.Context, account driver.Driver, dstDirPath string, file mod
if err != nil { if err != nil {
return errors.WithMessage(err, "failed get account") return errors.WithMessage(err, "failed get account")
} }
UploadTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64, struct{}]{ UploadTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
Name: fmt.Sprintf("upload %s to [%s](%s)", file.GetName(), account.GetAccount().VirtualPath, dstDirActualPath), Name: fmt.Sprintf("upload %s to [%s](%s)", file.GetName(), account.GetAccount().VirtualPath, dstDirActualPath),
Func: func(task *task.Task[uint64, struct{}]) error { Func: func(task *task.Task[uint64]) error {
return operations.Put(task.Ctx, account, dstDirActualPath, file, nil) return operations.Put(task.Ctx, account, dstDirActualPath, file, nil)
}, },
})) }))

32
pkg/task/manager.go
View File

@ -5,14 +5,14 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
type Manager[K comparable, V any] struct { type Manager[K comparable] struct {
workerC chan struct{} workerC chan struct{}
curID K curID K
updateID func(*K) updateID func(*K)
tasks generic_sync.MapOf[K, *Task[K, V]] tasks generic_sync.MapOf[K, *Task[K]]
} }
func (tm *Manager[K, V]) Submit(task *Task[K, V]) K { func (tm *Manager[K]) Submit(task *Task[K]) K {
if tm.updateID != nil { if tm.updateID != nil {
task.ID = tm.curID task.ID = tm.curID
tm.updateID(&task.ID) tm.updateID(&task.ID)
@ -22,7 +22,7 @@ func (tm *Manager[K, V]) Submit(task *Task[K, V]) K {
return task.ID return task.ID
} }
func (tm *Manager[K, V]) do(task *Task[K, V]) { func (tm *Manager[K]) do(task *Task[K]) {
go func() { go func() {
log.Debugf("task [%s] waiting for worker", task.Name) log.Debugf("task [%s] waiting for worker", task.Name)
select { select {
@ -36,20 +36,20 @@ func (tm *Manager[K, V]) do(task *Task[K, V]) {
}() }()
} }
func (tm *Manager[K, V]) GetAll() []*Task[K, V] { func (tm *Manager[K]) GetAll() []*Task[K] {
return tm.tasks.Values() return tm.tasks.Values()
} }
func (tm *Manager[K, V]) Get(tid K) (*Task[K, V], bool) { func (tm *Manager[K]) Get(tid K) (*Task[K], bool) {
return tm.tasks.Load(tid) return tm.tasks.Load(tid)
} }
func (tm *Manager[K, V]) MustGet(tid K) *Task[K, V] { func (tm *Manager[K]) MustGet(tid K) *Task[K] {
task, _ := tm.Get(tid) task, _ := tm.Get(tid)
return task return task
} }
func (tm *Manager[K, V]) Retry(tid K) error { func (tm *Manager[K]) Retry(tid K) error {
t, ok := tm.Get(tid) t, ok := tm.Get(tid)
if !ok { if !ok {
return ErrTaskNotFound return ErrTaskNotFound
@ -58,7 +58,7 @@ func (tm *Manager[K, V]) Retry(tid K) error {
return nil return nil
} }
func (tm *Manager[K, V]) Cancel(tid K) error { func (tm *Manager[K]) Cancel(tid K) error {
t, ok := tm.Get(tid) t, ok := tm.Get(tid)
if !ok { if !ok {
return ErrTaskNotFound return ErrTaskNotFound
@ -67,17 +67,17 @@ func (tm *Manager[K, V]) Cancel(tid K) error {
return nil return nil
} }
func (tm *Manager[K, V]) Remove(tid K) { func (tm *Manager[K]) Remove(tid K) {
tm.tasks.Delete(tid) tm.tasks.Delete(tid)
} }
// RemoveAll removes all tasks from the manager, this maybe shouldn't be used // RemoveAll removes all tasks from the manager, this maybe shouldn't be used
// because the task maybe still running. // because the task maybe still running.
func (tm *Manager[K, V]) RemoveAll() { func (tm *Manager[K]) RemoveAll() {
tm.tasks.Clear() tm.tasks.Clear()
} }
func (tm *Manager[K, V]) RemoveFinished() { func (tm *Manager[K]) RemoveFinished() {
tasks := tm.GetAll() tasks := tm.GetAll()
for _, task := range tasks { for _, task := range tasks {
if task.Status == FINISHED { if task.Status == FINISHED {
@ -86,7 +86,7 @@ func (tm *Manager[K, V]) RemoveFinished() {
} }
} }
func (tm *Manager[K, V]) RemoveError() { func (tm *Manager[K]) RemoveError() {
tasks := tm.GetAll() tasks := tm.GetAll()
for _, task := range tasks { for _, task := range tasks {
if task.Error != nil { if task.Error != nil {
@ -95,9 +95,9 @@ func (tm *Manager[K, V]) RemoveError() {
} }
} }
func NewTaskManager[K comparable, V any](maxWorker int, updateID ...func(*K)) *Manager[K, V] { func NewTaskManager[K comparable](maxWorker int, updateID ...func(*K)) *Manager[K] {
tm := &Manager[K, V]{ tm := &Manager[K]{
tasks: generic_sync.MapOf[K, *Task[K, V]]{}, tasks: generic_sync.MapOf[K, *Task[K]]{},
workerC: make(chan struct{}, maxWorker), workerC: make(chan struct{}, maxWorker),
} }
for i := 0; i < maxWorker; i++ { for i := 0; i < maxWorker; i++ {

24
pkg/task/task.go
View File

@ -16,34 +16,32 @@ var (
ERRORED = "errored" ERRORED = "errored"
) )
type Func[K comparable, V any] func(task *Task[K, V]) error type Func[K comparable] func(task *Task[K]) error
type Callback[K comparable, V any] func(task *Task[K, V]) type Callback[K comparable] func(task *Task[K])
type Task[K comparable, V any] struct { type Task[K comparable] struct {
ID K ID K
Name string Name string
Status string Status string
Error error Error error
Data V Func Func[K]
callback Callback[K]
Func Func[K, V]
callback Callback[K, V]
Ctx context.Context Ctx context.Context
progress int progress int
cancel context.CancelFunc cancel context.CancelFunc
} }
func (t *Task[K, V]) SetStatus(status string) { func (t *Task[K]) SetStatus(status string) {
t.Status = status t.Status = status
} }
func (t *Task[K, V]) SetProgress(percentage int) { func (t *Task[K]) SetProgress(percentage int) {
t.progress = percentage t.progress = percentage
} }
func (t *Task[K, V]) run() { func (t *Task[K]) run() {
t.Status = RUNNING t.Status = RUNNING
defer func() { defer func() {
if err := recover(); err != nil { if err := recover(); err != nil {
@ -68,11 +66,11 @@ func (t *Task[K, V]) run() {
} }
} }
func (t *Task[K, V]) retry() { func (t *Task[K]) retry() {
t.run() t.run()
} }
func (t *Task[K, V]) Cancel() { func (t *Task[K]) Cancel() {
if t.Status == FINISHED || t.Status == CANCELED { if t.Status == FINISHED || t.Status == CANCELED {
return return
} }
@ -83,7 +81,7 @@ func (t *Task[K, V]) Cancel() {
t.Status = CANCELING t.Status = CANCELING
} }
func WithCancelCtx[K comparable, V any](task *Task[K, V]) *Task[K, V] { func WithCancelCtx[K comparable](task *Task[K]) *Task[K] {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
task.Ctx = ctx task.Ctx = ctx
task.cancel = cancel task.cancel = cancel

18
pkg/task/task_test.go
View File

@ -9,12 +9,12 @@ import (
) )
func TestTask_Manager(t *testing.T) { func TestTask_Manager(t *testing.T) {
tm := NewTaskManager[uint64, struct{}](3, func(id *uint64) { tm := NewTaskManager[uint64](3, func(id *uint64) {
atomic.AddUint64(id, 1) atomic.AddUint64(id, 1)
}) })
id := tm.Submit(WithCancelCtx(&Task[uint64, struct{}]{ id := tm.Submit(WithCancelCtx(&Task[uint64]{
Name: "test", Name: "test",
Func: func(task *Task[uint64, struct{}]) error { Func: func(task *Task[uint64]) error {
time.Sleep(time.Millisecond * 500) time.Sleep(time.Millisecond * 500)
return nil return nil
}, },
@ -34,12 +34,12 @@ func TestTask_Manager(t *testing.T) {
} }
func TestTask_Cancel(t *testing.T) { func TestTask_Cancel(t *testing.T) {
tm := NewTaskManager[uint64, struct{}](3, func(id *uint64) { tm := NewTaskManager[uint64](3, func(id *uint64) {
atomic.AddUint64(id, 1) atomic.AddUint64(id, 1)
}) })
id := tm.Submit(WithCancelCtx(&Task[uint64, struct{}]{ id := tm.Submit(WithCancelCtx(&Task[uint64]{
Name: "test", Name: "test",
Func: func(task *Task[uint64, struct{}]) error { Func: func(task *Task[uint64]) error {
for { for {
if utils.IsCanceled(task.Ctx) { if utils.IsCanceled(task.Ctx) {
return nil return nil
@ -62,13 +62,13 @@ func TestTask_Cancel(t *testing.T) {
} }
func TestTask_Retry(t *testing.T) { func TestTask_Retry(t *testing.T) {
tm := NewTaskManager[uint64, struct{}](3, func(id *uint64) { tm := NewTaskManager[uint64](3, func(id *uint64) {
atomic.AddUint64(id, 1) atomic.AddUint64(id, 1)
}) })
num := 0 num := 0
id := tm.Submit(WithCancelCtx(&Task[uint64, struct{}]{ id := tm.Submit(WithCancelCtx(&Task[uint64]{
Name: "test", Name: "test",
Func: func(task *Task[uint64, struct{}]) error { Func: func(task *Task[uint64]) error {
num++ num++
if num&1 == 1 { if num&1 == 1 {
return errors.New("test error") return errors.New("test error")