Keyboard shortcuts

Press or to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

CPOW Process Binary Analysis

Synapse CPOW-Process: Data Pipeline & Implementation Deep Dive

Location: /mnt/code/dev/Rust/energy-quotient/pq-synapse/server/crates/synapse-cpow/process


1. OVERVIEW & ENTRY POINTS

Binary Configuration

  • Binary Name: cpow-process
  • Source: src/bin/default.rs
  • Features:
    • force_no_neon: Disables ARM NEON optimization (forces fallback implementation)
    • use_neon: Automatically enabled on ARM platforms (aarch64/arm)

Main Command-Line Architecture

cpow-process [OPTIONS] <COMMAND>

Global Options:
  --dev                    Enable development mode (custom shared memory naming)

Commands:
  default --verbose        Run the default wave processor
  parquet --output-dir     Parquet processor (future implementation)

Current Implementation Flow

The binary currently supports two execution paths:

  1. Production: default command (–dev flag not set)
  2. Development: default command with –dev flag

2. DATA PIPELINE ARCHITECTURE

High-Level Data Flow

Shared Memory (Raw Sensor Data)
           ↓
    SharedMemoryReader Thread
           ↓
  WaveBlockData → flume channel
           ↓
    Processor Thread
           ↓
   Data Conversion & Scaling
           ↓
Arrow RecordBatch Creation
           ↓
   Parquet File Writer
           ↓
    Parquet Files on Disk

Processing Stages

Stage 1: Shared Memory Reading (eq-memops::reader)

  • Thread: Reader spawned with priority -10 (high priority on Linux)
  • Input: SensorSharedMemory (circular buffer from CPOW acquire)
  • Output: WaveBlockData tuples sent via flume channel
  • Data Type: WaveBlockData = (Vec<SensorDataRow>, u64)
    • Vec<SensorDataRow>: 256 sensor samples (3 phases)
    • u64: Timestamp (nanoseconds since UNIX epoch)

Key Implementation (eq-memops/src/ipc/reader.rs:SharedMemoryReader::run):

#![allow(unused)]
fn main() {
// Reads N_ROWS_PER_BLOCK (256) rows at a time
// Block contains 7 channels × 256 rows of 24-bit data
// Timestamp marks start of block
// Detects overrun conditions and buffer state changes
}

Stage 2: Data Buffering & Overflow Handling

  • Component: WaveSink (custom channel wrapper)
  • Purpose: Manages bounded channel + overflow buffer + disk fallback
  • Buffer States:
    • Normal: Data flowing normally
    • Buffering: Channel full, using in-memory overflow buffer
    • WritingToDisk: Overflow buffer full, writing to disk
  • Configuration:
    • OVERFLOW_BUFFER_SIZE: 10 seconds of blocks (125 blocks)
    • MAX_FILE_SIZE: 32 MB per overflow file
    • MAX_TOTAL_SIZE: 512 MB total overflow storage

Stage 3: Processing & Format Conversion

  • Thread: Processor spawned with priority -10
  • Input: WaveBlockData from channel (flume receiver)
  • Processing Engine: DefaultWaveProcessorType

Key Transformations:

  1. 24-bit to 32-bit Conversion:

    #![allow(unused)]
    fn main() {
    // Each SensorDataRow contains 7 channels of 24-bit data
    // Converted to i32 (sign-extended)
    fn i24_to_i32(bytes: [u8; 3]) -> i32 {
        if (bytes[2] & 0x80) != 0 {  // Check sign bit
            (0xFF000000u32 | ...) as i32  // Negative (extend)
        } else {
            ((bytes[2] as u32) << 16 | ...) as i32  // Positive
        }
    }
    }
  2. Data Batching (DataBuffer from eq-memops::batch):

    • Accumulates raw 24-bit data into 32-bit format
    • Organizes by channel
    • Batch size: configurable (default: 320,000 rows = 10 seconds @ 32 kHz)
  3. Scaling & Arrow Conversion:

    • Current/voltage scaling factors applied
    • Conversion to Arrow arrays for columnar storage
    • Schema: [IA, VA, IB, VB, IC, VC, IN] (7 Int32 fields)

Stage 4: Parquet File Writing

  • Component: ArrowIpc + ParquetWriter
  • Output Location: $CPOW_DATA_DIR (from config)
  • File Format: Parquet with Zstandard compression
  • Metadata Storage:
    • start_time: RFC3339 timestamp with nanosecond precision
    • vscale: Voltage scaling factor
    • iscale: Current scaling factor

