diff --git a/go.mod b/go.mod index c82ae61..dfbb839 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index a3876b9..5ae943c 100644 --- a/go.sum +++ b/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.0-20230731165309-2d8b1969ccb7/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/474420502/execute v0.0.2 h1:CEhTCBa7UemBWzbfQ6Szj0/GJfx3/JJA0s6r807NPtw= +github.com/474420502/execute v0.0.2/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= diff --git a/main.go b/main.go index 12eb396..ddb37e5 100644 --- a/main.go +++ b/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 Consumption execute.Event +var BatchQueueExecute = triggered.NewEventTriggeredExecute[bool]() +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) { - 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) + BatchQueueExecute.SetShared(Consumption, nh) if err := nh.StartReplica(initialMembers, false, NewSMQueue, rc); err != nil { fmt.Fprintf(os.Stderr, "failed to add cluster, %v\n", err) diff --git a/sm.go b/sm.go index 79b72a7..05cea8a 100644 --- a/sm.go +++ b/sm.go @@ -20,8 +20,6 @@ type SMQueue struct { // Replica ID replicaID uint64 - counter *batchexecute.BatchExecute[int64] - // 互斥锁,保护队列Map的并发访问 mu *sync.Mutex // 组名到队列的映射