diff --git a/docs/pages/stages/index.md b/docs/pages/stages/index.md index e05d6c2..9a57ce0 100644 --- a/docs/pages/stages/index.md +++ b/docs/pages/stages/index.md @@ -7,7 +7,7 @@ icon: lucide/library Goccia ships stages in three categories: - [Ingress](ingress/index.md): create messages from external sources. -- [Processor](processor/index.md): transform, filter, fan out, or reorder messages. +- [Processor](processor/index.md): transform, filter, fan in, fan out, or reorder messages. - [Egress](egress/index.md): deliver messages to external destinations. ## Message Interfaces @@ -39,6 +39,7 @@ cfg := processor.NewGenericConfig(goccia.StageRunningModePool) `StageRunningModeSingle` runs one executor. `StageRunningModePool` enables the generic worker pool and exposes `cfg.Stage.Pool` for worker counts, queue sizes, and auto-scaling. Stages with custom runners, such as ingress stages, `tee`, -`rob`, `sink`, file egress, and TCP egress, document their own execution model. +`merge`, `rob`, `sink`, file egress, and TCP egress, document their own +execution model. For lifecycle details, see [Stages](../concepts/stages.md). diff --git a/docs/pages/stages/processor/index.md b/docs/pages/stages/processor/index.md index fea8915..a09e01d 100644 --- a/docs/pages/stages/processor/index.md +++ b/docs/pages/stages/processor/index.md @@ -11,6 +11,7 @@ Processor stages read from an input connector and write to an output connector. | Generic | [Generic](generic.md) | `processor.NewGenericStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | User-defined transform | | Generic | [Filter](filter.md) | `processor.NewFilterStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Drop messages by predicate | | Generic | [Tee](tee.md) | `processor.NewTeeStage` | Single runner | Clone one stream to many outputs | +| Generic | [Merge](merge.md) | `processor.NewMergeStage` | Single runner | Combine many streams into one output | | CSV | [Decoder](csv-decoder.md) | `processor.NewCSVDecoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Bytes to typed CSV rows | | CSV | [Encoder](csv-encoder.md) | `processor.NewCSVEncoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Typed CSV rows to bytes | | CAN | [CAN](can.md) | `processor.NewCANStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Raw CAN frames to decoded signals | diff --git a/docs/pages/stages/processor/merge.md b/docs/pages/stages/processor/merge.md new file mode 100644 index 0000000..c47cb7c --- /dev/null +++ b/docs/pages/stages/processor/merge.md @@ -0,0 +1,52 @@ +--- +icon: lucide/git-merge +--- + +# Merge Processor + +`MergeStage` combines multiple input connectors into one output connector. + +``` go +cfg := processor.NewMergeConfig() +stage := processor.NewMergeStage( + []connector.MessageConnector[*MyMessage]{inA, inB, inC}, + out, + cfg, +) +``` + +## Messages + +### Input Message + +Accepted body type: `T`. + +Additional input interfaces: none required beyond `message.Body`. Every input +connector must carry the same body type. + +### Output Message + +Produced body type: the same `T` body received from any input connector. + +Additional interfaces: preserved from `T`. The stage does not clone, copy, or +transform the message. It forwards each incoming message envelope to the output +connector. + +## Configuration + +| Field | Default | Description | +| --- | --- | --- | +| `OutputQueueSize` | `256` | Size of the internal fan-in ring buffer between input readers and the output bridge. | + +At least one input connector is required. + +## Internals + +`MergeStage` uses a custom single runner instead of the generic worker pool. +The runner starts one reader goroutine per input connector. Each reader moves +messages into an internal MPSC fan-in buffer, and one output bridge drains that +buffer into the output connector. + +The stage keeps reading until every input connector is closed and drained. It +then closes the internal buffer, waits for the output bridge to finish, and +closes the output connector during stage close. diff --git a/docs/zensical.toml b/docs/zensical.toml index b036fba..c02f288 100644 --- a/docs/zensical.toml +++ b/docs/zensical.toml @@ -34,6 +34,7 @@ nav = [ { "Generic" = "stages/processor/generic.md" }, { "Filter" = "stages/processor/filter.md" }, { "Tee" = "stages/processor/tee.md" }, + { "Merge" = "stages/processor/merge.md" }, { "CSV" = [ { "Decoder" = "stages/processor/csv-decoder.md" }, { "Encoder" = "stages/processor/csv-encoder.md" }, diff --git a/processor/merge.go b/processor/merge.go new file mode 100644 index 0000000..6e88fe4 --- /dev/null +++ b/processor/merge.go @@ -0,0 +1,185 @@ +package processor + +import ( + "context" + "errors" + "sync" + + "github.com/FerroO2000/goccia/connector" + "github.com/FerroO2000/goccia/internal/config" + "github.com/FerroO2000/goccia/internal/metrics" + "github.com/FerroO2000/goccia/internal/rb" + "github.com/FerroO2000/goccia/internal/stage" + "github.com/FerroO2000/goccia/internal/stage/env" +) + +// ─── Config ─────────────────────────────────────────────────────────────────| + +// Default values for the merge stage configuration. +const ( + DefaultMergeConfigOutputQueueSize = 256 +) + +// MergeConfig structs contains the configuration for the merge stage. +type MergeConfig struct { + // OutputQueueSize is the size of the fan-in ring buffer + // placed between the input connectors and the output connector. + OutputQueueSize int +} + +// NewMergeConfig returns a new MergeConfig with default values. +func NewMergeConfig() *MergeConfig { + return &MergeConfig{ + OutputQueueSize: DefaultMergeConfigOutputQueueSize, + } +} + +// Validate checks the configuration. +func (c *MergeConfig) Validate(ac *config.AnomalyCollector) { + config.CheckNotNegative(ac, "OutputQueueSize", &c.OutputQueueSize, DefaultMergeConfigOutputQueueSize) + config.CheckNotZero(ac, "OutputQueueSize", &c.OutputQueueSize, DefaultMergeConfigOutputQueueSize) +} + +// ─── Environment ────────────────────────────────────────────────────────────| + +type mergeEnv struct { + *env.BaseEnv[*MergeConfig, *metrics.EmptyMetrics] +} + +func newMergeEnv(config *MergeConfig) *mergeEnv { + return &mergeEnv{ + BaseEnv: env.NewProcessorEnv(config, metrics.NewEmptyMetrics()), + } +} + +// ─── Runner ─────────────────────────────────────────────────────────────────| + +var _ stage.Runner[*mergeEnv] = (*mergeRunner[msgBody])(nil) + +type mergeRunner[T msgBody] struct { + *mergeEnv + + inConnectors []msgConn[T] + outConnector msgConn[T] + + fanIn *rb.RingBuffer[*msg[T]] + + readersWg *sync.WaitGroup + + runOutputBridgeDone chan struct{} + runDone chan struct{} +} + +func newMergeRunner[T msgBody](inConnectors []msgConn[T], outConnector msgConn[T]) *mergeRunner[T] { + return &mergeRunner[T]{ + inConnectors: inConnectors, + outConnector: outConnector, + + readersWg: &sync.WaitGroup{}, + + runOutputBridgeDone: make(chan struct{}), + runDone: make(chan struct{}), + } +} + +func (mr *mergeRunner[T]) SetEnvironment(env *mergeEnv) { + mr.mergeEnv = env +} + +func (mr *mergeRunner[T]) Init(_ context.Context) error { + if len(mr.inConnectors) == 0 { + return errors.New("no input connector specified") + } + + fanInCapacity := uint64(mr.Config.OutputQueueSize) + mr.fanIn = rb.NewRingBuffer[*msg[T]](fanInCapacity, rb.BufferKindMPSC) + + return nil +} + +func (mr *mergeRunner[T]) Run(ctx context.Context) { + defer close(mr.runDone) + + mr.readersWg.Add(len(mr.inConnectors)) + for _, inConnector := range mr.inConnectors { + go mr.readInput(ctx, inConnector) + } + + go mr.runOutputBridge(context.WithoutCancel(ctx)) + + mr.readersWg.Wait() + mr.fanIn.Close() + <-mr.runOutputBridgeDone +} + +func (mr *mergeRunner[T]) readInput(ctx context.Context, inConnector msgConn[T]) { + defer mr.readersWg.Done() + + for { + msgIn, err := inConnector.Read(ctx) + if err != nil { + // This means the input connector is closed + // and there are no more messages in it + return + } + + if err := mr.fanIn.Write(msgIn); err != nil { + msgIn.Destroy() + } + } +} + +func (mr *mergeRunner[T]) runOutputBridge(ctx context.Context) { + defer close(mr.runOutputBridgeDone) + + for { + msg, err := mr.fanIn.Read(ctx) + if err != nil { + return + } + + if err := mr.outConnector.Write(msg); err != nil { + msg.Destroy() + } + } +} + +func (mr *mergeRunner[T]) Close(_ context.Context) { + <-mr.runDone + mr.outConnector.Close() +} + +func (mr *mergeRunner[T]) Inputs() []uintptr { + inputs := make([]uintptr, 0, len(mr.inConnectors)) + + for _, inConnector := range mr.inConnectors { + inputs = append(inputs, connector.GetConnectorID(inConnector)) + } + + return inputs +} + +func (mr *mergeRunner[T]) Outputs() []uintptr { + return []uintptr{connector.GetConnectorID(mr.outConnector)} +} + +// ─── Stage ──────────────────────────────────────────────────────────────────| + +// MergeStage is a stage that merges multiple input connectors into a single +// output connector. +// It is the counterpart of the tee stage. +type MergeStage[T msgBody] struct { + *stage.ProcessorStage[T, T, *mergeEnv] +} + +// NewMergeStage returns a new merge processor stage. +func NewMergeStage[T msgBody]( + inConnectors []msgConn[T], outConnector msgConn[T], config *MergeConfig, +) *MergeStage[T] { + + return &MergeStage[T]{ + ProcessorStage: stage.NewProcessorStageFromRunner[T, T]( + "merge", newMergeEnv(config), newMergeRunner(inConnectors, outConnector), + ), + } +} diff --git a/processor/merge_test.go b/processor/merge_test.go new file mode 100644 index 0000000..59d32c3 --- /dev/null +++ b/processor/merge_test.go @@ -0,0 +1,88 @@ +package processor + +import ( + "testing" + + "github.com/FerroO2000/goccia/connector" + "github.com/FerroO2000/goccia/internal/message" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_MergeStage(t *testing.T) { + const ( + connSize = uint64(32) + inputCount = 3 + msgCountPerInput = 4 + ) + + inputs := make([]msgConn[*dummyMsg], 0, inputCount) + expectedMessages := make(map[*msg[*dummyMsg]]struct{}, inputCount*msgCountPerInput) + + for inputIdx := range inputCount { + in := connector.NewRingBuffer[*dummyMsg](connSize) + inputs = append(inputs, in) + + for msgIdx := range msgCountPerInput { + msgIn := message.NewMessage(&dummyMsg{value: inputIdx*msgCountPerInput + msgIdx}) + expectedMessages[msgIn] = struct{}{} + + require.NoError(t, in.Write(msgIn)) + } + + in.Close() + } + + out := connector.NewRingBuffer[*dummyMsg](connSize) + + cfg := NewMergeConfig() + cfg.OutputQueueSize = 2 + + stage := NewMergeStage(inputs, out, cfg) + require.NoError(t, stage.Init(t.Context())) + + stage.Run(t.Context()) + stage.Close(t.Context()) + + for range inputCount * msgCountPerInput { + msgOut, err := out.Read(t.Context()) + require.NoError(t, err) + + assert.Contains(t, expectedMessages, msgOut) + delete(expectedMessages, msgOut) + } + + _, err := out.Read(t.Context()) + assert.ErrorIs(t, err, connector.ErrClosed) + assert.Empty(t, expectedMessages) +} + +func Test_MergeRunner_InitRequiresInput(t *testing.T) { + out := connector.NewRingBuffer[*dummyMsg](1) + runner := newMergeRunner([]msgConn[*dummyMsg]{}, out) + runner.SetEnvironment(newMergeEnv(NewMergeConfig())) + + require.EqualError(t, runner.Init(t.Context()), "no input connector specified") +} + +func Test_MergeConfig_ValidateOutputQueueSize(t *testing.T) { + tests := map[string]int{ + "zero": 0, + "negative": -1, + } + + for name, outputQueueSize := range tests { + t.Run(name, func(t *testing.T) { + in := connector.NewRingBuffer[*dummyMsg](1) + out := connector.NewRingBuffer[*dummyMsg](1) + + cfg := NewMergeConfig() + cfg.OutputQueueSize = outputQueueSize + + stage := NewMergeStage([]msgConn[*dummyMsg]{in}, out, cfg) + require.NoError(t, stage.Init(t.Context())) + + assert.Equal(t, DefaultMergeConfigOutputQueueSize, cfg.OutputQueueSize) + }) + } +}