TODO: 周期性处理

go.mod

@@ -3,13 +3,14 @@ module fusenrender
 go 1.20
 
 require (
+	github.com/474420502/batchexecute v0.0.2
+	github.com/474420502/execute v0.0.2
+	github.com/google/uuid v1.3.0
 	github.com/lni/dragonboat/v4 v4.0.0-20230709075559-54497b9553be
 	github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4
 )
 
 require (
-	github.com/474420502/batchexecute v0.0.2 // indirect
-	github.com/474420502/execute v0.0.1 // indirect
 	github.com/DataDog/zstd v1.4.5 // indirect
 	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
 	github.com/VictoriaMetrics/metrics v1.18.1 // indirect
@@ -22,7 +23,6 @@ require (
 	github.com/getsentry/sentry-go v0.12.0 // indirect
 	github.com/google/btree v1.0.0 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
-	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-immutable-radix v1.0.0 // indirect
 	github.com/hashicorp/go-msgpack v0.5.3 // indirect

go.sum

@@ -2,10 +2,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/474420502/batchexecute v0.0.2 h1:gTMyUh2x6HpV7/+nOLj4qFRO/l5/B/eg7dXsHxLJpbg=
 github.com/474420502/batchexecute v0.0.2/go.mod h1:IWazO1QTaB5LyWwMxSqIX/6g/UXwwpnqk0AVM5j24J0=
-github.com/474420502/execute v0.0.0-20230731165309-2d8b1969ccb7 h1:U0jTaGJGWzJM/mZSN6L8rBfACry8+xJeHRX9t99xxfE=
+github.com/474420502/execute v0.0.2 h1:CEhTCBa7UemBWzbfQ6Szj0/GJfx3/JJA0s6r807NPtw=
-github.com/474420502/execute v0.0.0-20230731165309-2d8b1969ccb7/go.mod h1:3/9VeQxxZx5lAqz1UWEolgp0OK3hautW9jhMyReuWZc=
+github.com/474420502/execute v0.0.2/go.mod h1:3/9VeQxxZx5lAqz1UWEolgp0OK3hautW9jhMyReuWZc=
-github.com/474420502/execute v0.0.1 h1:DCC7RiS+R+fZyw9cpYgCmGfC5iyBljdfpolSN20yVWo=
-github.com/474420502/execute v0.0.1/go.mod h1:3/9VeQxxZx5lAqz1UWEolgp0OK3hautW9jhMyReuWZc=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=

main.go

@@ -12,7 +12,7 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/474420502/execute"
+	"github.com/474420502/execute/triggered"
 	"github.com/lni/dragonboat/v4"
 	"github.com/lni/dragonboat/v4/config"
 	"github.com/lni/dragonboat/v4/logger"
@@ -22,8 +22,54 @@ func main() {
 
 }
 
-var BatchQueueExecute = execute.NewEventTriggeredExecute[bool]()
+var BatchQueueExecute = triggered.NewEventTriggeredExecute[bool]()
-var Consumption execute.Event
+var Consumption triggered.Event = BatchQueueExecute.RegisterExecute(func(params *triggered.Params[bool]) {
+	nh := params.Shared.(*dragonboat.NodeHost)
+	cs := nh.GetNoOPSession(128)
+	for {
+
+		cmd := Command{
+			Name:  "dequeue",
+			Group: "test",
+		}
+
+		data, err := cmd.Encode()
+		if err != nil {
+			log.Println(err)
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		result, err := nh.SyncPropose(ctx, cs, data)
+		cancel()
+		if err != nil {
+			log.Println(err)
+			break
+		} else {
+			if len(result.Data) == 0 {
+				log.Println("wait 10 second")
+				var m runtime.MemStats
+
+				runtime.ReadMemStats(&m)
+
+				allocMB := float64(m.Alloc) / 1024 / 1024
+				totalAllocMB := float64(m.TotalAlloc) / 1024 / 1024
+				sysMB := float64(m.Sys) / 1024 / 1024
+
+				fmt.Printf("Alloc = %.2f MB, TotalAlloc = %.2f MB, Sys = %.2f MB\n", allocMB, totalAllocMB, sysMB)
+				time.Sleep(time.Second * 10)
+				break
+			} else {
+				var item QueueItem
+				err := item.Decode(result.Data)
+				if err != nil {
+					log.Println(err)
+				}
+				log.Println(item)
+
+			}
+		}
+	}
+}, nil)
 
 var addresses []string = []string{
 	"localhost:5500",
@@ -112,52 +158,7 @@ func StartNode(replicaID uint64, exampleShardID uint64, addr string) *dragonboat
 		panic(err)
 	}
 
-	Consumption = BatchQueueExecute.RegisterExecute(func(*bool) {
+	BatchQueueExecute.SetShared(Consumption, nh)
-		cs := nh.GetNoOPSession(128)
-		for {
-
-			cmd := Command{
-				Name:  "dequeue",
-				Group: "test",
-			}
-
-			data, err := cmd.Encode()
-			if err != nil {
-				log.Println(err)
-			}
-
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			result, err := nh.SyncPropose(ctx, cs, data)
-			cancel()
-			if err != nil {
-				log.Println(err)
-				break
-			} else {
-				if len(result.Data) == 0 {
-					log.Println("wait 10 second")
-					var m runtime.MemStats
-
-					runtime.ReadMemStats(&m)
-
-					allocMB := float64(m.Alloc) / 1024 / 1024
-					totalAllocMB := float64(m.TotalAlloc) / 1024 / 1024
-					sysMB := float64(m.Sys) / 1024 / 1024
-
-					fmt.Printf("Alloc = %.2f MB, TotalAlloc = %.2f MB, Sys = %.2f MB\n", allocMB, totalAllocMB, sysMB)
-					time.Sleep(time.Second * 10)
-					break
-				} else {
-					var item QueueItem
-					err := item.Decode(result.Data)
-					if err != nil {
-						log.Println(err)
-					}
-					log.Println(item)
-
-				}
-			}
-		}
-	}, nil)
 
 	if err := nh.StartReplica(initialMembers, false, NewSMQueue, rc); err != nil {
 		fmt.Fprintf(os.Stderr, "failed to add cluster, %v\n", err)

sm.go

@@ -20,8 +20,6 @@ type SMQueue struct {
 	// Replica ID
 	replicaID uint64
 
-	counter *batchexecute.BatchExecute[int64]
-
 	// 互斥锁,保护队列Map的并发访问
 	mu *sync.Mutex
 	// 组名到队列的映射