File Rotation Logic:

  • Triggers when:
    • Row count exceeds 1,920,000 rows (~60 seconds @ 32 kHz)
    • 60 seconds elapsed (configurable)
    • Timestamp discontinuity > 1 second detected

3. SHARED MEMORY INTERFACE

SensorSharedMemory Structure (eq-cpow/sensor)

Memory Layout (cache-line aligned):

┌─────────────────────────────┐
│  SensorSharedControl (64B)  │  
│  - read_pos (atomic)        │
│  - write_pos (atomic)       │
│  - terminate_flag           │
│  - reader_active            │
│  - generation               │
├─────────────────────────────┤
│  SensorSharedData Header    │
│  - num_rows                 │
│  - vscale (f32)             │
│  - iscale (f32)             │
├─────────────────────────────┤
│  SensorDataRow[num_rows]    │  (21 bytes each, 256 rows/block)
│  ia, va, ib, vb, ic, vc, in│
│  (3 bytes each channel)     │
├─────────────────────────────┤
│  Timestamps[num_rows/256]   │  (8 bytes each, 1 per block)
└─────────────────────────────┘

Data Types

SensorDataRow (21 bytes, packed):

#![allow(unused)]
fn main() {
#[repr(C, packed)]
pub struct SensorDataRow {
    pub ia: [u8; 3],   // Phase A current
    pub va: [u8; 3],   // Phase A voltage
    pub ib: [u8; 3],   // Phase B current
    pub vb: [u8; 3],   // Phase B voltage
    pub ic: [u8; 3],   // Phase C current
    pub vc: [u8; 3],   // Phase C voltage
    pub i_n: [u8; 3],  // Neutral current
}
}

Key Constants:

  • N_ROWS_PER_BLOCK: 256 (= 1 frame × 64 rows/frame)
  • BLOCK_SIZE: 5,376 bytes (256 rows × 21 bytes)
  • BLOCK_DURATION: 8 milliseconds @ 32 kHz
  • SAMPLE_HZ: 32,000 Hz
  • N_CHANNELS: 7

Reading Process

#![allow(unused)]
fn main() {
// In SharedMemoryReader::run()
loop {
    let rows_available = write_pos - read_pos;
    
    if rows_available >= N_ROWS_PER_BLOCK {
        // 1. Prepare data at read position
        let prepared_data = shared_mem.prepare_data_at(read_pos, N_ROWS_PER_BLOCK)?;
        // Returns: (Vec<SensorDataRow>, timestamp)
        
        // 2. Send to sink (channel)
        sink.try_consume(prepared_data)?;
        
        // 3. Advance read position
        read_pos = (read_pos + N_ROWS_PER_BLOCK) % num_rows;
    } else {
        sleep(Duration::from_micros(10));
    }
}
}

Buffer State Detection (metrics tracking):

  • Normal: Standard operation
  • OverrunRiskWarning: Buffer > 50% full (wrapped case)
  • Overrun: Reader hasn’t kept up, data lost
  • WritingToDisk: Overflow buffer being persisted

4. ARROW SHARED MEMORY IMPLEMENTATION (Partially Complete)

Status: TODO - Incomplete

Location: eq-memops/src/ipc/arrow.rs (lines 364+)

Architecture Design

ArrowSharedMemory Structure (proposed):

#![allow(unused)]
fn main() {
pub struct ArrowSharedMemory {
    shmem: Arc<Shmem>,
    schema: SchemaRef,
    control: *mut ArrowSharedControl,
    data_offset: usize,
    total_size: usize,
    reader_mode: ReaderMode,  // Single or Multi(MultiReaderState)
}
}

Control Structure (for lock-free multi-reader):

#![allow(unused)]
fn main() {
#[repr(align(64))]
pub struct ArrowSharedControl {
    write_pos: AtomicUsize,
    generation: AtomicI64,
    batch_infos: [BatchInfo; 1024],  // Up to 1024 batches
}

struct BatchInfo {
    offset: AtomicUsize,
    size: AtomicUsize,
    timestamp: AtomicI64,
}
}

Reader Control (multi-reader support):

#![allow(unused)]
fn main() {
#[repr(align(64))]
struct ReaderControl {
    positions: [AtomicUsize; 16],    // Up to 16 readers
    active: [AtomicUsize; 16],       // Reader active flags
}
}

Design Goals

  1. Lock-free operation: Uses compare-and-swap for coordination
  2. Multi-reader support: 16 concurrent readers maximum
  3. Arrow Record Batch streaming: Direct serialization to shared memory
  4. Circular buffer: Automatic wrapping when full

