From e626cee8f7de0dd3ffca4c7297e59a87f790cc02 Mon Sep 17 00:00:00 2001 From: sangtran Date: Sat, 5 Sep 2026 17:06:30 +0700 Subject: feat: add on pebble state backend and using sync Pool to reuse buffer []byte --- README.md | 7 +++++++ core/pool.go | 38 ++++++++++++++++++++++++++++++++++ core/state_backend.go | 18 +++++++++++++++-- go.mod | 2 +- state/memory_backend.go | 8 ++++---- state/pebble_backend.go | 30 +++++++++++++++------------ state/pebble_state.go | 54 +++++++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 137 insertions(+), 20 deletions(-) create mode 100644 README.md create mode 100644 core/pool.go create mode 100644 state/pebble_state.go diff --git a/README.md b/README.md new file mode 100644 index 0000000..a180257 --- /dev/null +++ b/README.md @@ -0,0 +1,7 @@ +Flux +==== + +Flux is an event-driven stream processing engine written in Go, +designed from first principles to handle stateful stream computation, +out-of-order event time processing, high-throughput windowing, +and distributed fault-tolerant state backend. diff --git a/core/pool.go b/core/pool.go new file mode 100644 index 0000000..484b342 --- /dev/null +++ b/core/pool.go @@ -0,0 +1,38 @@ +package core + +import "sync" + + +const ( + DefaultBufferSize = 1024 + MaxBufferSize = 64 * 1024 +) + +var bytesPool = sync.Pool{ + New: func () any { + b := make([]byte, DefaultBufferSize) + return &b + }, +} + +func GetBuffer(size int) []byte { + bufPtr := bytesPool.Get().(*[]byte) + + buf := *bufPtr + + + if size > cap(buf) { + buf = make([]byte, size) + } + + return buf[:size] +} + + +func PutBuffer(buf []byte) { + if cap(buf) < MaxBufferSize{ + // reset + buf = buf[:0] + bytesPool.Put(&buf) + } +} diff --git a/core/state_backend.go b/core/state_backend.go index 5870739..ae9fe1d 100644 --- a/core/state_backend.go +++ b/core/state_backend.go @@ -2,6 +2,9 @@ package core import ( "context" + "encoding/json" + "fmt" + ) type StateBackend interface { @@ -22,6 +25,7 @@ type stateBackendKeyType struct{} var stateBackendKey stateBackendKeyType +// type adapter type typedValueState[T any] struct { inner ValueState[any] } @@ -32,14 +36,24 @@ func CastValueType[T any](inner ValueState[any]) ValueState[T] { func (s *typedValueState[T]) Value(ctx context.Context) (T, error) { val, err := s.inner.Value(ctx) + var zero T if err != nil { - var zero T return zero, err } if val == nil { - var zero T return zero, nil } + + if rawBytes, ok := val.([]byte); ok { + defer PutBuffer(rawBytes) + + var result T + if err := json.Unmarshal(rawBytes, &result); err != nil { + return zero, fmt.Errorf("failed to unmarshal state: %w", err) + } + return result, nil + } + return val.(T), nil } diff --git a/go.mod b/go.mod index df5991f..a99ec15 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module flux -go 1.26.3 +go 1.27 require github.com/cockroachdb/pebble/v2 v2.1.7 diff --git a/state/memory_backend.go b/state/memory_backend.go index dda577e..cce8194 100644 --- a/state/memory_backend.go +++ b/state/memory_backend.go @@ -39,15 +39,15 @@ func (ms *MemoryStateBackend) SetCurrentKey(key string) error { return nil } -func (ms *MemoryStateBackend) GetValueState(ctx context.Context, statename 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[statename]; !ok { + if _, ok := ms.states[stateName]; !ok { // create state name map if not existed - ms.states[statename] = make(map[int]map[string]map[any]any) + ms.states[stateName] = make(map[int]map[string]map[any]any) } - return &memoryStateValue{backend: ms, stateName: statename}, nil + return &memoryStateValue{backend: ms, stateName: stateName}, nil } func (ms *MemoryStateBackend) GetCurrentNamespace() any { diff --git a/state/pebble_backend.go b/state/pebble_backend.go index e428988..fbb8d68 100644 --- a/state/pebble_backend.go +++ b/state/pebble_backend.go @@ -33,10 +33,11 @@ func NewPebbleStateBackend(dir string, maxParalleism int) (*PebbleStateBackend, return nil, fmt.Errorf("failed to open pebbledb %w", err) } return &PebbleStateBackend{ - maxParallelism: maxParalleism, - db: db, - stateMap: make(map[string]uint8), - nextStateID: 1, + maxParallelism: maxParalleism, + db: db, + stateMap: make(map[string]uint8), + nextStateID: 1, + currentNamespace: &core.VoidNamespace{}, }, nil } @@ -60,23 +61,27 @@ func (p *PebbleStateBackend) GetCurrentNamespace() any { } // GetValueState(ctx context.Context, key string) (ValueState[any], error) -func (p *PebbleStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) { - stateID := p.getStateID(statename) - return nil, nil +func (p *PebbleStateBackend) GetValueState(ctx context.Context, stateName string) (core.ValueState[any], error) { + stateID := p.getStateID(stateName) + + return &pebbleStateValue{ + backend: p, + stateID: stateID, + }, nil } -func (p *PebbleStateBackend) getStateID(statename string) uint8 { +func (p *PebbleStateBackend) getStateID(stateName string) uint8 { p.mu.Lock() defer p.mu.Unlock() - if _, ok := p.stateMap[statename]; !ok { + if _, ok := p.stateMap[stateName]; !ok { id := p.nextStateID - p.stateMap[statename] = id + p.stateMap[stateName] = id p.nextStateID++ } - return p.stateMap[statename] + return p.stateMap[stateName] } func (p *PebbleStateBackend) calculateNamespaceSize() int { @@ -94,7 +99,7 @@ func (p *PebbleStateBackend) calculateNamespaceSize() int { return nsLen } -func (p *PebbleStateBackend) deserialize(statename string) []byte { +func (p *PebbleStateBackend) deserialize(stateID uint8) []byte { // this will store in SSTable, order is important, keygroup MUST be first for scaling // 2B(KeyGroup) + 1B(StateID) + 2B(KeyLen) + len(Key) + 16B(Window) @@ -105,7 +110,6 @@ func (p *PebbleStateBackend) deserialize(statename string) []byte { b = binary.BigEndian.AppendUint16(b, uint16(p.currentGroupKey)) // add state id - stateID := p.getStateID(statename) b = append(b, stateID) // add key diff --git a/state/pebble_state.go b/state/pebble_state.go new file mode 100644 index 0000000..3eeaa05 --- /dev/null +++ b/state/pebble_state.go @@ -0,0 +1,54 @@ +package state + +import ( + "context" + "encoding/json" + "errors" + "flux/core" + + "github.com/cockroachdb/pebble/v2" +) + +type pebbleStateValue struct { + backend *PebbleStateBackend + stateID uint8 +} + +func (p *pebbleStateValue) Update(ctx context.Context, newVal any) (any, error) { + key := p.backend.deserialize(p.stateID) + // convert newVal to []byte raw + v, err := json.Marshal(newVal) + if err != nil { + return nil, err + } + + err = p.backend.db.Set(key, v, pebble.NoSync) + if err != nil { + return nil, err + } + + return newVal, nil +} + +func (p *pebbleStateValue) Value(ctx context.Context) (any, error) { + key := p.backend.deserialize(p.stateID) + v, closer, err := p.backend.db.Get(key) + if err != nil { + if errors.Is(err, pebble.ErrNotFound) { + return nil, nil + } + return nil, err + } + + defer closer.Close() + + res := core.GetBuffer(len(v)) + copy(res, v) + return v, nil +} + +func (p *pebbleStateValue) Clear(ctx context.Context) error { + key := p.backend.deserialize(p.stateID) + + return p.backend.db.Delete(key, pebble.NoSync) +} -- cgit v1.3.1