Something went wrong. Try again.
A privacy-first, self-hosted, fully open source personal knowledge management software, written in typescript and golang. (PERSONAL FORK)
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457// SiYuan - Refactor your thinking// Copyright (c) 2020-present, b3log.org//// This program is free software: you can redistribute it and/or modify// it under the terms of the GNU Affero General Public License as published by// the Free Software Foundation, either version 3 of the License, or// (at your option) any later version.//// This program is distributed in the hope that it will be useful,// but WITHOUT ANY WARRANTY; without even the implied warranty of// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the// GNU Affero General Public License for more details.//// You should have received a copy of the GNU Affero General Public License// along with this program. If not, see <https://www.gnu.org/licenses/>.
package task
import ( "context" "reflect" "sync" "time"
"github.com/88250/gulu" "github.com/siyuan-note/logging" "github.com/siyuan-note/siyuan/kernel/util")
var ( taskQueue []*Task queueLock = sync.Mutex{})
type Task struct { Action string Handler reflect.Value Args []interface{} Created time.Time Async bool // 为 true 说明是异步任务，不会阻塞任务队列，满足 Delay 条件后立即执行 Delay time.Duration Timeout time.Duration}
func AppendTask(action string, handler interface{}, args ...interface{}) { appendTaskWithDelayTimeout(action, false, 0, 24*time.Hour, handler, args...)}
func AppendAsyncTaskWithDelay(action string, delay time.Duration, handler interface{}, args ...interface{}) { appendTaskWithDelayTimeout(action, true, delay, 24*time.Hour, handler, args...)}
func AppendTaskWithTimeout(action string, timeout time.Duration, handler interface{}, args ...interface{}) { appendTaskWithDelayTimeout(action, false, 0, timeout, handler, args...)}
func appendTaskWithDelayTimeout(action string, async bool, delay, timeout time.Duration, handler interface{}, args ...interface{}) { if util.IsExiting.Load() { //logging.LogWarnf("task queue is paused, action [%s] will be ignored", action) return }
task := &Task{ Action: action, Handler: reflect.ValueOf(handler), Args: args, Created: time.Now(), Async: async, Delay: delay, Timeout: timeout, }
if gulu.Str.Contains(action, uniqueActions) { if currentTasks := getCurrentTasks(); containTask(task, currentTasks) { //logging.LogWarnf("task [%s] is already in queue, will be ignored", action) return } }
queueLock.Lock() defer queueLock.Unlock() taskQueue = append(taskQueue, task)}
func containTask(task *Task, tasks []*Task) bool { for _, t := range tasks { if t.Action == task.Action { if len(t.Args) != len(task.Args) { return false }
for i, arg := range t.Args { if !areArgsEqual(arg, task.Args[i]) { return false } } return true } } return false}
// areArgsEqual 比较两个参数是否相等func areArgsEqual(a, b interface{}) bool {
// 如果两个参数都为 nil if a == nil && b == nil { return true } if a == nil || b == nil { return false }
// 快速处理常见的基本类型 switch av := a.(type) { case string: if bv, ok := b.(string); ok { return av == bv } case int: if bv, ok := b.(int); ok { return av == bv } case int64: if bv, ok := b.(int64); ok { return av == bv } case int32: if bv, ok := b.(int32); ok { return av == bv } case bool: if bv, ok := b.(bool); ok { return av == bv } case float64: if bv, ok := b.(float64); ok { return av == bv } case float32: if bv, ok := b.(float32); ok { return av == bv } case uint: if bv, ok := b.(uint); ok { return av == bv } case uint64: if bv, ok := b.(uint64); ok { return av == bv } case uint32: if bv, ok := b.(uint32); ok { return av == bv } case []string: if bv, ok := b.([]string); ok { if len(av) != len(bv) { return false } for i := range av { if av[i] != bv[i] { return false } } return true } case []int: if bv, ok := b.([]int); ok { if len(av) != len(bv) { return false } for i := range av { if av[i] != bv[i] { return false } } return true } }
// 未处理的复杂类型，回退到 reflect.DeepEqual return reflect.DeepEqual(a, b)}
func getCurrentTasks() (ret []*Task) { queueLock.Lock() defer queueLock.Unlock()
currentTaskLock.Lock() if nil != currentTask { ret = append(ret, currentTask) } currentTaskLock.Unlock()
for _, task := range taskQueue { ret = append(ret, task) } return}
const ( RepoCheckout = "task.repo.checkout" // 从快照中检出 RepoAutoPurge = "task.repo.autoPurge" // 自动清理数据仓库 DatabaseIndexFull = "task.database.index.full" // 重建索引 DatabaseIndex = "task.database.index" // 数据库索引 DatabaseIndexCommit = "task.database.index.commit" // 数据库索引提交 DatabaseIndexRef = "task.database.index.ref" // 数据库索引引用 DatabaseIndexFix = "task.database.index.fix" // 数据库索引订正 OCRImage = "task.ocr.image" // 图片 OCR 提取文本 HistoryGenerateFile = "task.history.generateFile" // 生成文件历史 HistoryDatabaseIndexFull = "task.history.database.index.full" // 历史数据库重建索引 HistoryDatabaseIndexCommit = "task.history.database.index.commit" // 历史数据库索引提交 DatabaseIndexEmbedBlock = "task.database.index.embedBlock" // 数据库索引嵌入块 ReloadUI = "task.reload.ui" // 重载 UI AssetContentDatabaseIndexFull = "task.asset.database.index.full" // 资源文件数据库重建索引 AssetContentDatabaseIndexCommit = "task.asset.database.index.commit" // 资源文件数据库索引提交 CacheVirtualBlockRef = "task.cache.virtualBlockRef" // 缓存虚拟块引用 ReloadAttributeView = "task.reload.attributeView" // 重新加载属性视图 ReloadProtyle = "task.reload.protyle" // 重新加载编辑器 ReloadTag = "task.reload.tag" // 重新加载标签面板 ReloadFiletree = "task.reload.filetree" // 重新加载文档树面板 SetRefDynamicText = "task.ref.setDynamicText" // 设置引用的动态锚文本 SetDefRefCount = "task.def.setRefCount" // 设置定义的引用计数 UpdateIDs = "task.update.ids" // 更新 ID PushMsg = "task.push.msg" // 推送消息)
// uniqueActions 描述了唯一的任务，即队列中只能存在一个在执行的任务。var uniqueActions = []string{ RepoCheckout, RepoAutoPurge, DatabaseIndexFull, DatabaseIndexCommit, OCRImage, HistoryGenerateFile, HistoryDatabaseIndexFull, HistoryDatabaseIndexCommit, AssetContentDatabaseIndexFull, AssetContentDatabaseIndexCommit, ReloadAttributeView, ReloadProtyle, ReloadTag, ReloadFiletree, SetRefDynamicText, SetDefRefCount, UpdateIDs,}
func ContainIndexTask() bool { tasks := getCurrentTasks() for _, task := range tasks { if gulu.Str.Contains(task.Action, []string{DatabaseIndexFull, DatabaseIndex}) { return true } } return false}
func StatusJob() { var items []map[string]interface{} count := map[string]int{} actionLangs := util.TaskActionLangs[util.Lang]
queueLock.Lock() for _, task := range taskQueue { action := task.Action if c := count[action]; 7 < c { logging.LogWarnf("too many tasks [%s], ignore show its status", action) continue } count[action]++
if skipPushTaskAction(action) { continue }
if nil != actionLangs { if label := actionLangs[task.Action]; nil != label { action = label.(string) } else { continue } }
item := map[string]interface{}{"action": action} items = append(items, item) } defer queueLock.Unlock()
currentTaskLock.Lock() if nil != currentTask && nil != actionLangs && !skipPushTaskAction(currentTask.Action) { if label := actionLangs[currentTask.Action]; nil != label { items = append([]map[string]interface{}{{"action": label.(string)}}, items...) } } currentTaskLock.Unlock()
if 1 > len(items) { items = []map[string]interface{}{} } data := map[string]interface{}{} data["tasks"] = items util.PushBackgroundTask(data)}
func skipPushTaskAction(action string) bool { switch action { case DatabaseIndexCommit: return util.StatusBarCfg.MsgTaskDatabaseIndexCommitDisabled case HistoryDatabaseIndexCommit: return util.StatusBarCfg.MsgTaskHistoryDatabaseIndexCommitDisabled case AssetContentDatabaseIndexCommit: return util.StatusBarCfg.MsgTaskAssetDatabaseIndexCommitDisabled case HistoryGenerateFile: return util.StatusBarCfg.MsgTaskHistoryGenerateFileDisabled default: return false }}
func ExecTaskJob() { task := popTask() if nil == task { return }
if util.IsExiting.Load() { return }
execTask(task)}
func popTask() (ret *Task) { queueLock.Lock() defer queueLock.Unlock()
if 1 > len(taskQueue) { return }
for i, task := range taskQueue { if time.Since(task.Created) <= task.Delay { continue }
if !task.Async { ret = task taskQueue = append(taskQueue[:i], taskQueue[i+1:]...) return } } return}
func ExecAsyncTaskJob() { tasks := popAsyncTasks() if 1 > len(tasks) { return }
if util.IsExiting.Load() { return }
for _, task := range tasks { go func() { execTask(task) }() }}
func popAsyncTasks() (ret []*Task) { queueLock.Lock() defer queueLock.Unlock()
if 1 > len(taskQueue) { return }
// writeIdx 指向下一个要写入的位置 writeIdx := 0 for readIdx := 0; readIdx < len(taskQueue); readIdx++ { task := taskQueue[readIdx]
// 判断是否应该弹出此任务 shouldPop := task.Async && time.Since(task.Created) > task.Delay if shouldPop { ret = append(ret, task) // 不写入 taskQueue，相当于删除 } else { // 保留此任务，移动到 writeIdx 位置 if writeIdx != readIdx { taskQueue[writeIdx] = task } writeIdx++ } }
// 清理队列尾部的引用，防止内存泄漏 for i := writeIdx; i < len(taskQueue); i++ { taskQueue[i] = nil } taskQueue = taskQueue[:writeIdx] return}
var ( currentTask *Task currentTaskLock = sync.Mutex{})
func execTask(task *Task) { if nil == task { return }
defer logging.Recover()
args := make([]reflect.Value, len(task.Args)) for i, v := range task.Args { if nil == v { args[i] = reflect.New(task.Handler.Type().In(i)).Elem() } else { args[i] = reflect.ValueOf(v) } }
if !task.Async { currentTaskLock.Lock() currentTask = task currentTaskLock.Unlock() }
ctx, cancel := context.WithTimeout(context.Background(), task.Timeout) defer cancel() ch := make(chan bool, 1) go func() { task.Handler.Call(args) ch <- true }()
select { case <-ctx.Done(): logging.LogWarnf("task [%s] timeout", task.Action) case <-ch: //logging.LogInfof("task [%s] done", task.Action) }
if !task.Async { currentTaskLock.Lock() currentTask = nil currentTaskLock.Unlock() }}