Implemented Methods

  • new(): Create/open shared memory region with schema
  • write(): Write RecordBatch as Arrow IPC format
  • read(): Read batch by reader ID
  • register_reader(): Register new reader (returns ID)
  • unregister_reader(): Deactivate reader

Known Issues (TODO markers)

#![allow(unused)]
fn main() {
todo!("Determine intended logic and fix inst of ArrowSharedMemory");  // Line 364
}

The initialization logic is incomplete - control pointer setup and boundary calculations need finalization.

Proposed Use Cases (Future)

  1. Live data streaming to web frontends via WebSocket
  2. Real-time analytics consumer processes
  3. Event detection system reading raw data simultaneously with file writing

5. NEON OPTIMIZATION (ARM-Specific)

Overview

The system uses ARM NEON (Single Instruction Multiple Data) instructions for high-performance 24-bit to 32-bit conversion and scaling.

Implementation

Location: eq-memops/src/neon_ops/

Rust Wrapper (conversion.rs):

#![allow(unused)]
fn main() {
pub fn i32_to_f32_scaled(input: &[i32], scale: f32) -> Vec<f32> {
    #[cfg(any(target_arch = "arm", target_arch = "aarch64"))]
    {
        convert_i32_to_f32_neon(input, scale)
    }

    #[cfg(not(any(target_arch = "arm", target_arch = "aarch64")))]
    {
        // Fallback for x86/other platforms
        input.iter().map(|&x| x as f32 * scale).collect()
    }
}
}

C Implementation (conversion.c):

  • Uses ARM NEON intrinsics for batch conversion
  • In-place transformation of i32 buffer to f32
  • Handles scaling factor application

Build Integration (build.rs):

#![allow(unused)]
fn main() {
if (target.contains("aarch64") || target.contains("arm")) && !force_no_neon {
    cc::Build::new()
        .file("../../eq-memops/src/neon_ops/conversion.c")
        .compile("neon_ops");
    println!("cargo:rustc-cfg=feature=\"use_neon\"");
}
}

Force Disable:

cargo build --features force_no_neon

6. COMMAND-LINE OPTIONS & DEFAULT FLAG

Currently Supported Commands

default Command

cargo run --bin synapse-cpow-process -- default [OPTIONS]

Options:
  -v, --verbose    Enable verbose output during processing

Behavior:

  • Reads from production shared memory path (no user suffix)
  • Threads spawned with priority -10
  • Metrics enabled by default
  • Normal error handling with automatic recovery

default with --dev Flag

cargo run --bin synapse-cpow-process -- --dev default --verbose

Options:
  --dev                    Add user suffix to shared memory name
  -v, --verbose            Verbose mode in processor

Behavior:

  • Loads config from database (SynapseConfig::from_db_sync())
  • Appends _{USER} to shared memory name
  • Example: raw_cpow_shmem_user instead of raw_cpow_shmem
  • Useful for development without interfering with production data

Future --default Flag (Not Yet Implemented)

The code structure suggests future support for a --default flag to select default processor behavior:

#![allow(unused)]
fn main() {
#[derive(Subcommand, Debug)]
enum ProcessorCommands {
    Default { .. },      // Current implementation
    Parquet { .. },      // Future: custom parquet output location
}
}

This would allow:

# Potential future syntax
cargo run --bin synapse-cpow-process -- --default

7. PROCESSING PIPELINE DETAILED WALKTHROUGH

Thread Initialization

Main Thread (src/bin/default.rs:main()):

#![allow(unused)]
fn main() {
// 1. Create channel for data flow
let (tx, rx) = flume::unbounded::<WaveBlockData>();

// 2. Create WaveSink with overflow handling
let sink = WaveSink::new(tx, OVERFLOW_BUFFER_SIZE)?;

// 3. Spawn reader thread
let reader_handle = spawn_thread::<_, _, _, WaveProcessError>(
    "shared_memory_reader", 
    move || {
        // Set priority to -10 (high)
        setpriority(PRIO_PROCESS, 0, -10);
        
        // Build and run reader
        let mut reader = SharedMemoryReaderBuilder::<
            SensorSharedMemory, 
            WaveSink, 
            WaveBlockData
        >::default(sink)?
            .with_metrics(true)
            .build()?;
        
        reader.run(Arc::clone(&term_flag_ref))
    }
)?;

// 4. Spawn processor thread
let processor = ProcessorBuilder::new()
    .with_source(rx)
    .build()?;

let mut proc_runner = processor.into_runner()?;

let proc_handle = spawn_thread::<_, _, _, WaveProcessError>(
    "processor",
    move || {
        setpriority(PRIO_PROCESS, 0, -10);
        proc_runner.run_as::<DefaultWaveProcessorType>(
            Arc::clone(&terminate_flag)
        )
    }
)?;
}

