# manux

**Repository Path**: gdutwang/manux

## Basic Information

- **Project Name**: manux
- **Description**: manux framework for c/c++. A high-performance C++14 message processing framework for industrial vision systems.
- **Primary Language**: C++
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 2
- **Forks**: 0
- **Created**: 2026-03-16
- **Last Updated**: 2026-10-06

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Manux 4.0 — Message Processing Framework

A high-performance C++14 message-processing framework for industrial vision systems.
All types are in `namespace mx`. Examples below assume `using namespace mx;`.

## Core Principles

1. **Frame ownership** — a message has exactly one owning frame per node visit; everything inside the frame borrows by reference; exactly one terminal transfer (forward or release) per visit
2. **Messages are data carriers** — payloads only, deleter-stamped at creation, framework-released on every exit path
3. **Callbacks are observers** — they read context + payload view; they can never take, mutate routing of, or forward the message
4. **Zero-copy passing** — ownership transfer via `std::unique_ptr`, never deep copy
5. **Lock-free queues** — SPSC/MPSC for performance
6. **Signal-based control** — lifecycle via `SIG_DRAIN`/`SIG_STOP`/`SIG_KILL`/`SIG_PAUSE`/`SIG_RESUME`; signals bypass the queue (Approach B)

## Quick Start

```cpp
#include "mx_node.hpp"
#include "mx_flow.hpp"

// Business node: inherit mx_TypedNode<Payload>, override do_process only.
class Detector : public mx_TypedNode<FrameData> {
protected:
    mx_ErrorCode do_process(FrameData& data) override {   // pure business
        if (data.empty()) return fail(mx_ErrorCode::INVALID_ARGUMENT, "empty frame");
        // ... business logic on data ...
        emit(sideChain_, make_gray_payload(...));          // optional side-chain
        return mx_ErrorCode::SUCCESS;                      // callbacks/forward/release: automatic
    }
};

int main() {
    Detector detector("Detector");
    mx_Flow flow("Pipeline");
    flow.add_step(&detector);
    flow.set_completion_callback([](const mx_MessageContext& ctx, mx_PayloadView) {
        // Observer: ctx has result and path; view can read the payload (as<T>())
        // or salvage it (detach()) — but never take the message.
    });

    flow.start();
    auto* payload = new FrameData(...);
    auto msg = mx_MessageFactory::create_owned<FrameData>(payload);  // deleter-stamped
    flow.post(msg);       // reference; moved on success, untouched on failure
    flow.stop();
    return 0;
}
```

## Key Components

| Component | Purpose |
|-----------|---------|
| `mx_Message<T>` | Data carrier (payload + context); deleter lives in `mx_MessageContext` |
| `mx_TypedNode<Payload>` | THE business node base — override `do_process(Payload&)` |
| `mx_Flow` | Sequential pipeline; structural forwarding via `next_`/`parent_` |
| `mx_Pool` | Parallel dispatch with strategies (RR / FIRST_AVAILABLE / ALL / ANY) |
| `mx_Executor` | Execution strategy (Inline/Single/Multi, 8-state machine) |
| `mx_PayloadView` | Observer-side payload projection: `as<T>()` read, `detach()` salvage |
| `mx_MessageUtils` | Signal creation, cloning, `release()` (single payload-release choke point) |
| `mx_MessageFactory` | `create_owned<Payload>(raw)` — deleter-stamped message creation |

## Message Design

```cpp
template<typename Payload = void*>
struct mx_Message {
    Payload payload;                     // Business data (raw pointer, 8B default)
    mx_MessageType type;                 // NORMAL / SIG_*
    std::unique_ptr<mx_MessageContext> context;   // framework metadata
};
// Layout is static_assert-locked: mx_Message<> and mx_Message<T*> share offsets.

struct mx_MessageContext {
    int64_t message_id;
    mx_MessageState state;
    std::vector<mx_ContextEntry> path;   // always recorded now (v3 bypass fixed)
    mx_ErrorCode result;
    std::string error_message;
    void (*payload_destroy)(void*) = nullptr;   // stamped by create_owned
    size_t payload_type_hash = 0;               // type-mismatch detection
};
```

**Ownership**: `create_owned<Payload>(raw)` adopts a heap payload and stamps the deleter.
The framework destroys it exactly once — on normal completion, forward failure,
SIG_STOP/SIG_KILL discard, exceptions, broadcast partial failure. **Application code
never deletes payloads.** `release()` also writes poison canaries pre-destroy, turning
same-tick dangling access into an immediate crash instead of silent heap corruption.

