From 6631d8b00e8c5bafea15ea24415c5b55439193ee Mon Sep 17 00:00:00 2001 From: sangtran Date: Mon, 31 Aug 2026 23:51:37 +0700 Subject: bug: fix memory_state clear and guard window state empty with implement main.go for test --- core/operator/window.go | 4 ++++ main.go | 55 +++++++++++++++++++++++++++++++++---------------- state/memory_backend.go | 21 ++++++++++--------- state/memory_state.go | 18 +++++++++++----- 4 files changed, 65 insertions(+), 33 deletions(-) diff --git a/core/operator/window.go b/core/operator/window.go index 610f6e3..3f75d60 100644 --- a/core/operator/window.go +++ b/core/operator/window.go @@ -79,6 +79,10 @@ func (o *WindowOperator[IN, OUT]) Process(ctx context.Context, record core.Strea return err } + if len(currentWindowState) == 0 { + continue + } + // NOTE: this is workaround create inline window, refactor later if err := o.userfunc.Apply(ctx, timer.Key, timer.Namespace.(core.Window), currentWindowState, out); err != nil { diff --git a/main.go b/main.go index 1b4a4be..91b96e3 100644 --- a/main.go +++ b/main.go @@ -19,24 +19,24 @@ type Transaction struct { Timestamp int64 } -type TransactionSummary[IN Transaction, OUT Transaction] struct { +type TransactionSummary[IN Transaction, OUT string] struct { totalTransaction int - totalAmount int average int } -func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out transport.Emitter[Transaction]) error { +func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out operator.Emitter[string]) error { + totalAmount := 0 for _, v := range in { - t.totalAmount += v.Amount - out.Emit(ctx, core.StreamRecord[Transaction]{ - Value: v, - Type: core.RecordData, - Timestamp: v.Timestamp, - }) + totalAmount += v.Amount } - return nil -} + msg := fmt.Sprintf("Khách: %s | Số đơn: %d | Tổng tiền: %d$", key, len(in), totalAmount) + + return out.Emit(ctx, core.StreamRecord[string]{ + Value: msg, + Type: core.RecordData, + Timestamp: window.MaxTimestamp(), + })} func main() { // init @@ -55,25 +55,45 @@ func main() { { User: "Sang", Amount: 4, - Timestamp: 1200, + Timestamp: 1500, + }, + { + User: "Sang", + Amount: 10, + Timestamp: 2000, + }, + { + User: "Sang", + Amount: 12, + Timestamp: 2500, }, { User: "Sang", Amount: 10, - Timestamp: 1600, + Timestamp: 3000, }, } ch1 := make(chan core.StreamRecord[Transaction], len(txList)) ch2 := make(chan core.StreamRecord[Transaction], len(txList)) + ch3 := make(chan core.StreamRecord[Transaction], len(txList)) + ch4 := make(chan core.StreamRecord[string], len(txList)) wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 { return in.Timestamp }) + ko := operator.NewKeyByOperator[Transaction](func(in Transaction) string { + return in.User + }) + + wop := operator.NewWindowOperator[Transaction, string](core.NewTumblingWindows[Transaction](1000), &TransactionSummary[Transaction, string]{}) + var wg sync.WaitGroup - runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo) + runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo) + runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch2), transport.NewLocalEmitChannel(ch3), ko) + runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch3), transport.NewLocalEmitChannel(ch4), wop) wg.Go(func() { for i, tx := range txList { @@ -84,13 +104,12 @@ func main() { }) wg.Go(func() { - for v := range ch2 { + for v := range ch4 { if v.Type == core.RecordData { - fmt.Println("data: ", v.Value) - } else { - fmt.Println("watermark: ", v.Timestamp) + fmt.Println("Total Amount after: ", v.Value) } } }) + wg.Wait() } diff --git a/state/memory_backend.go b/state/memory_backend.go index 6594657..f4043a1 100644 --- a/state/memory_backend.go +++ b/state/memory_backend.go @@ -2,6 +2,7 @@ package state import ( "context" + "flux/core" ) @@ -11,9 +12,9 @@ import ( // This should be optimize with ZERO GC Scanning, Memory Alignment type MemoryStateBackend struct { - maxParallelism int - currentKey string - currentGroupKey int + maxParallelism int + currentKey string + currentGroupKey int currentNamespace any // 4D structure StateName -> KeyGroup -> Key -> Namespace -> Value states map[string]map[int]map[string]map[any]any @@ -36,18 +37,18 @@ func (ms *MemoryStateBackend) SetCurrentKey(key string) error { return nil } -func (ms *MemoryStateBackend) GetValueState(ctx context.Context, key string) (core.ValueState[any], error) { +func (ms *MemoryStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) { // Assign if state not existed - - if _, ok := ms.states[key]; !ok { - ms.states[key] = make(map[int]map[string]map[any]any) + + if _, ok := ms.states[statename]; !ok { + // create state name map if not existed + ms.states[statename] = make(map[int]map[string]map[any]any) } - return &memoryStateValue{backend: ms, stateName: key}, nil + return &memoryStateValue{backend: ms, stateName: statename}, nil } -func (ms *MemoryStateBackend) GetCurrentNamespace() any{ - +func (ms *MemoryStateBackend) GetCurrentNamespace() any { return ms.currentNamespace } diff --git a/state/memory_state.go b/state/memory_state.go index 95c7318..97ecad1 100644 --- a/state/memory_state.go +++ b/state/memory_state.go @@ -1,7 +1,7 @@ package state import ( -"context" + "context" ) type memoryStateValue struct { @@ -13,7 +13,7 @@ func (m *memoryStateValue) Value(ctx context.Context) (any, error) { kg := m.backend.currentGroupKey key := m.backend.currentKey ns := m.backend.currentNamespace - if kgMap, exist := m.backend.states[m.stateName][kg][key]; exist { + if kgMap, ok := m.backend.states[m.stateName][kg][key]; ok { if v, hasKey := kgMap[ns]; hasKey { return v, nil } @@ -26,10 +26,15 @@ func (m *memoryStateValue) Update(ctx context.Context, newVal any) (any, error) kg := m.backend.currentGroupKey key := m.backend.currentKey ns := m.backend.currentNamespace - if _, exists := m.backend.states[m.stateName][kg][key]; !exists { + + if _, ok := m.backend.states[m.stateName][kg]; !ok { m.backend.states[m.stateName][kg] = make(map[string]map[any]any) } + if _, ok := m.backend.states[m.stateName][kg][key]; !ok { + m.backend.states[m.stateName][kg][key] = make(map[any]any) + } + m.backend.states[m.stateName][kg][key][ns] = newVal return newVal, nil } @@ -37,8 +42,11 @@ func (m *memoryStateValue) Update(ctx context.Context, newVal any) (any, error) func (m *memoryStateValue) Clear(ctx context.Context) error { kg := m.backend.currentGroupKey key := m.backend.currentKey - if kgMap, exists := m.backend.states[m.stateName][kg]; exists { - delete(kgMap, key) + ns := m.backend.currentNamespace + if kgMap, ok := m.backend.states[m.stateName][kg]; ok { + if nsMap, ok := kgMap[key]; ok { + delete(nsMap, ns) + } } return nil } -- cgit v1.3.1