Reader Thread Operation

Location: eq-memops/src/ipc/reader.rs:SharedMemoryReader::run()

#![allow(unused)]
fn main() {
// 1. Wait for writer to become active
while !terminate_flag.load(Ordering::SeqCst) {
    if self.shared_mem.is_writer_active() { break; }
    thread::sleep(Duration::from_secs(1));
}

// 2. Initialize read position (align to block boundary)
let mut read_pos = self.shared_mem.read_position();
read_pos = read_pos - (read_pos % N_ROWS_PER_BLOCK);
self.shared_mem.set_read_position(read_pos);

// 3. Main read loop
loop {
    let write_pos = self.shared_mem.write_position();
    let rows_available = (write_pos + num_rows - read_pos) % num_rows;
    
    if rows_available >= N_ROWS_PER_BLOCK {
        // Prepare data block
        let prepared_data = self.shared_mem.prepare_data_at(
            read_pos, 
            N_ROWS_PER_BLOCK
        )?;
        
        // Send to sink
        match self.sink.try_consume(prepared_data) {
            Ok(()) => {
                read_pos = (read_pos + N_ROWS_PER_BLOCK) % num_rows;
                self.shared_mem.set_read_position(read_pos);
            }
            Err(_) => {
                // Handle overflow - written to buffer/disk
            }
        }
    } else {
        thread::sleep(Duration::from_micros(10));
    }
    
    if terminate_flag.load(Ordering::Acquire) { break; }
}
}

Processor Thread Operation

Location: eq-processing/src/default.rs:DefaultWaveProcessorType::run()

#![allow(unused)]
fn main() {
// 1. Initialize Arrow IPC writer with schema
let schema = Arc::new(Schema::new(vec![
    Field::new("IA", DataType::Int32, false),
    Field::new("VA", DataType::Int32, false),
    // ... 7 channels total
]));

let mut arrow_ipc = ArrowIpc::new(schema, dest, scaler)?;
arrow_ipc.initialize_writer()?;

// 2. Main processing loop
loop {
    // Check if rotation needed (every 1 second)
    if last_rotate_check.elapsed() >= Duration::from_secs(1) {
        if arrow_ipc.check_rotation_needed()? {
            arrow_ipc.rotate()?;  // Close file, open new one
        }
        last_rotate_check = std::time::Instant::now();
    }
    
    // Receive block from channel
    match self.processor.source.recv() {
        Ok((rows, timestamp)) => {
            // Create buffer and convert to Arrow format
            let mut buffer = DataBuffer::new(rows.len());
            buffer.push_datarows(&rows)?;
            
            let mut view = buffer.raw_view(buffer.len())?;
            
            // Get first batch
            if let Some(batch) = view.next_result()? {
                // Check for timestamp discontinuity (>1s gap)
                if timestamp > (last_ts + 1_000_000_000) {
                    tracing::warn!("Timestamp discontinuity - rotating file");
                    arrow_ipc.rotate()?;
                }
                
                // Write to parquet
                arrow_ipc.write_batch(&batch, timestamp)?;
                last_ts = timestamp;
            }
        }
        Err(ChannelError::Empty) => {
            thread::sleep(Duration::from_micros(10));
        }
        Err(ChannelError::Disconnected) => break,
    }
    
    if terminate_flag.load(Ordering::Acquire) { break; }
}
}

Data Conversion Pipeline

24-bit to 32-bit in DataBuffer:

#![allow(unused)]
fn main() {
pub fn push_datarows(&mut self, rows: &[SensorDataRow]) -> Result<usize, BatchError> {
    for row in rows {
        let offset = self.write_pos * N_CHANNELS;
        
        // Unroll field access - direct 24→32 conversion
        self.data[offset]     = i24_to_i32(row.ia);  // Ch0: IA
        self.data[offset + 1] = i24_to_i32(row.va);  // Ch1: VA
        self.data[offset + 2] = i24_to_i32(row.ib);  // Ch2: IB
        self.data[offset + 3] = i24_to_i32(row.vb);  // Ch3: VB
        self.data[offset + 4] = i24_to_i32(row.ic);  // Ch4: IC
        self.data[offset + 5] = i24_to_i32(row.vc);  // Ch5: VC
        self.data[offset + 6] = i24_to_i32(row.i_n); // Ch6: IN
        
        self.write_pos = (self.write_pos + 1) % self.capacity;
        self.len += 1;
    }
    Ok(n_rows)
}
}