### Signals Bypass Queue (unchanged from v3.6)

| Signal | Effect |
|--------|--------|
| `SIG_DRAIN` | Process all pending, then stop |
| `SIG_STOP` | Process ≤1 queued msg, discard rest, stop |
| `SIG_KILL` | Immediate termination, discard all |
| `SIG_PAUSE` | Drain then pause (accepts only SIG_RESUME) |
| `SIG_RESUME` | Restore RUNNING from PAUSED |

### post() — By Reference

`post(MessagePtr& msg)` moves only on success; on failure (QUEUE_FULL,
NODE_NOT_RUNNING, ...) the caller retains ownership for retry or cleanup.

## Key Patterns

### Typed Business Node (the only node you write)

```cpp
class MyNode : public mx_TypedNode<MyPayload> {
protected:
    mx_ErrorCode do_process(MyPayload& data) override {
        // No casts, no null-msg checks, no callback calls, no deletes.
        // Return code is the protocol: SUCCESS → completion + forward;
        // anything else → error observer + release. Exceptions are contained.
        return mx_ErrorCode::SUCCESS;
    }
};
```

### Flow / Pool Assembly

```cpp
mx_Flow flow("Pipeline");
flow.add_step(&node1);
flow.add_step(&pool);            // mx_Pool("Workers", strategy) + add_worker(...) xN
flow.add_step(&node2);
flow.set_completion_callback([](const mx_MessageContext&, mx_PayloadView) {});
flow.set_error_callback([](const mx_MessageContext& ctx, mx_PayloadView view) {
    // Dispatch outcome → completion observer (ctx.result carries the code);
    // processing failure → this error observer. view.as<T>() to inspect,
    // view.detach() to salvage the payload.
});
flow.start();
auto msg = mx_MessageFactory::create_owned<Payload>(new Payload(...));
flow.post(msg);
flow.stop();                     // sequential drain is built-in (no manual per-node signals)
```

### Graceful Shutdown

```cpp
flow.post(mx_MessageUtils::create_drain_signal(999999));
while (!flow.is_drain_complete()) std::this_thread::sleep_for(std::chrono::milliseconds(10));
flow.stop();
```

### Executor Types

| Type | Threading | Queue | Use Case |
|------|-----------|-------|----------|
| `Inline` | None (caller thread) | None | Flow's own executor, logging |
| `Single` | 1 thread | MPSC (8192) | Camera, preprocessing |
| `Multi` | Emitter + N workers | MPSC + SPSC | Inference, post-processing |

---

# v3 → v4 Upgrade Summary

v4.0 is a **breaking release** built around one idea: *ownership transfers only at
frame boundaries; inside a frame everything borrows.* The full design lives in
`openspec/changes/v4-frame-ownership/` (proposal / design / specs / tasks); evidence
chain in `issuelist.md`.

## Why

Every production incident of the last six months traced to one v3 design: message
ownership moved ~6 times per node visit and framework duties (casts, null checks,
callback dispatch, payload release) lived in business code — ds2 #071 (QUEUE_FULL
drop leaked 31 MB payloads → 6.5 GB → OOM), #036 (SIG_STOP discard leaked 1–2 GB per
cycle), #072 (touching a message after a callback took ownership → LFH heap
corruption). v4 makes those bug classes impossible by construction, not by discipline.

## Breaking Changes

| # | Change | v3 | v4 |
|---|--------|----|----|
| 1 | Business node base | `mx_Node` + override `process(MessagePtr)` (framework duties in business code) | `mx_TypedNode<Payload>` + override `do_process(Payload&)` — the frame (checks, cast, exceptions, callbacks, terminal) is `final` |
| 2 | Callbacks | `void(ctx, MessagePtr*)` — forwarding hack, ownership-stealing | `void(ctx, mx_PayloadView)` — pure observers; forwarding is structural (`next_`/`parent_` + `on_child_completed`) |
| 3 | Payload lifetime | raw pointer, manual `delete` on every error path | deleter in context, stamped by `create_owned`, destroyed by the single `release()` choke point |
| 4 | Message-level callback APIs | `call_message_completion_callback` / `call_message_error_callback` ... | **Removed** — the frame fires observers automatically |
| 5 | Example nodes in core | `mx_SimpleNode` / `mx_LambdaNode` / `mx_CustomNode` | Deleted / moved to `test/test_nodes.hpp` |
| 6 | Flow internals | StepLink callback-wrapping machine | Deleted — structural `next_` chain; user step callbacks are personal observers |
| 7 | Error strings | passed with callback calls | `return fail(code, "msg")` — flows into `ctx.error_message` |

