Andreas/siyuan
1
0
Fork 0
mirror of https://github.com/siyuan-note/siyuan.git synced 2025-12-17 07:00:12 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

🎨 改进内核任务调度机制提升稳定性 https://github.com/siyuan-note/siyuan/issues/7113

Browse source
This commit is contained in:
Liang Ding 2023-01-23 18:30:52 +08:00
parent a83a08bdb5
commit 72093f7e2e
No known key found for this signature in database
GPG key ID: 136F30F901A2231D
7 changed files with 103 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

121
kernel/sql/queue.go
View file

@ -22,7 +22,6 @@ import (
"time"
"github.com/88250/lute/parse"
"github.com/emirpasic/gods/sets/hashset"
"github.com/siyuan-note/eventbus"
"github.com/siyuan-note/logging"
"github.com/siyuan-note/siyuan/kernel/task"
@ -30,19 +29,20 @@ import (
)
var (
operationQueue []*treeQueueOperation
upsertTreeQueueLock = sync.Mutex{}
operationQueue []*dbQueueOperation
dbQueueLock = sync.Mutex{}
txLock = sync.Mutex{}
)
type treeQueueOperation struct {
type dbQueueOperation struct {
inQueueTime time.Time
action string // upsert/delete/delete_id/rename
action string // upsert/delete/delete_id/rename/delete_box/delete_box_refs/insert_refs
upsertTree *parse.Tree // upsert
upsertTree *parse.Tree // upsert/insert_refs
removeTreeBox, removeTreePath string // delete
removeTreeIDBox, removeTreeID string // delete_id
box string // delete_box/delete_box_refs
renameTree *parse.Tree // rename
renameTreeOldHPath string // rename
}
@ -83,8 +83,8 @@ func IsEmptyQueue() bool {
}
func ClearQueue() {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
operationQueue = nil
}
@ -103,7 +103,6 @@ func FlushQueue() {
}
context := map[string]interface{}{eventbus.CtxPushMsg: eventbus.CtxPushMsgToStatusBar}
boxes := hashset.New()
for _, op := range ops {
switch op.action {
case "upsert":
@ -111,17 +110,21 @@ func FlushQueue() {
if err = upsertTree(tx, tree, context); nil != err {
logging.LogErrorf("upsert tree [%s] into database failed: %s", tree.Box+tree.Path, err)
}
boxes.Add(op.upsertTree.Box)
case "delete":
batchDeleteByPathPrefix(tx, op.removeTreeBox, op.removeTreePath)
boxes.Add(op.removeTreeBox)
case "delete_id":
DeleteByRootID(tx, op.removeTreeID)
boxes.Add(op.removeTreeIDBox)
case "rename":
batchUpdateHPath(tx, op.renameTree.Box, op.renameTree.ID, op.renameTreeOldHPath, op.renameTree.HPath)
updateRootContent(tx, path.Base(op.renameTree.HPath), op.renameTree.Root.IALAttr("updated"), op.renameTree.ID)
boxes.Add(op.renameTree.Box)
case "delete_box":
DeleteByBoxTx(tx, op.box)
case "delete_box_refs":
DeleteRefsByBoxTx(tx, op.box)
case "insert_refs":
InsertRefs(tx, op.upsertTree)
case "update_refs":
UpsertRefs(tx, op.upsertTree)
default:
logging.LogErrorf("unknown operation [%s]", op.action)
}
@ -133,20 +136,76 @@ func FlushQueue() {
}
}
func mergeUpsertTrees() (ops []*treeQueueOperation) {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
func mergeUpsertTrees() (ops []*dbQueueOperation) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
ops = operationQueue
operationQueue = nil
return
}
func UpsertTreeQueue(tree *parse.Tree) {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
func UpdateRefsTreeQueue(tree *parse.Tree) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &treeQueueOperation{upsertTree: tree, inQueueTime: time.Now(), action: "upsert"}
newOp := &dbQueueOperation{upsertTree: tree, inQueueTime: time.Now(), action: "update_refs"}
for i, op := range operationQueue {
if "update_refs" == op.action && op.upsertTree.ID == tree.ID {
operationQueue[i] = newOp
return
}
}
operationQueue = append(operationQueue, newOp)
}
func InsertRefsTreeQueue(tree *parse.Tree) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &dbQueueOperation{upsertTree: tree, inQueueTime: time.Now(), action: "insert_refs"}
for i, op := range operationQueue {
if "insert_refs" == op.action && op.upsertTree.ID == tree.ID {
operationQueue[i] = newOp
return
}
}
operationQueue = append(operationQueue, newOp)
}
func DeleteBoxRefsQueue(boxID string) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &dbQueueOperation{box: boxID, inQueueTime: time.Now(), action: "delete_box_refs"}
for i, op := range operationQueue {
if "delete_box_refs" == op.action && op.box == boxID {
operationQueue[i] = newOp
return
}
}
operationQueue = append(operationQueue, newOp)
}
func DeleteBoxQueue(boxID string) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &dbQueueOperation{box: boxID, inQueueTime: time.Now(), action: "delete_box"}
for i, op := range operationQueue {
if "delete_box" == op.action && op.box == boxID {
operationQueue[i] = newOp
return
}
}
operationQueue = append(operationQueue, newOp)
}
func UpsertTreeQueue(tree *parse.Tree) {
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &dbQueueOperation{upsertTree: tree, inQueueTime: time.Now(), action: "upsert"}
for i, op := range operationQueue {
if "upsert" == op.action && op.upsertTree.ID == tree.ID { // 相同树则覆盖
operationQueue[i] = newOp
@ -157,10 +216,10 @@ func UpsertTreeQueue(tree *parse.Tree) {
}
func RenameTreeQueue(tree *parse.Tree, oldHPath string) {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
newOp := &treeQueueOperation{
newOp := &dbQueueOperation{
renameTree: tree,
renameTreeOldHPath: oldHPath,
inQueueTime: time.Now(),
@ -175,10 +234,10 @@ func RenameTreeQueue(tree *parse.Tree, oldHPath string) {
}
func RemoveTreeQueue(box, rootID string) {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
var tmp []*treeQueueOperation
var tmp []*dbQueueOperation
// 将已有的 upsert 操作去重
for _, op := range operationQueue {
if "upsert" == op.action && op.upsertTree.ID != rootID {
@ -187,15 +246,15 @@ func RemoveTreeQueue(box, rootID string) {
}
operationQueue = tmp
newOp := &treeQueueOperation{removeTreeIDBox: box, removeTreeID: rootID, inQueueTime: time.Now(), action: "delete_id"}
newOp := &dbQueueOperation{removeTreeIDBox: box, removeTreeID: rootID, inQueueTime: time.Now(), action: "delete_id"}
operationQueue = append(operationQueue, newOp)
}
func RemoveTreePathQueue(treeBox, treePathPrefix string) {
upsertTreeQueueLock.Lock()
defer upsertTreeQueueLock.Unlock()
dbQueueLock.Lock()
defer dbQueueLock.Unlock()
var tmp []*treeQueueOperation
var tmp []*dbQueueOperation
// 将已有的 upsert 操作去重
for _, op := range operationQueue {
if "upsert" == op.action && (op.removeTreeBox != treeBox || op.upsertTree.Path != treePathPrefix) {
@ -204,6 +263,6 @@ func RemoveTreePathQueue(treeBox, treePathPrefix string) {
}
operationQueue = tmp
newOp := &treeQueueOperation{removeTreeBox: treeBox, removeTreePath: treePathPrefix, inQueueTime: time.Now(), action: "delete"}
newOp := &dbQueueOperation{removeTreeBox: treeBox, removeTreePath: treePathPrefix, inQueueTime: time.Now(), action: "delete"}
operationQueue = append(operationQueue, newOp)
}