Arrow RecordBatch Creation:

#![allow(unused)]
fn main() {
impl ProcessorSink<WaveBlockData> for ArrowIpc<...> {
    fn write_batch(&mut self, batch: &BatchSlice<i32>, timestamp: u64) {
        // Apply scaling to each channel
        let channel_arrays: Vec<Arc<dyn Array>> = batch
            .channels
            .par_iter()
            .enumerate()
            .map(|(channel_idx, channel_data)| {
                let scale_factor = self.scaler.scale_factor(channel_idx);
                
                // Scale: current (i%2==0) or voltage (i%2==1)
                let scaled_values: Vec<f32> = channel_data
                    .iter()
                    .map(|&v| scale_factor * v as f32)
                    .collect();
                
                Arc::new(arrow::array::Float32Array::from(scaled_values))
                    as Arc<dyn Array>
            })
            .collect();
        
        // Create batch and write to parquet
        let record_batch = RecordBatch::try_new(
            self.schema.clone(), 
            channel_arrays
        )?;
        
        self.writer.write_batch(&record_batch)?;
    }
}
}

8. PARQUET OUTPUT CONFIGURATION

File Format & Compression

Writer Configuration (ParquetWriter::init):

#![allow(unused)]
fn main() {
let props = WriterProperties::builder()
    .set_encoding(Encoding::DELTA_BINARY_PACKED)
    .set_compression(Compression::from(config.cpow.parquet.compression))
    .set_max_row_group_size(ROWS_PER_BATCH)
    .set_data_page_row_count_limit(ROWS_PER_BATCH / 4)
    .set_data_page_size_limit(PARQUET_DATA_PAGE_SIZE_LIMIT)
    .set_dictionary_page_size_limit(PARQUET_DICTIONARY_PAGE_SIZE_LIMIT)
    .set_statistics_enabled(EnabledStatistics::Page)
    .set_key_value_metadata(Some(metadata))
    .build();
}

Constants (ParquetConstants):

  • ROWS_PER_BATCH: 320,000 rows (10 seconds @ 32 kHz)
  • TERMINATION_CHECK_ROWS: 3,200 rows (100 ms)
  • PARQUET_DATA_PAGE_SIZE_LIMIT: 83,886,080 bytes (~80 MB)
  • PARQUET_DICTIONARY_PAGE_SIZE_LIMIT: 16,384 bytes (~16 KB)

File Structure

Filename Format:

{CPOW_DATA_DIR}/{YYYYMMDD_HHMMSS}.parquet

Metadata Key-Value Pairs:

  • start_time: RFC3339 timestamp (nanosecond precision)
  • vscale: Voltage scale factor (float)
  • iscale: Current scale factor (float)

Row Groups:

  • Size: 320,000 rows per group
  • Compression: Zstandard (configurable)
  • Encoding: Delta Binary Packed for integers

9. ERROR HANDLING & RECOVERY

Error Types

WaveProcessError hierarchy:

├─ Processing: Processor error
├─ Memops: Memory operations error
├─ Ipc: Inter-process communication error
├─ SharedMemory: Shared memory initialization error
├─ SharedMemoryReader: Reader-specific error
├─ Sensor: Sensor error
├─ Parquet: Parquet file error
├─ LedControl: LED control error
├─ DiskManager: Disk management error
├─ Channel: Custom channel error
│  ├─ Disconnected: Channel severed
│  ├─ Full: Channel at capacity
│  ├─ Timeout: Read timeout
│  ├─ BufferOverflow: Overflow buffer full
│  └─ DiskWrite: Failed to write overflow data
├─ Thread: Thread spawning error
├─ ThreadExecution: Thread runtime error
├─ Initialization: Startup error
├─ Configuration: Config loading error
├─ Io: Standard I/O error
└─ Other: Generic error

Overflow Handling Sequence

Normal Channel Send
    ↓
Channel Full? → Yes → handle_overflow()
    ↓ No
    Success

Overflow Handling:
  1. Try to pop from overflow buffer → primary channel
  2. If buffer not full → add block to buffer
  3. If buffer full → try_send to disk writer

Metrics & Monitoring

