♻️ Implement some delayed kernel events using task queues https://github.com/siyuan-note/siyuan/issues/12393

This commit is contained in:
Daniel 2024-09-05 12:10:42 +08:00
parent e55c12037a
commit 9560277d0c
No known key found for this signature in database
GPG key ID: 86211BA83DF03017
7 changed files with 105 additions and 65 deletions

View file

@ -212,7 +212,7 @@ func addAttributeViewBlocks(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func removeAttributeViewBlocks(c *gin.Context) { func removeAttributeViewBlocks(c *gin.Context) {
@ -237,7 +237,7 @@ func removeAttributeViewBlocks(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func addAttributeViewKey(c *gin.Context) { func addAttributeViewKey(c *gin.Context) {
@ -263,7 +263,7 @@ func addAttributeViewKey(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func removeAttributeViewKey(c *gin.Context) { func removeAttributeViewKey(c *gin.Context) {
@ -285,7 +285,7 @@ func removeAttributeViewKey(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func sortAttributeViewViewKey(c *gin.Context) { func sortAttributeViewViewKey(c *gin.Context) {
@ -312,7 +312,7 @@ func sortAttributeViewViewKey(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func sortAttributeViewKey(c *gin.Context) { func sortAttributeViewKey(c *gin.Context) {
@ -335,7 +335,7 @@ func sortAttributeViewKey(c *gin.Context) {
return return
} }
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }
func getAttributeViewFilterSort(c *gin.Context) { func getAttributeViewFilterSort(c *gin.Context) {
@ -609,5 +609,5 @@ func setAttributeViewBlockAttr(c *gin.Context) {
blockAttributeViewKeys := model.UpdateAttributeViewCell(nil, avID, keyID, rowID, cellID, value) blockAttributeViewKeys := model.UpdateAttributeViewCell(nil, avID, keyID, rowID, cellID, value)
ret.Data = blockAttributeViewKeys ret.Data = blockAttributeViewKeys
util.PushReloadAttrView(avID) model.ReloadAttrView(avID)
} }

View file

@ -19,6 +19,7 @@ package model
import ( import (
"bytes" "bytes"
"fmt" "fmt"
"github.com/siyuan-note/siyuan/kernel/task"
"os" "os"
"path/filepath" "path/filepath"
"slices" "slices"
@ -92,7 +93,7 @@ func AppendAttributeViewDetachedBlocksWithValues(avID string, blocksValues [][]*
return return
} }
util.PushReloadAttrView(avID) ReloadAttrView(avID)
return return
} }
@ -1044,7 +1045,7 @@ func unbindAttributeViewBlock(operation *Operation, tx *Transaction) (err error)
changedAvIDs = gulu.Str.RemoveDuplicatedElem(changedAvIDs) changedAvIDs = gulu.Str.RemoveDuplicatedElem(changedAvIDs)
for _, avID := range changedAvIDs { for _, avID := range changedAvIDs {
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
return return
} }
@ -1313,7 +1314,7 @@ func updateAttributeViewColRelation(operation *Operation) (err error) {
} }
if !isSameAv { if !isSameAv {
err = av.SaveAttributeView(destAv) err = av.SaveAttributeView(destAv)
util.PushReloadAttrView(destAv.ID) ReloadAttrView(destAv.ID)
} }
av.UpsertAvBackRel(srcAv.ID, destAv.ID) av.UpsertAvBackRel(srcAv.ID, destAv.ID)
@ -2147,7 +2148,7 @@ func removeAttributeViewBlock(srcIDs []string, avID string, tx *Transaction) (er
relatedAvIDs := av.GetSrcAvIDs(avID) relatedAvIDs := av.GetSrcAvIDs(avID)
for _, relatedAvID := range relatedAvIDs { for _, relatedAvID := range relatedAvIDs {
util.PushReloadAttrView(relatedAvID) ReloadAttrView(relatedAvID)
} }
err = av.SaveAttributeView(attrView) err = av.SaveAttributeView(attrView)
@ -2804,7 +2805,7 @@ func RemoveAttributeViewKey(avID, keyID string) (err error) {
if destAv != attrView { if destAv != attrView {
av.SaveAttributeView(destAv) av.SaveAttributeView(destAv)
util.PushReloadAttrView(destAv.ID) ReloadAttrView(destAv.ID)
} }
if !destAvRelSrcAv { if !destAvRelSrcAv {
@ -2940,7 +2941,7 @@ func replaceAttributeViewBlock(operation *Operation, tx *Transaction) (err error
changedAvIDs = gulu.Str.RemoveDuplicatedElem(changedAvIDs) changedAvIDs = gulu.Str.RemoveDuplicatedElem(changedAvIDs)
for _, avID := range changedAvIDs { for _, avID := range changedAvIDs {
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
return return
} }
@ -3182,7 +3183,7 @@ func UpdateAttributeViewCell(tx *Transaction, avID, keyID, rowID, cellID string,
relatedAvIDs := av.GetSrcAvIDs(avID) relatedAvIDs := av.GetSrcAvIDs(avID)
for _, relatedAvID := range relatedAvIDs { for _, relatedAvID := range relatedAvIDs {
util.PushReloadAttrView(relatedAvID) ReloadAttrView(relatedAvID)
} }
if err = av.SaveAttributeView(attrView); err != nil { if err = av.SaveAttributeView(attrView); err != nil {
@ -3579,3 +3580,12 @@ func updateBoundBlockAvsAttribute(avIDs []string) {
av.BatchUpsertBlockRel(avNodes) av.BatchUpsertBlockRel(avNodes)
} }
} }
func ReloadAttrView(avID string) {
task.AppendTaskWithDelay(task.ReloadAttributeView, 200*time.Millisecond, pushReloadAttrView, avID)
}
func pushReloadAttrView(avID string) {
util.BroadcastByType("protyle", "refreshAttributeView", 0, "", map[string]interface{}{"id": avID})
}

View file

@ -1113,7 +1113,7 @@ func DuplicateDoc(tree *parse.Tree) {
"id": n.ID, "id": n.ID,
"isDetached": false, "isDetached": false,
}}, avID, "", "", false) }}, avID, "", "", false)
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
return ast.WalkContinue return ast.WalkContinue
}) })

View file

@ -312,7 +312,7 @@ func RollbackDocHistory(boxID, historyPath string) (err error) {
// 刷新属性视图 // 刷新属性视图
for _, avID := range avIDs { for _, avID := range avIDs {
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
}() }()
return nil return nil

View file

@ -813,7 +813,7 @@ func syncDelete2AvBlock(node *ast.Node) {
go func() { go func() {
time.Sleep(256 * time.Millisecond) time.Sleep(256 * time.Millisecond)
for _, avID := range changedAvIDs { for _, avID := range changedAvIDs {
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
}() }()
} }
@ -1074,7 +1074,7 @@ func (tx *Transaction) doInsert(operation *Operation) (ret *TxErr) {
"id": insertedNode.ID, "id": insertedNode.ID,
"isDetached": false, "isDetached": false,
}}, avID, "", previousID, false) }}, avID, "", previousID, false)
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
operation.ID = insertedNode.ID operation.ID = insertedNode.ID
@ -1187,7 +1187,7 @@ func upsertAvBlockRel(node *ast.Node) {
}) })
avIDs = gulu.Str.RemoveDuplicatedElem(avIDs) avIDs = gulu.Str.RemoveDuplicatedElem(avIDs)
for _, avID := range avIDs { for _, avID := range avIDs {
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
} }
@ -1533,7 +1533,7 @@ func refreshDynamicRefTexts(updatedDefNodes map[string]*ast.Node, updatedTrees m
} }
if changedAv { if changedAv {
av.SaveAttributeView(attrView) av.SaveAttributeView(attrView)
util.PushReloadAttrView(avID) ReloadAttrView(avID)
} }
} }
} }

