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

kernel/sql/queue.go

@@ -17,6 +17,9 @@
 package sql
 
 import (
+	"database/sql"
+	"errors"
+	"fmt"
 	"path"
 	"sync"
 	"time"
@@ -103,13 +106,47 @@ func FlushQueue() {
 		return
 	}
 
+	var execOps int
 	context := map[string]interface{}{eventbus.CtxPushMsg: eventbus.CtxPushMsgToStatusBar}
-	execOps := 0
 	for i, op := range ops {
 		if util.IsExiting {
-			break
+			return
 		}
 
+		err = execOp(op, tx, context)
+		execOps++
+
+		if nil != err {
+			logging.LogErrorf("queue operation failed: %s", err)
+			return
+		}
+
+		if 0 < i && 0 == execOps%64 {
+			if err = commitTx(tx); nil != err {
+				logging.LogErrorf("commit tx failed: %s", err)
+				return
+			}
+
+			execOps = 0
+			tx, err = beginTx()
+			if nil != err {
+				return
+			}
+		}
+	}
+
+	if 0 < execOps {
+		if err = commitTx(tx); nil != err {
+			logging.LogErrorf("commit tx failed: %s", err)
+		}
+	}
+	elapsed := time.Now().Sub(start).Milliseconds()
+	if 5000 < elapsed {
+		logging.LogInfof("op tx [%dms]", elapsed)
+	}
+}
+
+func execOp(op *dbQueueOperation, tx *sql.Tx, context map[string]interface{}) (err error) {
 	switch op.action {
 	case "index":
 		err = indexTree(tx, op.box, op.indexPath, context)
@@ -134,39 +171,11 @@ func FlushQueue() {
 	case "update_refs":
 		err = upsertRefs(tx, op.upsertTree)
 	default:
-			logging.LogErrorf("unknown operation [%s]", op.action)
+		msg := fmt.Sprint("unknown operation [%s]", op.action)
-			break
+		logging.LogErrorf(msg)
-		}
+		err = errors.New(msg)
-
-		execOps++
-		if nil != err {
-			logging.LogErrorf("queue operation failed: %s", err)
-			break
-		}
-
-		if 0 < i && 0 == i%64 {
-			if err = commitTx(tx); nil != err {
-				logging.LogErrorf("commit tx failed: %s", err)
-				break
-			}
-
-			execOps = 0
-			tx, err = beginTx()
-			if nil != err {
-				break
-			}
-		}
-	}
-
-	if 0 < execOps {
-		if err = commitTx(tx); nil != err {
-			logging.LogErrorf("commit tx failed: %s", err)
-		}
-	}
-	elapsed := time.Now().Sub(start).Milliseconds()
-	if 5000 < elapsed {
-		logging.LogInfof("op tx [%dms]", elapsed)
 	}
+	return
 }
 
 func mergeUpsertTrees() (ops []*dbQueueOperation) {