Reader Metrics (ReaderMetrics):

  • blocks_processed: Total blocks read
  • last_read_time: UNIX timestamp of last read
  • shared_mem_buffer_state: Current shared memory state
  • sink_buffer_state: Current channel state
  • overflow_count: Times channel was full
  • read_errors: Total read failures
  • processing_rate: Blocks per second

10. PERFORMANCE CONSIDERATIONS

Thread Priorities

  • Both reader and processor threads set to priority -10 (Linux)
  • Higher priority = lower latency, prevents preemption

Memory Alignment

  • SensorSharedControl: 64-byte cache-line aligned
  • SensorSharedData: 64-byte cache-line aligned
  • Minimizes false sharing between reader and processor

Batching Strategy

  • Reader: Reads 256 rows per transaction (8 ms)
  • Processor: Batches to 320,000 rows (10 seconds) before writing
  • Channel: Unbounded primary + 1,250-row overflow buffer

NEON Optimization Impact

  • ARM platforms: 4x+ speedup on 24→32-bit conversion
  • Fallback: Automatic scalar processing on other architectures
  • Force disable: Use force_no_neon feature for benchmarking

File I/O Strategy

  • Asynchronous rotation: Checks every 1 second
  • Triggers: Row count threshold OR time-based (60 sec)
  • Timestamp discontinuity: Immediate rotation on > 1 second gap
  • Compression: Zstandard (configurable level)

11. CONFIGURATION INTEGRATION

SynapseConfig Integration

#![allow(unused)]
fn main() {
// From config database
let config = SynapseConfig::from_db_sync()?;

// Used for:
- config.cpow.processing.shared_memory.raw.name
- config.cpow.parquet.compression
- Event detection parameters
- Processing thresholds
}

Development Mode User Suffix

#![allow(unused)]
fn main() {
let raw_shmem_name = if parsed.dev {
    let user = std::env::var("USER").unwrap_or_else(|_| "user".to_string());
    format!("{}_{}", config.cpow.processing.shared_memory.raw.name, user)
} else {
    config.cpow.processing.shared_memory.raw.name.clone()
};
}

12. KEY FILES & MODULES REFERENCE

ModuleLocationPurpose
WaveSinksynapse-cpow/process/src/wave_sink.rsChannel + overflow buffer management
SharedMemoryReadereq-memops/src/ipc/reader.rsGeneric shared memory reading
DefaultWaveProcessorTypeeq-processing/src/default.rsMain processor logic
ArrowIpceq-memops/src/ipc/arrow.rsArrow serialization + file writing
ParquetWritereq-processing/src/default.rsParquet file writer
SensorSharedMemoryeq-cpow/sensor/src/lib.rsShared memory interface
DataBuffereq-memops/src/ipc/batch.rsData batching & conversion
WaveDataScalereq-cpow/data/src/scale.rsChannel scaling logic
NEON opseq-memops/src/neon_ops/ARM SIMD acceleration

13. SUMMARY: DATA FLOW WITH NUMBERS

Sensor @ 32 kHz → 7 channels × 24-bit data
        ↓
Shared Memory (circular buffer, 256-row blocks)
        ↓
Reader Thread (block every 8 ms, async priority -10)
        ↓
256 rows/block → WaveBlockData tuple
        ↓
Flume Channel (unbounded + 10-sec overflow buffer)
        ↓
Processor Thread (async priority -10)
        ↓
24-bit → 32-bit conversion (NEON accelerated on ARM)
        ↓
320,000-row batch accumulation (10 seconds)
        ↓
Arrow RecordBatch with scaling applied
        ↓
Parquet File Writing (Zstandard compression)
        ↓
1,920,000-row file limit OR 60-second rotation
        ↓
Timestamped Parquet files on disk (~YYYYMMDD_HHMMSS.parquet)

14. FUTURE ENHANCEMENTS & TODOS

  1. Arrow Shared Memory (Partially Complete)

    • Complete ArrowSharedMemory::new() initialization
    • Implement multi-reader batch consumption
    • Integration with web services
  2. Parquet Processor Command

    • Custom output directory support
    • Alternative compression strategies
    • File naming conventions
  3. Metrics Enhancement

    • Export to Prometheus
    • Real-time performance dashboard
    • Per-channel statistics
  4. Overflow Handler Improvements

    • Memory-mapped files instead of standard I/O
    • Compression for overflow data
    • Automatic cleanup of old overflow files
  5. Configuration Enhancements

    • Per-command configuration overrides
    • Runtime parameter adjustment
    • Hot-reload support


© 2026 EQ Systems Inc.