From b34c723bca05d93ee641f7f847d9a0a605cf391e Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 20:16:55 +0800 Subject: [PATCH 01/10] fix(fs):After the file is copied, the cache of the copied directory is refreshed --- internal/fs/copy.go | 141 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 138 insertions(+), 3 deletions(-) diff --git a/internal/fs/copy.go b/internal/fs/copy.go index bd8be1f29..f264b33a4 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" stdpath "path" + "sync" "time" "github.com/OpenListTeam/OpenList/v4/internal/conf" @@ -46,6 +47,13 @@ func (t *CopyTask) Run() error { t.ClearEndTime() t.SetStartTime(time.Now()) defer func() { t.SetEndTime(time.Now()) }() + + // 注册任务到批量跟踪器 + taskID := t.GetID() + if taskID == "" { + taskID = utils.RandomString(16) + } + var err error if t.srcStorage == nil { t.srcStorage, err = op.GetStorageByMountPath(t.SrcStorageMp) @@ -56,11 +64,128 @@ func (t *CopyTask) Run() error { if err != nil { return errors.WithMessage(err, "failed get storage") } - return copyBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath) + + // 注册任务到批量跟踪器 + batchTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) + + // 执行复制操作 + err = copyBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath) + + // 标记任务完成并检查是否需要刷新缓存 + if err == nil { + shouldRefresh, dstStorage, dstDirPath := batchTracker.markTaskCompleted(taskID) + if shouldRefresh { + op.ClearCache(dstStorage, dstDirPath) + } + } else { + // 即使失败也要清理跟踪器中的记录 + batchTracker.markTaskCompleted(taskID) + } + + return err } var CopyTaskManager *tache.Manager[*CopyTask] +// 批量复制任务跟踪器 - 按目标目录聚合所有复制任务 +type batchCopyTracker struct { + mu sync.Mutex + dirTasks map[string]*dirTaskInfo // dstStoragePath+dstDirPath -> dirTaskInfo + pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath +} + +type dirTaskInfo struct { + dstStorage driver.Driver + dstDirPath string + pendingTasks map[string]bool // taskID -> true + lastActivity time.Time // 最后活动时间 +} + +var batchTracker = &batchCopyTracker{ + dirTasks: make(map[string]*dirTaskInfo), + pendingTasks: make(map[string]string), +} + +// 生成目标目录的唯一键 +func (bt *batchCopyTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { + return dstStorage.GetStorage().MountPath + ":" + dstDirPath +} + +// 注册复制任务到目标目录 +func (bt *batchCopyTracker) registerTask(taskID string, dstStorage driver.Driver, dstDirPath string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + dirKey := bt.getDirKey(dstStorage, dstDirPath) + + // 记录任务到目录的映射 + bt.pendingTasks[taskID] = dirKey + + // 初始化或更新目录任务信息 + if info, exists := bt.dirTasks[dirKey]; exists { + info.pendingTasks[taskID] = true + info.lastActivity = time.Now() + } else { + bt.dirTasks[dirKey] = &dirTaskInfo{ + dstStorage: dstStorage, + dstDirPath: dstDirPath, + pendingTasks: map[string]bool{taskID: true}, + lastActivity: time.Now(), + } + } +} + +// 标记任务完成，返回是否需要刷新缓存 +func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Driver, string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + dirKey, exists := bt.pendingTasks[taskID] + if !exists { + return false, nil, "" + } + + // 从待处理任务中移除 + delete(bt.pendingTasks, taskID) + + info, exists := bt.dirTasks[dirKey] + if !exists { + return false, nil, "" + } + + // 从目录任务中移除 + delete(info.pendingTasks, taskID) + info.lastActivity = time.Now() + + // 如果该目录下没有待处理的任务了，触发缓存刷新 + if len(info.pendingTasks) == 0 { + dstStorage := info.dstStorage + dstDirPath := info.dstDirPath + delete(bt.dirTasks, dirKey) + return true, dstStorage, dstDirPath + } + + return false, nil, "" +} + +// 清理超时的任务（可选的清理机制，防止内存泄漏） +func (bt *batchCopyTracker) cleanupStaleEntries() { + bt.mu.Lock() + defer bt.mu.Unlock() + + now := time.Now() + for dirKey, info := range bt.dirTasks { + // 如果超过1小时没有活动，清理该条目 + if now.Sub(info.lastActivity) > time.Hour { + // 清理相关的待处理任务 + for taskID := range info.pendingTasks { + delete(bt.pendingTasks, taskID) + } + delete(bt.dirTasks, dirKey) + } + } +} + // Copy if in the same storage, call move method // if not, add copy task func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (task.TaskExtensionInfo, error) { @@ -76,6 +201,10 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool if srcStorage.GetStorage() == dstStorage.GetStorage() { err = op.Copy(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...) if !errors.Is(err, errs.NotImplement) && !errors.Is(err, errs.NotSupport) { + if err == nil { + // 同存储复制成功后刷新目标目录缓存 + op.ClearCache(dstStorage, dstDirActualPath) + } return nil, err } } @@ -101,7 +230,12 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool if err != nil { return nil, errors.WithMessagef(err, "failed get [%s] stream", srcObjPath) } - return nil, op.Put(ctx, dstStorage, dstDirActualPath, ss, nil, false) + err = op.Put(ctx, dstStorage, dstDirActualPath, ss, nil, false) + if err == nil { + // 直接文件复制成功后刷新目标目录缓存 + op.ClearCache(dstStorage, dstDirActualPath) + } + return nil, err } } // not in the same storage @@ -134,6 +268,7 @@ func copyBetween2Storages(t *CopyTask, srcStorage, dstStorage driver.Driver, src if err != nil { return errors.WithMessagef(err, "failed list src [%s] objs", srcObjPath) } + for _, obj := range objs { if utils.IsCanceled(t.Ctx()) { return nil @@ -181,4 +316,4 @@ func copyFileBetween2Storages(tsk *CopyTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath) } return op.Put(tsk.Ctx(), dstStorage, dstDirPath, ss, tsk.SetProgress, true) -} +} \ No newline at end of file From 6c5d5be8a063ecebb77e8e827fe776233b696d8b Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 20:21:52 +0800 Subject: [PATCH 02/10] fixed randomstring --- internal/fs/copy.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/internal/fs/copy.go b/internal/fs/copy.go index f264b33a4..1d6556538 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -18,6 +18,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/pkg/utils" "github.com/OpenListTeam/OpenList/v4/server/common" "github.com/OpenListTeam/tache" + "github.com/google/uuid" "github.com/pkg/errors" ) @@ -51,7 +52,7 @@ func (t *CopyTask) Run() error { // 注册任务到批量跟踪器 taskID := t.GetID() if taskID == "" { - taskID = utils.RandomString(16) + taskID = uuid.NewString() } var err error From 8a0baba99fffd6a46858e15fb10872bfad1533ad Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 20:24:49 +0800 Subject: [PATCH 03/10] fixed EOL and Sync branch chore(quark_uc): `webdav_policy` default to native_proxy --- drivers/quark_uc/driver.go | 7 ++++--- internal/fs/copy.go | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/drivers/quark_uc/driver.go b/drivers/quark_uc/driver.go index 44d3425cc..4235acc61 100644 --- a/drivers/quark_uc/driver.go +++ b/drivers/quark_uc/driver.go @@ -37,10 +37,11 @@ func (d *QuarkOrUC) GetAddition() driver.Additional { func (d *QuarkOrUC) Init(ctx context.Context) error { _, err := d.request("/config", http.MethodGet, nil, nil) if err == nil { - if d.AdditionVersion != 1 { - d.AdditionVersion = 1 - if !d.UseTransCodingAddress { + if d.AdditionVersion != 2 { + d.AdditionVersion = 2 + if !d.UseTransCodingAddress && len(d.DownProxyUrl) == 0 { d.WebProxy = true + d.WebdavPolicy = "native_proxy" } } } diff --git a/internal/fs/copy.go b/internal/fs/copy.go index 1d6556538..22e2ce364 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -317,4 +317,4 @@ func copyFileBetween2Storages(tsk *CopyTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath) } return op.Put(tsk.Ctx(), dstStorage, dstDirPath, ss, tsk.SetProgress, true) -} \ No newline at end of file +} From d5f750dc06bb02775962ff76114618a613234aee Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 20:58:28 +0800 Subject: [PATCH 04/10] fixed uuid and other bugs --- internal/fs/copy.go | 45 ++++++++++++++++++++++++++++----------------- 1 file changed, 28 insertions(+), 17 deletions(-) diff --git a/internal/fs/copy.go b/internal/fs/copy.go index 22e2ce364..a1e659a2c 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -18,7 +18,6 @@ import ( "github.com/OpenListTeam/OpenList/v4/pkg/utils" "github.com/OpenListTeam/OpenList/v4/server/common" "github.com/OpenListTeam/tache" - "github.com/google/uuid" "github.com/pkg/errors" ) @@ -49,12 +48,6 @@ func (t *CopyTask) Run() error { t.SetStartTime(time.Now()) defer func() { t.SetEndTime(time.Now()) }() - // 注册任务到批量跟踪器 - taskID := t.GetID() - if taskID == "" { - taskID = uuid.NewString() - } - var err error if t.srcStorage == nil { t.srcStorage, err = op.GetStorageByMountPath(t.SrcStorageMp) @@ -66,13 +59,16 @@ func (t *CopyTask) Run() error { return errors.WithMessage(err, "failed get storage") } + // 使用任务对象的内存地址作为唯一标识符 + taskID := fmt.Sprintf("%p", t) + // 注册任务到批量跟踪器 batchTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) // 执行复制操作 err = copyBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath) - // 标记任务完成并检查是否需要刷新缓存 + // 标记任务完��并检查是否需要刷新缓存 if err == nil { shouldRefresh, dstStorage, dstDirPath := batchTracker.markTaskCompleted(taskID) if shouldRefresh { @@ -93,18 +89,20 @@ type batchCopyTracker struct { mu sync.Mutex dirTasks map[string]*dirTaskInfo // dstStoragePath+dstDirPath -> dirTaskInfo pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath + lastCleanup time.Time // 上次清理时间 } type dirTaskInfo struct { dstStorage driver.Driver dstDirPath string pendingTasks map[string]bool // taskID -> true - lastActivity time.Time // 最后活动时间 + lastActivity time.Time // 最后活动时间（用于检测异常情况） } var batchTracker = &batchCopyTracker{ dirTasks: make(map[string]*dirTaskInfo), pendingTasks: make(map[string]string), + lastCleanup: time.Now(), } // 生成目标目录的唯一键 @@ -117,6 +115,9 @@ func (bt *batchCopyTracker) registerTask(taskID string, dstStorage driver.Driver bt.mu.Lock() defer bt.mu.Unlock() + // 定期清理过期条目 + bt.cleanupIfNeeded() + dirKey := bt.getDirKey(dstStorage, dstDirPath) // 记录任务到目录的映射 @@ -156,27 +157,37 @@ func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Drive // 从目录任务中移除 delete(info.pendingTasks, taskID) - info.lastActivity = time.Now() // 如果该目录下没有待处理的任务了，触发缓存刷新 if len(info.pendingTasks) == 0 { dstStorage := info.dstStorage dstDirPath := info.dstDirPath - delete(bt.dirTasks, dirKey) + delete(bt.dirTasks, dirKey) // 直接删除，不需要更新lastActivity return true, dstStorage, dstDirPath } + // 只有当还有其他任务时才更新lastActivity（表示该目录仍有活跃任务） + info.lastActivity = time.Now() return false, nil, "" } -// 清理超时的任务（可选的清理机制，防止内存泄漏） +// 检查是否需要清理，如果需要则执行清理 +func (bt *batchCopyTracker) cleanupIfNeeded() { + now := time.Now() + // 每10分钟清理一次 + if now.Sub(bt.lastCleanup) > 10*time.Minute { + bt.cleanupStaleEntries() + bt.lastCleanup = now + } +} + +// 清理超时的任务（防止内存泄漏） +// 主要用于清理因异常情况（如任务崩溃、进程重启等）导致的残留条目 func (bt *batchCopyTracker) cleanupStaleEntries() { - bt.mu.Lock() - defer bt.mu.Unlock() - now := time.Now() for dirKey, info := range bt.dirTasks { - // 如果超过1小时没有活动，清理该条目 + // 如果超过1小时没有活动，说明可能出现了异常情况，清理该条目 + // 正常情况下，任务完成时会调用markTaskCompleted并删除整个条目 if now.Sub(info.lastActivity) > time.Hour { // 清理相关的待处理任务 for taskID := range info.pendingTasks { @@ -317,4 +328,4 @@ func copyFileBetween2Storages(tsk *CopyTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath) } return op.Put(tsk.Ctx(), dstStorage, dstDirPath, ss, tsk.SetProgress, true) -} +} \ No newline at end of file From 9b9d86e8709bc8f0b737fc6052935714b35537e6 Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 21:10:01 +0800 Subject: [PATCH 05/10] fixed comments --- internal/fs/copy.go | 56 ++++++++++++++++++++++----------------------- 1 file changed, 28 insertions(+), 28 deletions(-) diff --git a/internal/fs/copy.go b/internal/fs/copy.go index a1e659a2c..325893aa6 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -59,23 +59,23 @@ func (t *CopyTask) Run() error { return errors.WithMessage(err, "failed get storage") } - // 使用任务对象的内存地址作为唯一标识符 + // Use the task object's memory address as a unique identifier taskID := fmt.Sprintf("%p", t) - // 注册任务到批量跟踪器 + // Register task to batch tracker batchTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) - // 执行复制操作 + // Execute copy operation err = copyBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath) - // 标记任务完��并检查是否需要刷新缓存 + // Mark task completed and check if cache refresh is needed if err == nil { shouldRefresh, dstStorage, dstDirPath := batchTracker.markTaskCompleted(taskID) if shouldRefresh { op.ClearCache(dstStorage, dstDirPath) } } else { - // 即使失败也要清理跟踪器中的记录 + // Clean up tracker records even if task failed batchTracker.markTaskCompleted(taskID) } @@ -84,19 +84,19 @@ func (t *CopyTask) Run() error { var CopyTaskManager *tache.Manager[*CopyTask] -// 批量复制任务跟踪器 - 按目标目录聚合所有复制任务 +// Batch copy task tracker - aggregates all copy tasks by target directory type batchCopyTracker struct { mu sync.Mutex dirTasks map[string]*dirTaskInfo // dstStoragePath+dstDirPath -> dirTaskInfo pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath - lastCleanup time.Time // 上次清理时间 + lastCleanup time.Time // last cleanup time } type dirTaskInfo struct { dstStorage driver.Driver dstDirPath string pendingTasks map[string]bool // taskID -> true - lastActivity time.Time // 最后活动时间（用于检测异常情况） + lastActivity time.Time // last activity time (used for detecting abnormal situations) } var batchTracker = &batchCopyTracker{ @@ -105,25 +105,25 @@ var batchTracker = &batchCopyTracker{ lastCleanup: time.Now(), } -// 生成目标目录的唯一键 +// Generate unique key for target directory func (bt *batchCopyTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { return dstStorage.GetStorage().MountPath + ":" + dstDirPath } -// 注册复制任务到目标目录 +// Register copy task to target directory func (bt *batchCopyTracker) registerTask(taskID string, dstStorage driver.Driver, dstDirPath string) { bt.mu.Lock() defer bt.mu.Unlock() - // 定期清理过期条目 + // Periodically clean up expired entries bt.cleanupIfNeeded() dirKey := bt.getDirKey(dstStorage, dstDirPath) - // 记录任务到目录的映射 + // Record task to directory mapping bt.pendingTasks[taskID] = dirKey - // 初始化或更新目录任务信息 + // Initialize or update directory task information if info, exists := bt.dirTasks[dirKey]; exists { info.pendingTasks[taskID] = true info.lastActivity = time.Now() @@ -137,7 +137,7 @@ func (bt *batchCopyTracker) registerTask(taskID string, dstStorage driver.Driver } } -// 标记任务完成，返回是否需要刷新缓存 +// Mark task completed, return whether cache refresh is needed func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Driver, string) { bt.mu.Lock() defer bt.mu.Unlock() @@ -147,7 +147,7 @@ func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Drive return false, nil, "" } - // 从待处理任务中移除 + // Remove from pending tasks delete(bt.pendingTasks, taskID) info, exists := bt.dirTasks[dirKey] @@ -155,41 +155,41 @@ func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Drive return false, nil, "" } - // 从目录任务中移除 + // Remove from directory tasks delete(info.pendingTasks, taskID) - // 如果该目录下没有待处理的任务了，触发缓存刷新 + // If no pending tasks left in this directory, trigger cache refresh if len(info.pendingTasks) == 0 { dstStorage := info.dstStorage dstDirPath := info.dstDirPath - delete(bt.dirTasks, dirKey) // 直接删除，不需要更新lastActivity + delete(bt.dirTasks, dirKey) // Delete directly, no need to update lastActivity return true, dstStorage, dstDirPath } - // 只有当还有其他任务时才更新lastActivity（表示该目录仍有活跃任务） + // Only update lastActivity when there are other tasks (indicating the directory still has active tasks) info.lastActivity = time.Now() return false, nil, "" } -// 检查是否需要清理，如果需要则执行清理 +// Check if cleanup is needed, execute cleanup if necessary func (bt *batchCopyTracker) cleanupIfNeeded() { now := time.Now() - // 每10分钟清理一次 + // Clean up every 10 minutes if now.Sub(bt.lastCleanup) > 10*time.Minute { bt.cleanupStaleEntries() bt.lastCleanup = now } } -// 清理超时的任务（防止内存泄漏） -// 主要用于清理因异常情况（如任务崩溃、进程重启等）导致的残留条目 +// Clean up timed-out tasks (prevent memory leaks) +// Mainly used to clean up residual entries caused by abnormal situations (such as task crashes, process restarts, etc.) func (bt *batchCopyTracker) cleanupStaleEntries() { now := time.Now() for dirKey, info := range bt.dirTasks { - // 如果超过1小时没有活动，说明可能出现了异常情况，清理该条目 - // 正常情况下，任务完成时会调用markTaskCompleted并删除整个条目 + // If no activity for more than 1 hour, it may indicate an abnormal situation, clean up this entry + // Under normal circumstances, markTaskCompleted will be called when the task is completed and the entire entry will be deleted if now.Sub(info.lastActivity) > time.Hour { - // 清理相关的待处理任务 + // Clean up related pending tasks for taskID := range info.pendingTasks { delete(bt.pendingTasks, taskID) } @@ -214,7 +214,7 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool err = op.Copy(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...) if !errors.Is(err, errs.NotImplement) && !errors.Is(err, errs.NotSupport) { if err == nil { - // 同存储复制成功后刷新目标目录缓存 + // Refresh target directory cache after successful same-storage copy op.ClearCache(dstStorage, dstDirActualPath) } return nil, err @@ -244,7 +244,7 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool } err = op.Put(ctx, dstStorage, dstDirActualPath, ss, nil, false) if err == nil { - // 直接文件复制成功后刷新目标目录缓存 + // Refresh target directory cache after successful direct file copy op.ClearCache(dstStorage, dstDirActualPath) } return nil, err From 34f3209bf4e4d56e9dabd1af854df7201b1624d9 Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 21:13:27 +0800 Subject: [PATCH 06/10] fixed EOL --- internal/fs/copy.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/fs/copy.go b/internal/fs/copy.go index 325893aa6..de3102e19 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -328,4 +328,4 @@ func copyFileBetween2Storages(tsk *CopyTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed get [%s] stream", srcFilePath) } return op.Put(tsk.Ctx(), dstStorage, dstDirPath, ss, tsk.SetProgress, true) -} \ No newline at end of file +} From 50936c357fe5a719faac6ef457350103f3e84016 Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 21:19:29 +0800 Subject: [PATCH 07/10] add move refresh --- internal/fs/move.go | 162 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 157 insertions(+), 5 deletions(-) diff --git a/internal/fs/move.go b/internal/fs/move.go index 0410e9ce4..78e3d75de 100644 --- a/internal/fs/move.go +++ b/internal/fs/move.go @@ -130,6 +130,12 @@ func (t *MoveTask) Run() error { return errors.WithMessage(err, "failed get storage") } + // Use the task object's memory address as a unique identifier + taskID := fmt.Sprintf("%p", t) + + // Register task to batch tracker + batchMoveTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) + // Phase 1: Async validation (all validation happens in background) t.mu.Lock() t.Status = "validating source and destination" @@ -138,6 +144,8 @@ func (t *MoveTask) Run() error { // Check if source exists srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath) if err != nil { + // Clean up tracker records if task failed + batchMoveTracker.markTaskCompleted(taskID) return errors.WithMessagef(err, "source file [%s] not found", stdpath.Base(t.SrcObjPath)) } @@ -145,6 +153,8 @@ func (t *MoveTask) Run() error { if t.ValidateExistence { dstFilePath := stdpath.Join(t.DstDirPath, srcObj.GetName()) if res, _ := op.Get(t.Ctx(), t.dstStorage, dstFilePath); res != nil { + // Clean up tracker records if task failed + batchMoveTracker.markTaskCompleted(taskID) return errors.Errorf("destination file [%s] already exists", srcObj.GetName()) } } @@ -156,11 +166,24 @@ func (t *MoveTask) Run() error { t.IsRootTask = true t.RootTaskID = t.GetID() t.mu.Unlock() - return t.runRootMoveTask() + err = t.runRootMoveTask() + } else { + // Use safe move logic for files + err = t.safeMoveOperation(srcObj) } - // Use safe move logic for files - return t.safeMoveOperation(srcObj) + // Mark task completed and check if cache refresh is needed + if err == nil { + shouldRefresh, dstStorage, dstDirPath := batchMoveTracker.markTaskCompleted(taskID) + if shouldRefresh { + op.ClearCache(dstStorage, dstDirPath) + } + } else { + // Clean up tracker records even if task failed + batchMoveTracker.markTaskCompleted(taskID) + } + + return err } func (t *MoveTask) runRootMoveTask() error { @@ -236,11 +259,128 @@ func (t *MoveTask) runRootMoveTask() error { t.mu.Unlock() t.updateProgress() + // Refresh target directory cache after successful root move task + op.ClearCache(t.dstStorage, t.DstDirPath) + return nil } var MoveTaskManager *tache.Manager[*MoveTask] +// Batch move task tracker - aggregates all move tasks by target directory +type batchMoveTracker struct { + mu sync.Mutex + dirTasks map[string]*moveDirTaskInfo // dstStoragePath+dstDirPath -> moveDirTaskInfo + pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath + lastCleanup time.Time // last cleanup time +} + +type moveDirTaskInfo struct { + dstStorage driver.Driver + dstDirPath string + pendingTasks map[string]bool // taskID -> true + lastActivity time.Time // last activity time (used for detecting abnormal situations) +} + +var batchMoveTracker = &batchMoveTracker{ + dirTasks: make(map[string]*moveDirTaskInfo), + pendingTasks: make(map[string]string), + lastCleanup: time.Now(), +} + +// Generate unique key for target directory +func (bt *batchMoveTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { + return dstStorage.GetStorage().MountPath + ":" + dstDirPath +} + +// Register move task to target directory +func (bt *batchMoveTracker) registerTask(taskID string, dstStorage driver.Driver, dstDirPath string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + // Periodically clean up expired entries + bt.cleanupIfNeeded() + + dirKey := bt.getDirKey(dstStorage, dstDirPath) + + // Record task to directory mapping + bt.pendingTasks[taskID] = dirKey + + // Initialize or update directory task information + if info, exists := bt.dirTasks[dirKey]; exists { + info.pendingTasks[taskID] = true + info.lastActivity = time.Now() + } else { + bt.dirTasks[dirKey] = &moveDirTaskInfo{ + dstStorage: dstStorage, + dstDirPath: dstDirPath, + pendingTasks: map[string]bool{taskID: true}, + lastActivity: time.Now(), + } + } +} + +// Mark task completed, return whether cache refresh is needed +func (bt *batchMoveTracker) markTaskCompleted(taskID string) (bool, driver.Driver, string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + dirKey, exists := bt.pendingTasks[taskID] + if !exists { + return false, nil, "" + } + + // Remove from pending tasks + delete(bt.pendingTasks, taskID) + + info, exists := bt.dirTasks[dirKey] + if !exists { + return false, nil, "" + } + + // Remove from directory tasks + delete(info.pendingTasks, taskID) + + // If no pending tasks left in this directory, trigger cache refresh + if len(info.pendingTasks) == 0 { + dstStorage := info.dstStorage + dstDirPath := info.dstDirPath + delete(bt.dirTasks, dirKey) // Delete directly, no need to update lastActivity + return true, dstStorage, dstDirPath + } + + // Only update lastActivity when there are other tasks (indicating the directory still has active tasks) + info.lastActivity = time.Now() + return false, nil, "" +} + +// Check if cleanup is needed, execute cleanup if necessary +func (bt *batchMoveTracker) cleanupIfNeeded() { + now := time.Now() + // Clean up every 10 minutes + if now.Sub(bt.lastCleanup) > 10*time.Minute { + bt.cleanupStaleEntries() + bt.lastCleanup = now + } +} + +// Clean up timed-out tasks (prevent memory leaks) +// Mainly used to clean up residual entries caused by abnormal situations (such as task crashes, process restarts, etc.) +func (bt *batchMoveTracker) cleanupStaleEntries() { + now := time.Now() + for dirKey, info := range bt.dirTasks { + // If no activity for more than 1 hour, it may indicate an abnormal situation, clean up this entry + // Under normal circumstances, markTaskCompleted will be called when the task is completed and the entire entry will be deleted + if now.Sub(info.lastActivity) > time.Hour { + // Clean up related pending tasks + for taskID := range info.pendingTasks { + delete(bt.pendingTasks, taskID) + } + delete(bt.dirTasks, dirKey) + } + } +} + // GetMoveProgress returns the progress of a move task by task ID func GetMoveProgress(taskID string) (*MoveProgress, bool) { if progress, ok := moveProgressMap.Load(taskID); ok { @@ -492,7 +632,7 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src return nil } srcSubObjPath := stdpath.Join(srcObjPath, obj.GetName()) - MoveTaskManager.Add(&MoveTask{ + subTask := &MoveTask{ TaskExtension: task.TaskExtension{ Creator: t.GetCreator(), ApiUrl: t.ApiUrl, @@ -503,7 +643,8 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src DstDirPath: dstObjPath, SrcStorageMp: srcStorage.GetStorage().MountPath, DstStorageMp: dstStorage.GetStorage().MountPath, - }) + } + MoveTaskManager.Add(subTask) } t.Status = "cleaning up source directory" @@ -513,6 +654,10 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src } else { t.Status = "completed" } + + // Refresh target directory cache after successful directory move + op.ClearCache(dstStorage, dstDirPath) + return nil } else { return moveFileBetween2Storages(t, srcStorage, dstStorage, srcObjPath, dstDirPath) @@ -554,6 +699,9 @@ func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed to delete src [%s] file from storage [%s] after successful copy", srcFilePath, srcStorage.GetStorage().MountPath) } + // For single file moves, refresh cache immediately since there's no batching needed + op.ClearCache(dstStorage, dstDirPath) + tsk.SetProgress(100) tsk.Status = "completed" return nil @@ -588,6 +736,10 @@ func _moveWithValidation(ctx context.Context, srcObjPath, dstDirPath string, val if srcStorage.GetStorage() == dstStorage.GetStorage() { err = op.Move(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...) if !errors.Is(err, errs.NotImplement) && !errors.Is(err, errs.NotSupport) { + if err == nil { + // Refresh target directory cache after successful same-storage move + op.ClearCache(dstStorage, dstDirActualPath) + } return nil, err } } From 9d2daab0149b78671401679d2683ed24ee9580a7 Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 21:28:19 +0800 Subject: [PATCH 08/10] fixed builds --- internal/fs/move.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/internal/fs/move.go b/internal/fs/move.go index 78e3d75de..3a4f2afd4 100644 --- a/internal/fs/move.go +++ b/internal/fs/move.go @@ -134,7 +134,7 @@ func (t *MoveTask) Run() error { taskID := fmt.Sprintf("%p", t) // Register task to batch tracker - batchMoveTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) + batchMoveTrackerInstance.registerTask(taskID, t.dstStorage, t.DstDirPath) // Phase 1: Async validation (all validation happens in background) t.mu.Lock() @@ -145,7 +145,7 @@ func (t *MoveTask) Run() error { srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath) if err != nil { // Clean up tracker records if task failed - batchMoveTracker.markTaskCompleted(taskID) + batchMoveTrackerInstance.markTaskCompleted(taskID) return errors.WithMessagef(err, "source file [%s] not found", stdpath.Base(t.SrcObjPath)) } @@ -154,7 +154,7 @@ func (t *MoveTask) Run() error { dstFilePath := stdpath.Join(t.DstDirPath, srcObj.GetName()) if res, _ := op.Get(t.Ctx(), t.dstStorage, dstFilePath); res != nil { // Clean up tracker records if task failed - batchMoveTracker.markTaskCompleted(taskID) + batchMoveTrackerInstance.markTaskCompleted(taskID) return errors.Errorf("destination file [%s] already exists", srcObj.GetName()) } } @@ -174,13 +174,13 @@ func (t *MoveTask) Run() error { // Mark task completed and check if cache refresh is needed if err == nil { - shouldRefresh, dstStorage, dstDirPath := batchMoveTracker.markTaskCompleted(taskID) + shouldRefresh, dstStorage, dstDirPath := batchMoveTrackerInstance.markTaskCompleted(taskID) if shouldRefresh { op.ClearCache(dstStorage, dstDirPath) } } else { // Clean up tracker records even if task failed - batchMoveTracker.markTaskCompleted(taskID) + batchMoveTrackerInstance.markTaskCompleted(taskID) } return err @@ -282,7 +282,7 @@ type moveDirTaskInfo struct { lastActivity time.Time // last activity time (used for detecting abnormal situations) } -var batchMoveTracker = &batchMoveTracker{ +var batchMoveTrackerInstance = &batchMoveTracker{ dirTasks: make(map[string]*moveDirTaskInfo), pendingTasks: make(map[string]string), lastCleanup: time.Now(), From 0cbd65f6199ec611849dc8d6168e1c558f0e6ffd Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 21:48:35 +0800 Subject: [PATCH 09/10] fixed batch --- internal/fs/batch_tracker.go | 153 +++++++++++++++++++++++++++++++++++ internal/fs/copy.go | 130 ++--------------------------- internal/fs/move.go | 147 +++------------------------------ 3 files changed, 169 insertions(+), 261 deletions(-) create mode 100644 internal/fs/batch_tracker.go diff --git a/internal/fs/batch_tracker.go b/internal/fs/batch_tracker.go new file mode 100644 index 000000000..15073e505 --- /dev/null +++ b/internal/fs/batch_tracker.go @@ -0,0 +1,153 @@ +package fs + +import ( + "sync" + "time" + + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/op" +) + +// BatchTracker manages batch operations for cache refresh optimization +// It aggregates multiple file operations by target directory and only refreshes +// the cache once when all operations in a directory are completed +type BatchTracker struct { + mu sync.Mutex + dirTasks map[string]*dirTaskInfo // dstStoragePath+dstDirPath -> dirTaskInfo + pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath + lastCleanup time.Time // last cleanup time + name string // tracker name for debugging +} + +type dirTaskInfo struct { + dstStorage driver.Driver + dstDirPath string + pendingTasks map[string]bool // taskID -> true + lastActivity time.Time // last activity time (used for detecting abnormal situations) +} + +// NewBatchTracker creates a new batch tracker instance +func NewBatchTracker(name string) *BatchTracker { + return &BatchTracker{ + dirTasks: make(map[string]*dirTaskInfo), + pendingTasks: make(map[string]string), + lastCleanup: time.Now(), + name: name, + } +} + +// getDirKey generates unique key for target directory +func (bt *BatchTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { + return dstStorage.GetStorage().MountPath + ":" + dstDirPath +} + +// RegisterTask registers a task to target directory for batch tracking +func (bt *BatchTracker) RegisterTask(taskID string, dstStorage driver.Driver, dstDirPath string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + // Periodically clean up expired entries + bt.cleanupIfNeeded() + + dirKey := bt.getDirKey(dstStorage, dstDirPath) + + // Record task to directory mapping + bt.pendingTasks[taskID] = dirKey + + // Initialize or update directory task information + if info, exists := bt.dirTasks[dirKey]; exists { + info.pendingTasks[taskID] = true + info.lastActivity = time.Now() + } else { + bt.dirTasks[dirKey] = &dirTaskInfo{ + dstStorage: dstStorage, + dstDirPath: dstDirPath, + pendingTasks: map[string]bool{taskID: true}, + lastActivity: time.Now(), + } + } +} + +// MarkTaskCompleted marks a task as completed and returns whether cache refresh is needed +// Returns (shouldRefresh, dstStorage, dstDirPath) +func (bt *BatchTracker) MarkTaskCompleted(taskID string) (bool, driver.Driver, string) { + bt.mu.Lock() + defer bt.mu.Unlock() + + dirKey, exists := bt.pendingTasks[taskID] + if !exists { + return false, nil, "" + } + + // Remove from pending tasks + delete(bt.pendingTasks, taskID) + + info, exists := bt.dirTasks[dirKey] + if !exists { + return false, nil, "" + } + + // Remove from directory tasks + delete(info.pendingTasks, taskID) + + // If no pending tasks left in this directory, trigger cache refresh + if len(info.pendingTasks) == 0 { + dstStorage := info.dstStorage + dstDirPath := info.dstDirPath + delete(bt.dirTasks, dirKey) // Delete directly, no need to update lastActivity + return true, dstStorage, dstDirPath + } + + // Only update lastActivity when there are other tasks (indicating the directory still has active tasks) + info.lastActivity = time.Now() + return false, nil, "" +} + +// MarkTaskCompletedWithRefresh marks a task as completed and automatically refreshes cache if needed +func (bt *BatchTracker) MarkTaskCompletedWithRefresh(taskID string) { + shouldRefresh, dstStorage, dstDirPath := bt.MarkTaskCompleted(taskID) + if shouldRefresh { + op.ClearCache(dstStorage, dstDirPath) + } +} + +// cleanupIfNeeded checks if cleanup is needed and executes cleanup if necessary +func (bt *BatchTracker) cleanupIfNeeded() { + now := time.Now() + // Clean up every 10 minutes + if now.Sub(bt.lastCleanup) > 10*time.Minute { + bt.cleanupStaleEntries() + bt.lastCleanup = now + } +} + +// cleanupStaleEntries cleans up timed-out tasks to prevent memory leaks +// Mainly used to clean up residual entries caused by abnormal situations (such as task crashes, process restarts, etc.) +func (bt *BatchTracker) cleanupStaleEntries() { + now := time.Now() + for dirKey, info := range bt.dirTasks { + // If no activity for more than 1 hour, it may indicate an abnormal situation, clean up this entry + // Under normal circumstances, MarkTaskCompleted will be called when the task is completed and the entire entry will be deleted + if now.Sub(info.lastActivity) > time.Hour { + // Clean up related pending tasks + for taskID := range info.pendingTasks { + delete(bt.pendingTasks, taskID) + } + delete(bt.dirTasks, dirKey) + } + } +} + +// GetPendingTaskCount returns the number of pending tasks for debugging +func (bt *BatchTracker) GetPendingTaskCount() int { + bt.mu.Lock() + defer bt.mu.Unlock() + return len(bt.pendingTasks) +} + +// GetDirTaskCount returns the number of directories being tracked for debugging +func (bt *BatchTracker) GetDirTaskCount() int { + bt.mu.Lock() + defer bt.mu.Unlock() + return len(bt.dirTasks) +} \ No newline at end of file diff --git a/internal/fs/copy.go b/internal/fs/copy.go index de3102e19..31cf56056 100644 --- a/internal/fs/copy.go +++ b/internal/fs/copy.go @@ -5,7 +5,6 @@ import ( "fmt" "net/http" stdpath "path" - "sync" "time" "github.com/OpenListTeam/OpenList/v4/internal/conf" @@ -63,140 +62,21 @@ func (t *CopyTask) Run() error { taskID := fmt.Sprintf("%p", t) // Register task to batch tracker - batchTracker.registerTask(taskID, t.dstStorage, t.DstDirPath) + copyBatchTracker.RegisterTask(taskID, t.dstStorage, t.DstDirPath) // Execute copy operation err = copyBetween2Storages(t, t.srcStorage, t.dstStorage, t.SrcObjPath, t.DstDirPath) - // Mark task completed and check if cache refresh is needed - if err == nil { - shouldRefresh, dstStorage, dstDirPath := batchTracker.markTaskCompleted(taskID) - if shouldRefresh { - op.ClearCache(dstStorage, dstDirPath) - } - } else { - // Clean up tracker records even if task failed - batchTracker.markTaskCompleted(taskID) - } + // Mark task completed and automatically refresh cache if needed + copyBatchTracker.MarkTaskCompletedWithRefresh(taskID) return err } var CopyTaskManager *tache.Manager[*CopyTask] -// Batch copy task tracker - aggregates all copy tasks by target directory -type batchCopyTracker struct { - mu sync.Mutex - dirTasks map[string]*dirTaskInfo // dstStoragePath+dstDirPath -> dirTaskInfo - pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath - lastCleanup time.Time // last cleanup time -} - -type dirTaskInfo struct { - dstStorage driver.Driver - dstDirPath string - pendingTasks map[string]bool // taskID -> true - lastActivity time.Time // last activity time (used for detecting abnormal situations) -} - -var batchTracker = &batchCopyTracker{ - dirTasks: make(map[string]*dirTaskInfo), - pendingTasks: make(map[string]string), - lastCleanup: time.Now(), -} - -// Generate unique key for target directory -func (bt *batchCopyTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { - return dstStorage.GetStorage().MountPath + ":" + dstDirPath -} - -// Register copy task to target directory -func (bt *batchCopyTracker) registerTask(taskID string, dstStorage driver.Driver, dstDirPath string) { - bt.mu.Lock() - defer bt.mu.Unlock() - - // Periodically clean up expired entries - bt.cleanupIfNeeded() - - dirKey := bt.getDirKey(dstStorage, dstDirPath) - - // Record task to directory mapping - bt.pendingTasks[taskID] = dirKey - - // Initialize or update directory task information - if info, exists := bt.dirTasks[dirKey]; exists { - info.pendingTasks[taskID] = true - info.lastActivity = time.Now() - } else { - bt.dirTasks[dirKey] = &dirTaskInfo{ - dstStorage: dstStorage, - dstDirPath: dstDirPath, - pendingTasks: map[string]bool{taskID: true}, - lastActivity: time.Now(), - } - } -} - -// Mark task completed, return whether cache refresh is needed -func (bt *batchCopyTracker) markTaskCompleted(taskID string) (bool, driver.Driver, string) { - bt.mu.Lock() - defer bt.mu.Unlock() - - dirKey, exists := bt.pendingTasks[taskID] - if !exists { - return false, nil, "" - } - - // Remove from pending tasks - delete(bt.pendingTasks, taskID) - - info, exists := bt.dirTasks[dirKey] - if !exists { - return false, nil, "" - } - - // Remove from directory tasks - delete(info.pendingTasks, taskID) - - // If no pending tasks left in this directory, trigger cache refresh - if len(info.pendingTasks) == 0 { - dstStorage := info.dstStorage - dstDirPath := info.dstDirPath - delete(bt.dirTasks, dirKey) // Delete directly, no need to update lastActivity - return true, dstStorage, dstDirPath - } - - // Only update lastActivity when there are other tasks (indicating the directory still has active tasks) - info.lastActivity = time.Now() - return false, nil, "" -} - -// Check if cleanup is needed, execute cleanup if necessary -func (bt *batchCopyTracker) cleanupIfNeeded() { - now := time.Now() - // Clean up every 10 minutes - if now.Sub(bt.lastCleanup) > 10*time.Minute { - bt.cleanupStaleEntries() - bt.lastCleanup = now - } -} - -// Clean up timed-out tasks (prevent memory leaks) -// Mainly used to clean up residual entries caused by abnormal situations (such as task crashes, process restarts, etc.) -func (bt *batchCopyTracker) cleanupStaleEntries() { - now := time.Now() - for dirKey, info := range bt.dirTasks { - // If no activity for more than 1 hour, it may indicate an abnormal situation, clean up this entry - // Under normal circumstances, markTaskCompleted will be called when the task is completed and the entire entry will be deleted - if now.Sub(info.lastActivity) > time.Hour { - // Clean up related pending tasks - for taskID := range info.pendingTasks { - delete(bt.pendingTasks, taskID) - } - delete(bt.dirTasks, dirKey) - } - } -} +// Batch tracker for copy operations +var copyBatchTracker = NewBatchTracker("copy") // Copy if in the same storage, call move method // if not, add copy task diff --git a/internal/fs/move.go b/internal/fs/move.go index 3a4f2afd4..e25cf1897 100644 --- a/internal/fs/move.go +++ b/internal/fs/move.go @@ -134,7 +134,7 @@ func (t *MoveTask) Run() error { taskID := fmt.Sprintf("%p", t) // Register task to batch tracker - batchMoveTrackerInstance.registerTask(taskID, t.dstStorage, t.DstDirPath) + moveBatchTracker.RegisterTask(taskID, t.dstStorage, t.DstDirPath) // Phase 1: Async validation (all validation happens in background) t.mu.Lock() @@ -145,7 +145,7 @@ func (t *MoveTask) Run() error { srcObj, err := op.Get(t.Ctx(), t.srcStorage, t.SrcObjPath) if err != nil { // Clean up tracker records if task failed - batchMoveTrackerInstance.markTaskCompleted(taskID) + moveBatchTracker.MarkTaskCompletedWithRefresh(taskID) return errors.WithMessagef(err, "source file [%s] not found", stdpath.Base(t.SrcObjPath)) } @@ -154,7 +154,7 @@ func (t *MoveTask) Run() error { dstFilePath := stdpath.Join(t.DstDirPath, srcObj.GetName()) if res, _ := op.Get(t.Ctx(), t.dstStorage, dstFilePath); res != nil { // Clean up tracker records if task failed - batchMoveTrackerInstance.markTaskCompleted(taskID) + moveBatchTracker.MarkTaskCompletedWithRefresh(taskID) return errors.Errorf("destination file [%s] already exists", srcObj.GetName()) } } @@ -172,16 +172,8 @@ func (t *MoveTask) Run() error { err = t.safeMoveOperation(srcObj) } - // Mark task completed and check if cache refresh is needed - if err == nil { - shouldRefresh, dstStorage, dstDirPath := batchMoveTrackerInstance.markTaskCompleted(taskID) - if shouldRefresh { - op.ClearCache(dstStorage, dstDirPath) - } - } else { - // Clean up tracker records even if task failed - batchMoveTrackerInstance.markTaskCompleted(taskID) - } + // Mark task completed and automatically refresh cache if needed + moveBatchTracker.MarkTaskCompletedWithRefresh(taskID) return err } @@ -259,127 +251,14 @@ func (t *MoveTask) runRootMoveTask() error { t.mu.Unlock() t.updateProgress() - // Refresh target directory cache after successful root move task - op.ClearCache(t.dstStorage, t.DstDirPath) - + return nil } var MoveTaskManager *tache.Manager[*MoveTask] -// Batch move task tracker - aggregates all move tasks by target directory -type batchMoveTracker struct { - mu sync.Mutex - dirTasks map[string]*moveDirTaskInfo // dstStoragePath+dstDirPath -> moveDirTaskInfo - pendingTasks map[string]string // taskID -> dstStoragePath+dstDirPath - lastCleanup time.Time // last cleanup time -} - -type moveDirTaskInfo struct { - dstStorage driver.Driver - dstDirPath string - pendingTasks map[string]bool // taskID -> true - lastActivity time.Time // last activity time (used for detecting abnormal situations) -} - -var batchMoveTrackerInstance = &batchMoveTracker{ - dirTasks: make(map[string]*moveDirTaskInfo), - pendingTasks: make(map[string]string), - lastCleanup: time.Now(), -} - -// Generate unique key for target directory -func (bt *batchMoveTracker) getDirKey(dstStorage driver.Driver, dstDirPath string) string { - return dstStorage.GetStorage().MountPath + ":" + dstDirPath -} - -// Register move task to target directory -func (bt *batchMoveTracker) registerTask(taskID string, dstStorage driver.Driver, dstDirPath string) { - bt.mu.Lock() - defer bt.mu.Unlock() - - // Periodically clean up expired entries - bt.cleanupIfNeeded() - - dirKey := bt.getDirKey(dstStorage, dstDirPath) - - // Record task to directory mapping - bt.pendingTasks[taskID] = dirKey - - // Initialize or update directory task information - if info, exists := bt.dirTasks[dirKey]; exists { - info.pendingTasks[taskID] = true - info.lastActivity = time.Now() - } else { - bt.dirTasks[dirKey] = &moveDirTaskInfo{ - dstStorage: dstStorage, - dstDirPath: dstDirPath, - pendingTasks: map[string]bool{taskID: true}, - lastActivity: time.Now(), - } - } -} - -// Mark task completed, return whether cache refresh is needed -func (bt *batchMoveTracker) markTaskCompleted(taskID string) (bool, driver.Driver, string) { - bt.mu.Lock() - defer bt.mu.Unlock() - - dirKey, exists := bt.pendingTasks[taskID] - if !exists { - return false, nil, "" - } - - // Remove from pending tasks - delete(bt.pendingTasks, taskID) - - info, exists := bt.dirTasks[dirKey] - if !exists { - return false, nil, "" - } - - // Remove from directory tasks - delete(info.pendingTasks, taskID) - - // If no pending tasks left in this directory, trigger cache refresh - if len(info.pendingTasks) == 0 { - dstStorage := info.dstStorage - dstDirPath := info.dstDirPath - delete(bt.dirTasks, dirKey) // Delete directly, no need to update lastActivity - return true, dstStorage, dstDirPath - } - - // Only update lastActivity when there are other tasks (indicating the directory still has active tasks) - info.lastActivity = time.Now() - return false, nil, "" -} - -// Check if cleanup is needed, execute cleanup if necessary -func (bt *batchMoveTracker) cleanupIfNeeded() { - now := time.Now() - // Clean up every 10 minutes - if now.Sub(bt.lastCleanup) > 10*time.Minute { - bt.cleanupStaleEntries() - bt.lastCleanup = now - } -} - -// Clean up timed-out tasks (prevent memory leaks) -// Mainly used to clean up residual entries caused by abnormal situations (such as task crashes, process restarts, etc.) -func (bt *batchMoveTracker) cleanupStaleEntries() { - now := time.Now() - for dirKey, info := range bt.dirTasks { - // If no activity for more than 1 hour, it may indicate an abnormal situation, clean up this entry - // Under normal circumstances, markTaskCompleted will be called when the task is completed and the entire entry will be deleted - if now.Sub(info.lastActivity) > time.Hour { - // Clean up related pending tasks - for taskID := range info.pendingTasks { - delete(bt.pendingTasks, taskID) - } - delete(bt.dirTasks, dirKey) - } - } -} +// Batch tracker for move operations +var moveBatchTracker = NewBatchTracker("move") // GetMoveProgress returns the progress of a move task by task ID func GetMoveProgress(taskID string) (*MoveProgress, bool) { @@ -655,9 +534,7 @@ func moveBetween2Storages(t *MoveTask, srcStorage, dstStorage driver.Driver, src t.Status = "completed" } - // Refresh target directory cache after successful directory move - op.ClearCache(dstStorage, dstDirPath) - + return nil } else { return moveFileBetween2Storages(t, srcStorage, dstStorage, srcObjPath, dstDirPath) @@ -699,9 +576,7 @@ func moveFileBetween2Storages(tsk *MoveTask, srcStorage, dstStorage driver.Drive return errors.WithMessagef(err, "failed to delete src [%s] file from storage [%s] after successful copy", srcFilePath, srcStorage.GetStorage().MountPath) } - // For single file moves, refresh cache immediately since there's no batching needed - op.ClearCache(dstStorage, dstDirPath) - + tsk.SetProgress(100) tsk.Status = "completed" return nil @@ -737,7 +612,7 @@ func _moveWithValidation(ctx context.Context, srcObjPath, dstDirPath string, val err = op.Move(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...) if !errors.Is(err, errs.NotImplement) && !errors.Is(err, errs.NotSupport) { if err == nil { - // Refresh target directory cache after successful same-storage move + // For same-storage moves, refresh cache immediately since no batch tracking is used op.ClearCache(dstStorage, dstDirActualPath) } return nil, err From 6e124211ad1ecce6e626053aac3358cfd7c54c4f Mon Sep 17 00:00:00 2001 From: Suyunmeng <69945917+Suyunmeng@users.noreply.github.com> Date: Fri, 4 Jul 2025 22:20:15 +0800 Subject: [PATCH 10/10] change betch to task.go --- internal/fs/{batch_tracker.go => task.go} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename internal/fs/{batch_tracker.go => task.go} (99%) diff --git a/internal/fs/batch_tracker.go b/internal/fs/task.go similarity index 99% rename from internal/fs/batch_tracker.go rename to internal/fs/task.go index 15073e505..94f84c77e 100644 --- a/internal/fs/batch_tracker.go +++ b/internal/fs/task.go @@ -150,4 +150,4 @@ func (bt *BatchTracker) GetDirTaskCount() int { bt.mu.Lock() defer bt.mu.Unlock() return len(bt.dirTasks) -} \ No newline at end of file +}