## Migration Mapping

| v3 pattern | v4 pattern |
|------------|------------|
| `class NodeX : mx_Node` + `process(MessagePtr)` | `class NodeX : mx_TypedNode<PayloadX>` + `do_process(PayloadX&)` |
| `static_cast<...>(msg.get())->payload` + null checks | deleted (single cast inside the template, type-mismatch = loud error) |
| `call_message_*(std::move(msg), r)` | deleted — `return r` is the protocol |
| `delete infer_data; payload=nullptr;` (×N error paths) | deleted — framework `release()` |
| `new Payload(...); msg->payload = p;` | `mx_MessageFactory::create_owned<Payload>(p)` |
| mid-process side post (postGrayData pattern) | `emit(target, owned_payload)` — fire-and-forget, never interrupts the main pipeline |
| flow error callback with manual payload delete | observer with `view.as<T>()` / `view.detach()` |

**Teardown contract**: borrowed steps must outlive the Flow, or the Flow must be
`tini()`d/destroyed first — `Flow::tini()` now detaches and clears steps, making the
explicit `stop() → tini() → destroy` order structurally safe.

## Verified Results (same machine, same day, same DEBUG-log conditions)

| Scenario (1000 msgs) | v3.11 | v4.0 | Speedup |
|---|---|---|---|
| 5-node linear flow | 4622 ms | 3397 ms | 1.36× |
| Node→Pool→Node→Pool→Node | 12752 ms | 3554 ms | 3.59× |
| 8-node deep chain + pools | 28572 ms | 4973 ms | 5.75× |

At 10000 msgs v4 scales **linearly** (×8–9.4 duration) while v3.11 degraded
superlinearly (×12.7–15.3) and stalled one scenario (4516/10000 completed, exit=1);
v4 completed 10000/10000 everywhere. Structural forwarding (direct `next_->post`)
replaces v3's per-hop callback-wrapper chain — deeper pipelines save more.

Full test battery: unit 16/32/5/15/26 + regression 37/37 + integration 5/5 +
stress 5/5 @100/1000/10000 — all green, message counts identical to v3.11.

## Build & Test

```bash
msbuild test\test_unit_flow.vcxproj /p:Configuration=Release /p:Platform=x64   # etc.
cd src\bin\Release
test_unit_node.exe  test_unit_flow.exe  test_unit_pool.exe  test_unit_executor.exe
test_signal_control.exe  test_p0_regression.exe  test_complex.exe
stress_100k.exe 1000
```

## File Structure

```
include/manux/
├── mx_types.hpp        # Types, enums, Message/Context, PayloadView, release(), MessageUtils
├── mx_queue.hpp        # SPSC/MPSC lock-free queues (push by-ref)
├── mx_executor.hpp     # Inline/Single/Multi executors (state machine, inline signals)
├── mx_node.hpp         # mx_Node + mx_TypedNode<Payload> (frame, fail/emit, terminal)
├── mx_flow.hpp         # mx_Flow — structural forwarding (next_/parent_ + on_child_completed)
├── mx_pool.hpp         # mx_Pool — dispatch + HazardManager reclamation
├── mx_message.hpp      # mx_MessageFactory — create_owned / stamp_owned
├── mx_utils.hpp        # String, time, encoding, path, math utilities
├── mx_perf.hpp         # Performance profiling
├── mx_log.h            # Async logging (MX_DEBUG_ENABLE = log-channel switch only)
└── mx_sqlite3.h        # Optional SQLite3 access (mx_SqliteDb/mx_SqlGateway, links sqlite3.lib)
```

## See Also

- `userguide.md` — User guide: per-component usage (when / key API / example / pitfalls)
- `manux.md` — Detailed design documentation (v4: frame ownership, message/PayloadView details, D1–D10)
- `openspec/changes/v4-frame-ownership/` — Full v4 change artifacts (design decisions D1–D10)
- `issuelist.md` — Change history & evidence chain (v4.0 entry: 2026-10-03)
- `AGENTS.md` — Development environment & conventions