View file

@ -37,50 +37,80 @@ type Task struct {
Handler reflect.Value Handler reflect.Value
Args []interface{} Args []interface{}
Created time.Time Created time.Time
Delay time.Duration
Timeout time.Duration Timeout time.Duration
} }
func AppendTask(action string, handler interface{}, args ...interface{}) { func AppendTask(action string, handler interface{}, args ...interface{}) {
AppendTaskWithTimeout(action, 24*time.Hour, handler, args...) appendTaskWithDelayTimeout(action, 0, 24*time.Hour, handler, args...)
}
func AppendTaskWithDelay(action string, delay time.Duration, handler interface{}, args ...interface{}) {
appendTaskWithDelayTimeout(action, delay, 24*time.Hour, handler, args...)
} }
func AppendTaskWithTimeout(action string, timeout time.Duration, handler interface{}, args ...interface{}) { func AppendTaskWithTimeout(action string, timeout time.Duration, handler interface{}, args ...interface{}) {
appendTaskWithDelayTimeout(action, 0, timeout, handler, args...)
}
func appendTaskWithDelayTimeout(action string, delay, timeout time.Duration, handler interface{}, args ...interface{}) {
if util.IsExiting.Load() { if util.IsExiting.Load() {
//logging.LogWarnf("task queue is paused, action [%s] will be ignored", action) //logging.LogWarnf("task queue is paused, action [%s] will be ignored", action)
return return
} }
currentActions := getCurrentActions() task := &Task{
if gulu.Str.Contains(action, currentActions) && gulu.Str.Contains(action, uniqueActions) { Action: action,
Handler: reflect.ValueOf(handler),
Args: args,
Created: time.Now(),
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) //logging.LogWarnf("task [%s] is already in queue, will be ignored", action)
return return
} }
}
queueLock.Lock() queueLock.Lock()
defer queueLock.Unlock() defer queueLock.Unlock()
taskQueue = append(taskQueue, &Task{ taskQueue = append(taskQueue, task)
Action: action,
Timeout: timeout,
Handler: reflect.ValueOf(handler),
Args: args,
Created: time.Now(),
})
} }
func getCurrentActions() (ret []string) { func containTask(task *Task, tasks []*Task) bool {
queueLock.Lock() for _, t := range tasks {
if t.Action == task.Action {
currentTaskActionLock.Lock() if len(t.Args) != len(task.Args) {
if "" != currentTaskAction { return false
ret = append(ret, currentTaskAction)
} }
currentTaskActionLock.Unlock()
for i, arg := range t.Args {
if arg != task.Args[i] {
return false
}
}
return true
}
}
return false
}
func getCurrentTasks() (ret []*Task) {
queueLock.Lock()
defer queueLock.Unlock()
currentTaskLock.Lock()
if nil != currentTask {
ret = append(ret, currentTask)
}
currentTaskLock.Unlock()
for _, task := range taskQueue { for _, task := range taskQueue {
ret = append(ret, task.Action) ret = append(ret, task)
} }
queueLock.Unlock()
return return
} }
@ -100,6 +130,7 @@ const (
AssetContentDatabaseIndexFull = "task.asset.database.index.full" // 资源文件数据库重建索引 AssetContentDatabaseIndexFull = "task.asset.database.index.full" // 资源文件数据库重建索引
AssetContentDatabaseIndexCommit = "task.asset.database.index.commit" // 资源文件数据库索引提交 AssetContentDatabaseIndexCommit = "task.asset.database.index.commit" // 资源文件数据库索引提交
CacheVirtualBlockRef = "task.cache.virtualBlockRef" // 缓存虚拟块引用 CacheVirtualBlockRef = "task.cache.virtualBlockRef" // 缓存虚拟块引用
ReloadAttributeView = "task.reload.attributeView" // 重新加载属性视图
) )
// uniqueActions 描述了唯一的任务,即队列中只能存在一个在执行的任务。 // uniqueActions 描述了唯一的任务,即队列中只能存在一个在执行的任务。
@ -116,9 +147,9 @@ var uniqueActions = []string{
} }
func ContainIndexTask() bool { func ContainIndexTask() bool {
actions := getCurrentActions() tasks := getCurrentTasks()
for _, action := range actions { for _, task := range tasks {
if gulu.Str.Contains(action, []string{DatabaseIndexFull, DatabaseIndex}) { if gulu.Str.Contains(task.Action, []string{DatabaseIndexFull, DatabaseIndex}) {
return true return true
} }
} }
@ -150,15 +181,13 @@ func StatusJob() {
} }
defer queueLock.Unlock() defer queueLock.Unlock()
currentTaskActionLock.Lock() currentTaskLock.Lock()
if "" != currentTaskAction { if nil != currentTask && nil != actionLangs {
if nil != actionLangs { if label := actionLangs[currentTask.Action]; nil != label {
if label := actionLangs[currentTaskAction]; nil != label {
items = append([]map[string]interface{}{{"action": label.(string)}}, items...) items = append([]map[string]interface{}{{"action": label.(string)}}, items...)
} }
} }
} currentTaskLock.Unlock()
currentTaskActionLock.Unlock()
if 1 > len(items) { if 1 > len(items) {
items = []map[string]interface{}{} items = []map[string]interface{}{}
@ -189,14 +218,19 @@ func popTask() (ret *Task) {
return return
} }
ret = taskQueue[0] for i, task := range taskQueue {
taskQueue = taskQueue[1:] if time.Since(task.Created) > task.Delay {
ret = task
taskQueue = append(taskQueue[:i], taskQueue[i+1:]...)
return
}
}
return return
} }
var ( var (
currentTaskAction string currentTask *Task
currentTaskActionLock = sync.Mutex{} currentTaskLock = sync.Mutex{}
) )
func execTask(task *Task) { func execTask(task *Task) {
@ -211,9 +245,9 @@ func execTask(task *Task) {
} }
} }
currentTaskActionLock.Lock() currentTaskLock.Lock()
currentTaskAction = task.Action currentTask = task
currentTaskActionLock.Unlock() currentTaskLock.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), task.Timeout) ctx, cancel := context.WithTimeout(context.Background(), task.Timeout)
defer cancel() defer cancel()
@ -230,7 +264,7 @@ func execTask(task *Task) {
//logging.LogInfof("task [%s] done", task.Action) //logging.LogInfof("task [%s] done", task.Action)
} }
currentTaskActionLock.Lock() currentTaskLock.Lock()
currentTaskAction = "" currentTask = nil
currentTaskActionLock.Unlock() currentTaskLock.Unlock()
} }

View file

@ -240,10 +240,6 @@ func PushClearProgress() {
BroadcastByType("main", "cprogress", 0, "", nil) BroadcastByType("main", "cprogress", 0, "", nil)
} }
func PushReloadAttrView(avID string) {
BroadcastByType("protyle", "refreshAttributeView", 0, "", map[string]interface{}{"id": avID})
}
func PushReloadDoc(rootID string) { func PushReloadDoc(rootID string) {
BroadcastByType("main", "reloaddoc", 0, "", rootID) BroadcastByType("main", "reloaddoc", 0, "", rootID)
} }