diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index fe4cf47..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1,11 +0,0 @@ -# CLAUDE.md - -## Overview - -This repository contains the code for the Julia package DBN.jl. The package is used to read and write Databento Binary Encoding (DBN) files. See @README.md for more details. - -The package is not yet complete, but it is a work in progress. We are going to finish it together. - -## Reference Implementations - -The official DBN implementations have been downloaded in the dbn directory. diff --git a/FUTURE_OPTIMIZATIONS.md b/FUTURE_OPTIMIZATIONS.md deleted file mode 100644 index 9da83e4..0000000 --- a/FUTURE_OPTIMIZATIONS.md +++ /dev/null @@ -1,436 +0,0 @@ -# Future Optimization Opportunities for DBN.jl - -Based on current performance (2.8M rec/s) and profiling analysis. - -## Current Performance Baseline - -**After All Optimizations:** -- Trades 1M: 2.81M rec/s -- Trades 10M: 2.42-2.95M rec/s -- MBO 1M: 2.61M rec/s -- Allocations: 1:1 ratio (1 per record) -- Type stability: ✓ Complete - -**Comparison:** -- Python (Rust bindings): 10-12M rec/s -- **Gap: ~3.5-4.8x slower** - ---- - -## Parallelization Opportunities - -### 1. Multi-threaded Batch Processing ⭐⭐⭐ (HIGH IMPACT) - -**Opportunity**: Process records in parallel batches - -**Challenges:** -- DBN format is sequential (records have variable length) -- Cannot seek to arbitrary positions without parsing header -- File I/O is inherently sequential - -**Viable Approach**: Producer-Consumer Pattern -```julia -function read_dbn_parallel(filename::String; nthreads=Threads.nthreads()) - decoder = DBNDecoder(filename) - - # Single-threaded producer: read and batch - batches = Channel{Vector{UInt8}}(nthreads * 2) - results = Channel{Vector{DBNRecord}}(nthreads * 2) - - # Producer thread: Read raw bytes in batches - @spawn begin - batch_size = 10_000 # records per batch - current_batch = UInt8[] - - while !eof(decoder.io) - # Read batch of raw bytes - # ... (complex: need to track record boundaries) - put!(batches, current_batch) - end - close(batches) - end - - # Consumer threads: Parse batches in parallel - @threads for _ in 1:nthreads - for batch in batches - records = parse_batch(batch) - put!(results, records) - end - end - - # Collect results - all_records = Vector{DBNRecord}() - for batch_results in results - append!(all_records, batch_results) - end - - return all_records -end -``` - -**Complexity**: HIGH -- Need to track record boundaries (variable length) -- Ordering must be preserved -- Overhead of batch coordination - -**Expected Impact**: -- Best case: ~2-3x on multi-core systems (limited by I/O) -- Realistic: ~1.5-2x due to coordination overhead - -**Recommendation**: LOW PRIORITY -- I/O bound, not CPU bound -- Coordination overhead likely > benefits -- Better to optimize single-threaded path first - ---- - -### 2. Parallel File Processing (Multiple Files) ⭐⭐⭐⭐ (VERY HIGH IMPACT) - -**Opportunity**: Process multiple DBN files in parallel - -**Implementation**: -```julia -function read_dbn_batch(filenames::Vector{String}) - results = Vector{Vector{DBNRecord}}(undef, length(filenames)) - - @threads for i in eachindex(filenames) - results[i] = read_dbn(filenames[i]) - end - - return results -end -``` - -**Complexity**: TRIVIAL (already works!) - -**Expected Impact**: Linear scaling with cores (4 cores = 4x throughput) - -**Recommendation**: ⭐ DOCUMENT THIS -- Already possible with current implementation -- Perfect scaling for batch workloads -- Add examples to documentation - ---- - -### 3. SIMD Vectorization ⭐⭐ (MEDIUM IMPACT) - -**Opportunity**: Use SIMD for reading fixed-size fields - -**Current Approach** (scalar): -```julia -ts_recv = read(decoder.io, Int64) # 8 bytes -order_id = read(decoder.io, UInt64) # 8 bytes -size = read(decoder.io, UInt32) # 4 bytes -# ... many more fields -``` - -**SIMD Approach** (vectorized): -```julia -using SIMD - -# Read multiple fields at once (if aligned) -function read_mbo_msg_simd(decoder::DBNDecoder, hd::RecordHeader) - # Read 32 bytes at once (4 Int64s) - vec = vload(Vec{4, Int64}, decoder.io) - - ts_recv = vec[1] - order_id = reinterpret(UInt64, vec[2]) - # ... extract other fields -end -``` - -**Challenges:** -- Julia's `read` does more than memcpy (endianness, type checking) -- Need aligned reads -- Variable record sizes make batching hard - -**Expected Impact**: 10-20% for field-heavy records (MBO, InstrumentDef) - -**Recommendation**: LOW-MEDIUM PRIORITY -- Implementation complexity moderate -- Limited gains for simple records -- Better done after I/O optimization - ---- - -## I/O Optimizations - -### 4. Buffered I/O ⭐⭐⭐⭐⭐ (HIGHEST IMPACT) - -**Opportunity**: Reduce system calls by buffering reads - -**Current**: Julia's base IO makes a syscall for each `read(io, T)` call - -**Better Approach**: Custom buffered reader -```julia -mutable struct BufferedDBNDecoder{IO_T <: IO} - io::IO_T - buffer::Vector{UInt8} - buffer_pos::Int - buffer_size::Int - # ... metadata fields -end - -@inline function read_buffered(decoder::BufferedDBNDecoder, ::Type{T}) where T - # Check if we need to refill buffer - if decoder.buffer_pos + sizeof(T) > decoder.buffer_size - refill_buffer!(decoder) - end - - # Read from buffer (no syscall!) - val = unsafe_load(Ptr{T}(pointer(decoder.buffer, decoder.buffer_pos))) - decoder.buffer_pos += sizeof(T) - - return val -end - -function refill_buffer!(decoder::BufferedDBNDecoder) - # Single read syscall for entire buffer - decoder.buffer_size = readbytes!(decoder.io, decoder.buffer) - decoder.buffer_pos = 1 -end -``` - -**Complexity**: MEDIUM -- Need careful buffer management -- Handle buffer boundaries -- Maintain compatibility - -**Expected Impact**: 30-50% throughput improvement -- System calls are expensive (especially on Windows) -- Buffer hits are nearly free (L1 cache) - -**Recommendation**: ⭐⭐⭐⭐⭐ HIGHEST PRIORITY -- Biggest single optimization remaining -- Pure Julia implementation -- No API changes needed - ---- - -### 5. Memory-Mapped Files ⭐⭐⭐ (HIGH IMPACT) - -**Opportunity**: Use `mmap` for zero-copy file access - -**Implementation**: -```julia -function read_dbn_mmap(filename::String) - data = Mmap.mmap(filename) - decoder = DBNDecoder(IOBuffer(data)) - # ... parse as usual -end -``` - -**Challenges:** -- Compressed files can't be mmapped -- Need to handle both compressed and uncompressed -- Windows has different mmap behavior - -**Expected Impact**: 20-30% for large uncompressed files - -**Recommendation**: MEDIUM PRIORITY -- Good for specific use cases (large uncompressed files) -- Doesn't help with compressed files (most common) -- Can coexist with buffered I/O - ---- - -## Algorithmic Optimizations - -### 6. Pre-allocation with Exact Count ⭐⭐⭐⭐ (HIGH IMPACT) - -**Current**: Use `sizehint!` with estimate -```julia -records = Vector{DBNRecord}(undef, 0) -sizehint!(records, estimated_count) - -while !eof(decoder.io) - push!(records, record) # May reallocate! -end -``` - -**Better**: Pre-allocate exact size if known -```julia -# Many DBN files have record count in metadata -if has_record_count(decoder.metadata) - records = Vector{DBNRecord}(undef, decoder.metadata.record_count) - idx = 1 - - while !eof(decoder.io) - records[idx] = read_record(decoder) - idx += 1 - end -else - # Fallback to current approach - # ... -end -``` - -**Complexity**: LOW - -**Expected Impact**: 10-15% reduction in GC time - -**Recommendation**: ⭐⭐⭐⭐ HIGH PRIORITY -- Simple to implement -- No downside (fallback for unknown counts) -- Reduces allocations further - ---- - -### 7. String Interning ⭐⭐ (MEDIUM IMPACT) - -**Opportunity**: Reuse common strings (symbols, exchanges, etc.) - -**Current**: Every record allocates new strings -```julia -raw_symbol = String(strip(String(read(decoder.io, 22)), '\0')) -exchange = String(strip(String(read(decoder.io, 5)), '\0')) -``` - -**Better**: Use a string cache -```julia -mutable struct DBNDecoder{IO_T <: IO} - # ... existing fields - string_cache::Dict{UInt64, String} # Hash -> String -end - -function read_cached_string(decoder, bytes) - h = hash(bytes) - get!(decoder.string_cache, h) do - String(strip(String(bytes), '\0')) - end -end -``` - -**Expected Impact**: 5-10% memory reduction, 2-5% speed improvement - -**Recommendation**: MEDIUM PRIORITY -- Helps with memory-intensive workloads -- Most benefit for files with repeated symbols - ---- - -## Micro-Optimizations - -### 8. Inline More Aggressively ⭐ (LOW IMPACT) - -**Opportunity**: Force inlining of small hot functions - -```julia -# Current -@inline function read_mbo_msg(...) - -# More aggressive -@inline @propagate_inbounds function read_mbo_msg(...) -``` - -**Expected Impact**: 1-3% - -**Recommendation**: LOW PRIORITY -- Julia already does this well -- Marginal gains - ---- - -### 9. Constant Propagation for Fixed Values ⭐ (LOW IMPACT) - -**Opportunity**: Use `@const` for truly constant values - -```julia -const LENGTH_MULTIPLIER = UInt8(4) # Already done! -const DBN_MAGIC = b"DBN" # Could add more -``` - -**Expected Impact**: <1% - -**Recommendation**: LOW PRIORITY -- Already mostly done -- Compiler does this automatically - ---- - -## Summary: Optimization Priorities - -### Tier 1: High Impact, Reasonable Complexity ⭐⭐⭐⭐⭐ - -1. **Buffered I/O Reader** (Expected: +30-50%) - - Custom buffer with batched reads - - Reduces system calls dramatically - - Pure Julia, no API changes - -2. **Exact Pre-allocation** (Expected: +10-15%) - - Use metadata record count when available - - Eliminates vector growth overhead - - Trivial to implement - -3. **Document Parallel File Processing** (Expected: Linear scaling) - - Already works! - - Just needs documentation/examples - -### Tier 2: Good Impact, More Complexity ⭐⭐⭐ - -4. **Memory-Mapped Files** (Expected: +20-30% for uncompressed) - - Good for large uncompressed files - - Doesn't help compressed (most common) - -5. **String Interning** (Expected: +2-5% speed, +5-10% memory) - - Helps with symbol-heavy workloads - - Moderate complexity - -### Tier 3: Specialized Use Cases ⭐⭐ - -6. **SIMD Vectorization** (Expected: +10-20% for complex records) - - Only helps certain record types - - Significant complexity - -7. **Multi-threaded Batch Processing** (Expected: +50-100% best case) - - Very complex - - I/O bound limits benefits - - Coordination overhead - -### Tier 4: Marginal Gains ⭐ - -8. **Aggressive Inlining** (Expected: +1-3%) -9. **Constant Propagation** (Expected: <1%) - ---- - -## Recommended Implementation Order - -**Phase 1** (Next session): -1. Implement buffered I/O reader -2. Add exact pre-allocation path -3. Document parallel file processing - -**Expected Combined Impact**: +40-65% throughput - -**Phase 2** (Later): -4. Add memory-mapped file support -5. Implement string interning - -**Expected Additional Impact**: +25-35% - -**Phase 3** (Advanced): -6. Experiment with SIMD for complex records -7. Prototype multi-threaded batch processing (if justified) - ---- - -## Final Thoughts - -**Current Status**: Already very competitive! -- 2.8M rec/s with pure Julia -- 1:1 allocation ratio -- Complete type stability - -**Realistic Target**: 4-5M rec/s (40-80% improvement) -- Buffered I/O will get us most of the way -- Pre-allocation and mmap for the rest - -**Theoretical Maximum**: ~7-8M rec/s -- Would require all optimizations -- Diminishing returns beyond buffered I/O - -**Gap to Rust**: Will remain ~2-3x -- Rust has advantages: zero-cost abstractions, better I/O -- Julia has advantages: composability, ecosystem, ease of development -- For pure Julia: 4-5M rec/s is excellent performance diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..fb49233 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Tyler Beason + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/OPTIMIZATION_SUMMARY.md b/OPTIMIZATION_SUMMARY.md deleted file mode 100644 index 5572665..0000000 --- a/OPTIMIZATION_SUMMARY.md +++ /dev/null @@ -1,327 +0,0 @@ -# DBN.jl Performance Optimization Summary -## Complete Optimization Journey - -Date: 2025-10-26 - -### Overview - -Successfully implemented two major performance optimizations that transformed DBN.jl from having severe type instability issues to achieving competitive performance through Julia's type specialization system. - ---- - -## Optimization 1: Vector{DBNRecord} Union Type - -### Problem -- Used `Vector{Any}` to store heterogeneous record types -- Caused boxing of every record → heap allocation overhead -- GC time: 80-90% - -### Solution -Created type-stable union of all 18 message types: - -```julia -const DBNRecord = Union{ - MBOMsg, TradeMsg, MBP1Msg, MBP10Msg, OHLCVMsg, - StatusMsg, ImbalanceMsg, StatMsg, ErrorMsg, SymbolMappingMsg, SystemMsg, - InstrumentDefMsg, CMBP1Msg, CBBO1sMsg, CBBO1mMsg, TCBBOMsg, BBO1sMsg, BBO1mMsg -} - -# Changed from: -records = Vector{Any}(undef, 0) - -# To: -records = Vector{DBNRecord}(undef, 0) -``` - -### Results -- **GC time**: 80-90% → 52% -- **Type safety**: Maintained while supporting multiple types -- **All 3503 tests**: ✓ Passing - ---- - -## Optimization 2: Function Barrier Refactoring - -### Problem -- 616-line `read_record()` mega-function -- All variables typed as `ANY` (complete type instability) -- **1.1M allocations for 100K records** (11 per record!) - -### Solution -Split into type-stable helper functions using function barrier pattern: - -**Before** (616 lines of type-unstable code): -```julia -function read_record(decoder::DBNDecoder) - # ... 616 lines of if/elseif ... - # All variables: ANY type -end -``` - -**After** (21 small, type-stable functions): -```julia -# Main dispatcher (18 lines) -function read_record(decoder::DBNDecoder) - if eof(decoder.io) - return nothing - end - hd_result = read_record_header(decoder.io) - if hd_result isa Tuple - _, rtype_raw, record_length = hd_result - skip(decoder.io, record_length - 2) - return nothing - end - hd = hd_result - return read_record_dispatch(decoder, hd, hd.rtype) -end - -# Type-stable dispatch (42 lines) -@inline function read_record_dispatch(decoder::DBNDecoder, hd::RecordHeader, rtype::RType.T) - if rtype == RType.MBO_MSG - return read_mbo_msg(decoder, hd) - elseif rtype == RType.MBP_0_MSG - return read_trade_msg(decoder, hd) - # ... all 18 record types ... - end -end - -# 19 type-stable helpers (one per record type) -@inline function read_mbo_msg(decoder::DBNDecoder, hd::RecordHeader) - # All variables: concrete types - ts_recv = read(decoder.io, Int64) # Int64, not ANY - order_id = read(decoder.io, UInt64) # UInt64, not ANY - # ... - return MBOMsg(hd, ...) # Concrete type, not ANY -end -``` - -### Results (100K records) -- **Allocations**: 1.1M → 499K (-55%) -- **GC time**: 80-90% → 78-82% (limited by abstract IO type) -- **Code quality**: 616-line function → 21 small functions -- **Type stability**: ALL variables `ANY` → Most type-stable -- **All 3503 tests**: ✓ Passing - ---- - -## Optimization 3: Parametric DBNDecoder{IO_T} - -### Problem -Abstract `IO` type in struct prevented type specialization: - -```julia -mutable struct DBNDecoder - io::IO # ❌ Abstract type - base_io::IO # ❌ Runtime dispatch on every operation - # ... -end -``` - -**Evidence from @code_warntype**: -```julia -decoder::DBNDecoder -%7 = Base.getproperty(decoder, :io)::IO # ← Abstract! -%8 = eof(%7)::ANY # ← Type instability! -``` - -### Solution -Parametrize on concrete IO type: - -```julia -mutable struct DBNDecoder{IO_T <: IO} - io::IO_T # ✓ Concrete type parameter - base_io::IO # Can stay abstract (rarely accessed) - # ... -end - -# Constructor with automatic type inference -DBNDecoder(io::IO_T) where {IO_T <: IO} = DBNDecoder{IO_T}(io, io, nothing, nothing, 0) -``` - -**After @code_warntype**: -```julia -decoder::DBNDecoder{IOStream} # ✓ Concrete type! -%7 = Base.getproperty(decoder, :io)::IOStream # ✓ Concrete! -%8 = eof(%7)::Bool # ✓ Type stable! -``` - -### Results - -**100K Records:** -| Metric | Before Parametric | After Parametric | Improvement | -|--------|-------------------|------------------|-------------| -| Allocations | 499K (5/record) | 100K (1/record) | **-80%** | -| Memory | 14.5 MiB | 6.9 MiB | **-52%** | -| Total Time | 0.29s | 0.25s | **-14%** | -| GC Time % | 78-82% | 88-89% | -10% | -| **Actual Compute Time** | 0.064s | 0.028s | **-56%** | - -**1M Records:** -| Metric | Before Parametric | After Parametric | Improvement | -|--------|-------------------|------------------|-------------| -| Throughput | 1.22M rec/s | 1.69M rec/s | **+39%** | -| GC Time % | ~50% | ~50% | ~ | -| Allocations | 1:5 ratio | 1:1 ratio | **-80%** | - -### Type Stability Achieved -- `decoder.io`: `IO` (abstract) → `IOStream` (concrete) ✓ -- `eof(decoder.io)`: `ANY` → `Bool` ✓ -- All IO operations: Runtime dispatch → Compile-time specialization ✓ - ---- - -## Combined Impact: All Optimizations - -### Allocation Reduction -``` -Original: 1.1M allocations (11 per record) - ↓ Function Barriers (-55%) -After Opt 2: 499K allocations (5 per record) - ↓ Parametric Types (-80%) -Final: 100K allocations (1 per record) - -Total Reduction: 91% (1.1M → 100K) -``` - -### Memory Reduction -``` -Original: Unknown (likely >20 MiB) -After Opt 2: 14.5 MiB -Final: 6.9 MiB - -Confirmed: 52% reduction from function barriers onwards -``` - -### Throughput Improvement (1M records) -``` -Original: ~0.87-1.28M rec/s (baseline) -After Opt 2: 1.22M rec/s -Final: 1.69M rec/s - -Total Improvement: ~30-95% depending on baseline -``` - -### Code Quality -``` -Before: 616-line type-unstable mega-function -After: 21 small, type-stable, @inline functions -``` - ---- - -## Performance Comparison to Python - -**Before Optimizations:** -- Julia: 0.87-1.28M rec/s -- Python (Rust bindings): 10-12M rec/s -- **Gap: 8-14x slower** - -**After All Optimizations:** -- Julia: 1.69M rec/s (pure Julia implementation) -- Python (Rust bindings): 10-12M rec/s -- **Gap: ~6-7x slower** - -### Progress Toward Performance Parity -- Achieved: **40-70% faster** than baseline -- Remaining gap: Primarily due to: - 1. I/O layer differences (Julia's base IO vs Rust's buffered I/O) - 2. Vector growth strategy (room for improvement) - 3. String handling and conversions - -**Note**: Python uses Rust bindings (compiled C-level performance), while this is pure Julia. Achieving 1/6th of Rust's performance with pure Julia is actually quite competitive, especially considering Julia's advantage in user-level composability and ecosystem integration. - ---- - -## Technical Insights: Why These Optimizations Worked - -### 1. Union Types for Type Stability -**Julia Principle**: Unions of concrete types are type-stable -- `Union{Int64, String}` is type-stable -- `Any` is NOT type-stable -- Compiler can generate specialized code for each union member - -### 2. Function Barriers Eliminate Type Instability -**Julia Principle**: Small functions enable better type inference -- Large functions: Type inference gives up → `ANY` -- Small functions: Compiler can track all code paths -- `@inline` removes function call overhead - -### 3. Parametric Types Enable Specialization -**Julia Principle**: Concrete types > Abstract types -- `IO` (abstract): Requires virtual dispatch -- `IOStream` (concrete): Direct method calls -- `DBNDecoder{IOStream}`: Compiler generates specialized version - -### 4. Allocation Reduction -**Before**: Every operation allocated because types unknown -**After**: Compiler knows types → stack allocation, inlining, SIMD - ---- - -## Lessons Learned - -### What Worked -1. **Profile first**: Used `@time` and `@code_warntype` to find bottlenecks -2. **Systematic approach**: One optimization at a time -3. **Test always**: Ran full suite after each change -4. **Type stability is paramount**: Small changes, huge impact - -### Common Julia Performance Pitfalls (Avoided) -✓ Global variables (we used locals) -✓ Abstract types in structs (now parametric) -✓ Type-unstable functions (now type-stable helpers) -✓ Unnecessary allocations (reduced by 91%) -✓ Large functions (split into small helpers) - -### Best Practices Applied -✓ Function barriers for type stability -✓ Parametric types for specialization -✓ Union types instead of `Any` -✓ `@inline` for small hot functions -✓ `@code_warntype` for verification - ---- - -## Next Optimization Opportunities - -### 1. Vector Growth Strategy (Medium Impact) -**Current**: `push!` with `sizehint!` -**Better**: Pre-allocate when count is known - -Expected: -10-20% allocation overhead - -### 2. I/O Buffering (Potentially High Impact) -**Current**: Using Julia's base IO -**Better**: Custom buffered reader - -Expected: 20-40% throughput improvement - -### 3. String Handling (Low-Medium Impact) -**Current**: String allocations for symbols/metadata -**Better**: Use `Cstring` or `StaticString` where possible - -Expected: -5-10% allocations - -### 4. SIMD Optimization (Low Impact) -**Current**: Scalar operations -**Better**: SIMD for bulk field reading - -Expected: 10-15% throughput improvement - ---- - -## Conclusion - -Through systematic application of Julia performance principles, we achieved: - -✅ **91% reduction in allocations** (1.1M → 100K for 100K records) -✅ **52% reduction in memory usage** -✅ **39% improvement in throughput** (1.22M → 1.69M rec/s) -✅ **Complete type stability** in hot paths -✅ **All 3503 tests passing** with byte-for-byte compatibility - -**Key Achievement**: Transformed a type-unstable, allocation-heavy codebase into a lean, type-stable implementation that achieves competitive performance with pure Julia code. - -**Performance Gap Closed**: From 8-14x slower than Python/Rust to ~6-7x slower, with pure Julia vs compiled Rust comparison. - -**The optimization journey demonstrates Julia's power**: With proper type stability and specialization, Julia can achieve near-C/Rust performance while maintaining high-level expressiveness.** diff --git a/PARALLELIZATION_ANALYSIS.md b/PARALLELIZATION_ANALYSIS.md deleted file mode 100644 index b0d3951..0000000 --- a/PARALLELIZATION_ANALYSIS.md +++ /dev/null @@ -1,284 +0,0 @@ -# Parallelization and Advanced Optimization Analysis - -## Executive Summary - -**Current Performance**: 2.8M rec/s (trades, 1M records) -**Target**: 4-5M rec/s (achievable with buffered I/O + micro-optimizations) -**Theoretical Max**: ~7-8M rec/s (diminishing returns) - ---- - -## Parallelization: When It Works and When It Doesn't - -### ✅ What DOES Work: Multi-File Parallelism - -**Already Supported!** Current DBN.jl can process multiple files in parallel: - -```julia -using DBN - -# Process 4 files in parallel (perfect scaling) -filenames = ["data1.dbn", "data2.dbn", "data3.dbn", "data4.dbn"] - -results = Vector{Vector{DBNRecord}}(undef, length(filenames)) - -Threads.@threads for i in eachindex(filenames) - results[i] = read_dbn(filenames[i]) -end - -# On 4-core machine: ~11M rec/s aggregate throughput! -``` - -**Performance**: -- Linear scaling with number of cores -- No coordination overhead -- Perfect for batch workloads - -**Recommendation**: ⭐⭐⭐⭐⭐ **Document this!** -- Add examples to README -- Show batch processing patterns -- Highlight for ETL pipelines - ---- - -### ❌ What DOESN'T Work Well: Single-File Parallelism - -**Challenge**: DBN format is inherently sequential - -**Why It's Hard**: -1. **Variable-length records** - Can't seek to arbitrary positions -2. **Must parse to know boundaries** - Can't split file without parsing -3. **I/O bound** - Not CPU bound (bottleneck is reading, not parsing) - -**Attempted Solution** (Producer-Consumer): -```julia -# Producer thread: Read and batch -batches = Channel{Vector{UInt8}}(nthreads * 2) - -@spawn begin - while !eof(decoder.io) - batch = read_batch(decoder, 10_000) # Complex! - put!(batches, batch) - end -end - -# Consumer threads: Parse batches -@threads for _ in 1:nthreads - for batch in batches - records = parse_batch(batch) # Parallel - # ... but ordering matters! - end -end -``` - -**Problems**: -- Batching overhead (must track record boundaries) -- Channel coordination overhead -- Memory pressure (multiple buffers) -- Ordering preservation complexity - -**Benchmarks** (estimated): -- Best case: +50% on 4 cores -- Realistic: +20% (overhead eats gains) -- **Buffered I/O alone: +40%** (simpler, better) - -**Recommendation**: ❌ **Don't implement** -- Too complex for marginal gains -- Buffered I/O is simpler and nearly as good -- I/O bound anyway - ---- - -## The Real Bottleneck: System Calls - -### Current Cost Breakdown (Profiling) - -For reading 1M records: -- **System calls**: ~40% of time - - Each `read(io, Int64)` = 1 syscall - - 10-15 fields per record = 10-15M syscalls - - Each syscall: ~50-100ns overhead - -- **Actual parsing**: ~40% of time - - Type conversions - - Struct construction - - Safety checks - -- **GC**: ~20% of time - - Vector allocation - - String allocations - - Metadata - -### Solution: Buffered I/O - -**Impact**: -``` -Before: 15M syscalls for 1M records -After: ~15K buffer refills (1000x reduction!) - -Time saved: 40% of total time = +66% throughput -``` - -**Implementation Priority**: ⭐⭐⭐⭐⭐ **HIGHEST** - ---- - -## Recommended Next Steps (In Order) - -### Phase 1: Low-Hanging Fruit (This Session) - -1. **Add Parallel File Processing Example** ✓ - - Already works, just needs documentation - - Show batch workload patterns - - Expected: Enables 4x throughput on 4-core systems - -2. **Implement Buffered I/O** (Next) - - 64KB read buffer - - Reduce syscalls by 1000x - - Expected: +40-50% single-file throughput - -3. **Exact Pre-allocation** - - Use metadata record count when available - - Eliminate vector growth - - Expected: +10-15% (cumulative) - -**Combined Impact**: ~60-80% throughput improvement - -### Phase 2: Advanced Optimizations (Future) - -4. **Memory-Mapped Files** - - Good for large uncompressed files - - Expected: +20-30% (specific use cases) - -5. **String Interning** - - Cache repeated symbols - - Expected: +5-10% memory, +2-5% speed - -### Phase 3: Experimental (If Justified) - -6. **SIMD Vectorization** - - For complex records (InstrumentDef, MBP10) - - Expected: +10-20% (specific record types) - -7. **Custom Memory Allocator** - - Arena allocation for records - - Expected: +5-10% (diminishing returns) - ---- - -## Parallel File Processing Example - -Add to README.md: - -```julia -### Parallel Batch Processing - -Process multiple DBN files in parallel for maximum throughput: - -\`\`\`julia -using DBN - -# List of files to process -files = [ - "trades_2024-01-01.dbn", - "trades_2024-01-02.dbn", - "trades_2024-01-03.dbn", - "trades_2024-01-04.dbn" -] - -# Process in parallel (one thread per file) -results = Vector{Vector{DBNRecord}}(undef, length(files)) - -@threads for i in eachindex(files) - results[i] = read_dbn(files[i]) -end - -# Aggregate results -all_records = reduce(vcat, results) - -println("Processed $(length(all_records)) total records") -\`\`\` - -**Performance**: Linear scaling with CPU cores -- 4 cores: ~4x throughput -- 8 cores: ~8x throughput - -**Use Cases**: -- Daily file processing -- Historical data loading -- ETL pipelines -- Backtesting workflows -``` - ---- - -## Why Not Just Use More Threads? - -**Amdahl's Law in Action**: - -Even with perfect parallelization: -``` -Parallel portion: 80% (parsing) -Serial portion: 20% (I/O) - -Max speedup = 1 / (0.2 + 0.8/N) - -N=1: 1.00x -N=2: 1.67x -N=4: 2.50x -N=8: 3.33x -N=∞: 5.00x (limited by serial I/O!) -``` - -**For single file**: I/O is serial → max 5x improvement -**For multiple files**: Each file independent → linear scaling! - -**Conclusion**: Multi-file parallelism >> Single-file parallelism - ---- - -## Performance Roadmap - -**Current**: 2.8M rec/s (pure Julia, fully type-stable) - -**After Buffered I/O**: 4.0-4.5M rec/s (+40-60%) -- Simple implementation -- No API changes -- Works everywhere - -**After All Optimizations**: 5.0-6.0M rec/s (+80-115%) -- Buffered I/O -- Exact pre-allocation -- Memory mapping (where applicable) -- String interning - -**Theoretical Max**: ~7-8M rec/s -- Requires perfect implementation -- Diminishing returns -- May not be worth complexity - -**Gap to Rust**: Will be ~2-3x -- Acceptable for pure Julia -- Rust advantages: zero-cost abstractions, better codegen -- Julia advantages: ecosystem, composability, development speed - ---- - -## Conclusion - -**Best Parallelization Strategy**: Process multiple files in parallel -- Already works perfectly -- Linear scaling -- Zero implementation cost -- Just needs documentation - -**Best Single-File Optimization**: Buffered I/O -- 40-50% improvement -- Simple to implement -- No API changes -- Works for everyone - -**Don't Bother With**: Single-file multi-threading -- Too complex -- Marginal gains -- Better alternatives exist diff --git a/PERFORMANCE_OPTIMIZATION_REPORT.md b/PERFORMANCE_OPTIMIZATION_REPORT.md deleted file mode 100644 index b1fe92f..0000000 --- a/PERFORMANCE_OPTIMIZATION_REPORT.md +++ /dev/null @@ -1,210 +0,0 @@ -# DBN.jl Performance Optimization Report -## Function Barrier Refactoring - -Date: 2025-10-26 - -### Summary - -Successfully implemented function barrier refactoring to address type instability in the `read_record()` function. This optimization reduced allocations by 55% but revealed additional performance bottlenecks. - -### Optimization 1: Vector{DBNRecord} Union Type - -**Problem**: Used `Vector{Any}` to store records, causing boxing overhead - -**Solution**: Created `DBNRecord` union type containing all 18 message types - -```julia -const DBNRecord = Union{ - MBOMsg, TradeMsg, MBP1Msg, MBP10Msg, OHLCVMsg, - StatusMsg, ImbalanceMsg, StatMsg, ErrorMsg, SymbolMappingMsg, SystemMsg, - InstrumentDefMsg, CMBP1Msg, CBBO1sMsg, CBBO1mMsg, TCBBOMsg, BBO1sMsg, BBO1mMsg -} -``` - -**Results**: -- GC time reduced from 80-90% → 52% -- Maintained type safety while supporting multiple record types -- All 3503 tests passing - -### Optimization 2: Function Barrier Refactoring - -**Problem**: 616-line `read_record()` mega-function with all variables typed as `ANY`, causing massive allocations (1.1M for 100K records = 11 per record) - -**Solution**: Split into type-stable helper functions using function barrier pattern: - -1. Compact `read_record()` dispatcher (18 lines) -2. `read_record_dispatch()` with type-stable branches (42 lines) -3. 19 `@inline` type-stable helper functions (one per record type) - -**Code Structure**: - -```julia -function read_record(decoder::DBNDecoder) - if eof(decoder.io) - return nothing - end - hd_result = read_record_header(decoder.io) - if hd_result isa Tuple - _, rtype_raw, record_length = hd_result - skip(decoder.io, record_length - 2) - return nothing - end - hd = hd_result - return read_record_dispatch(decoder, hd, hd.rtype) -end - -@inline function read_record_dispatch(decoder::DBNDecoder, hd::RecordHeader, rtype::RType.T) - if rtype == RType.MBO_MSG - return read_mbo_msg(decoder, hd) - elseif rtype == RType.MBP_0_MSG - return read_trade_msg(decoder, hd) - # ... etc for all 18 record types - else - skip(decoder.io, hd.length - 16) - return nothing - end -end - -@inline function read_mbo_msg(decoder::DBNDecoder, hd::RecordHeader) - # Read fields in binary order - ts_recv = read(decoder.io, Int64) - order_id = read(decoder.io, UInt64) - # ... read all fields - return MBOMsg(hd, order_id, price, size, flags, channel_id, action, side, ts_recv, ts_in_delta, sequence) -end -``` - -**Results** (100K records): -- **Before**: 1.1M allocations (11 per record), 80-90% GC time -- **After**: 499K allocations (5 per record), 78-82% GC time -- **Improvement**: 55% reduction in allocations -- **Status**: All 3503 tests passing - -### Performance Benchmarks (Post-Optimization) - -**Trades 100K records**: -- Read: 770K rec/s (0.77M rec/s) -- Streaming: 915K rec/s (0.92M rec/s) - -**Trades 1M records**: -- Read: 1.22M rec/s -- Streaming: 1.24M rec/s - -**Trades 10M records**: -- Read: 1.00M rec/s -- Streaming: 1.29M rec/s - -**MBO 1M records**: -- Read: 1.15M rec/s -- Streaming: 1.25M rec/s - -### Remaining Performance Bottlenecks - -#### 1. Abstract IO Type in DBNDecoder - -**Problem**: The `DBNDecoder` struct uses abstract `IO` type for fields: - -```julia -mutable struct DBNDecoder - io::IO # ❌ Abstract type - base_io::IO # ❌ Abstract type - header::Union{DBNHeader,Nothing} - metadata::Union{Metadata,Nothing} - upgrade_policy::UInt8 -end -``` - -**Impact**: -- `@code_warntype` shows `decoder.io::IO` and `eof(decoder.io)::ANY` -- Julia cannot specialize on concrete IO type -- Every IO operation requires runtime dispatch -- This is the primary cause of remaining 78-82% GC time - -**Evidence from @code_warntype**: -```julia - decoder::DBNDecoder - %7 = Base.getproperty(decoder, :io)::IO # ← Abstract type! - %8 = (%6)(%7)::ANY # ← Type instability -``` - -**Recommended Solution**: -Use parametric type to support multiple IO types without abstraction: - -```julia -mutable struct DBNDecoder{IO_T <: IO} - io::IO_T # ✓ Concrete type parameter - base_io::IO # Could also be parametrized if needed - header::Union{DBNHeader,Nothing} - metadata::Union{Metadata,Nothing} - upgrade_policy::UInt8 -end -``` - -**Expected Impact**: -- Eliminate remaining type instability -- Reduce GC time from 78% → ~5-10% -- Potentially 5-10x throughput improvement -- Closer to Python wrapper performance (10-12M rec/s) - -#### 2. Vector Growth Strategy - -The current code uses `push!` with `sizehint!`: - -```julia -records = Vector{DBNRecord}(undef, 0) -sizehint!(records, estimated_count) -while !eof(decoder.io) - record = read_record(decoder) - if record !== nothing - push!(records, record) - end -end -``` - -**Potential Improvement**: Pre-allocate exact size if count is known, or use a more efficient append strategy. - -### Testing Results - -- ✅ All 3503 tests passing -- ✅ Byte-for-byte compatibility with Rust implementation maintained -- ✅ No regressions in functionality - -### Files Modified - -1. **src/messages.jl** (lines 704-713): Added `DBNRecord` union type -2. **src/decode.jl** (lines 382-1042): Complete refactoring - - Main `read_record()` function (18 lines) - - `read_record_dispatch()` function (42 lines) - - 19 type-stable `@inline` helper functions - - Fixed `read_instrument_def_v2()` and `read_instrument_def_v3()` return statements - -### Performance Improvement Summary - -| Metric | Before | After | Improvement | -|--------|--------|-------|-------------| -| Allocations (100K records) | 1.1M (11/rec) | 499K (5/rec) | -55% | -| GC Time | 80-90% | 78-82% | ~0% (limited by IO type) | -| Throughput (1M records) | 0.87-1.28M rec/s | 1.15-1.29M rec/s | ~0-10% | -| Code Structure | 616-line mega-function | 21 small functions | Much improved | -| Type Stability | ALL variables `ANY` | Most type-stable | Significantly improved | - -### Next Steps for Further Optimization - -1. **High Priority**: Parametrize `DBNDecoder` with concrete IO type - - Expected: 5-10x throughput improvement - - Expected: GC time reduction to 5-10% - - Expected: Approach Python wrapper performance (10-12M rec/s) - -2. **Medium Priority**: Optimize vector growth strategy - - Pre-allocate when count is known - - Use batch append when possible - -3. **Low Priority**: Profile other hot paths - - String handling in metadata/symbols - - Enum conversions - -### Conclusion - -The function barrier refactoring successfully eliminated the primary type instability issue in `read_record()`, reducing allocations by 55%. However, we've identified that the abstract `IO` type in `DBNDecoder` is now the main performance bottleneck, preventing further improvements. Addressing this issue should yield substantial performance gains and bring Julia's performance much closer to the Python wrapper's 10-12M rec/s. - -The refactoring demonstrates Julia's sensitivity to type stability - even with union types like `DBNRecord`, concrete types enable significant optimization compared to `Any` or abstract types. diff --git a/PERFORMANCE_REPORT.md b/PERFORMANCE_REPORT.md deleted file mode 100644 index 7e3cb2d..0000000 --- a/PERFORMANCE_REPORT.md +++ /dev/null @@ -1,50 +0,0 @@ -# DBN.jl Performance Benchmarks - -## mbo.1m.dbn (1,000,000 records, 53.41 MB) - -| Implementation | Throughput | -|----------------|------------| -| Julia `read_mbo()` (optimized) | 17.07 M/s | -| Python `databento` | 10.86 M/s | -| Julia `DBNStream()` | 3.17 M/s | -| Julia `read_dbn()` | 3.00 M/s | -| Julia `write_dbn()` | 2.38 M/s | -| Rust `dbn` CLI | 2.29 M/s | - -## trades.1m.dbn (1,000,000 records, 45.78 MB) - -| Implementation | Throughput | -|----------------|------------| -| Julia `read_trades()` (optimized) | 17.88 M/s | -| Python `databento` | 11.87 M/s | -| Julia `DBNStream()` | 3.14 M/s | -| Julia `read_dbn()` | 3.10 M/s | -| Rust `dbn` CLI | 2.67 M/s | -| Julia `write_dbn()` | 2.20 M/s | - -## trades.10m.dbn (10,000,000 records, 457.76 MB) - -| Implementation | Throughput | -|----------------|------------| -| Julia `read_trades()` (optimized) | 19.30 M/s | -| Python `databento` | 11.25 M/s | -| Julia `DBNStream()` | 9.36 M/s | -| Julia `read_dbn()` | 5.55 M/s | -| Rust `dbn` CLI | 2.76 M/s | - -## trades.100k.dbn (100,000 records, 4.58 MB) - -| Implementation | Throughput | -|----------------|------------| -| Julia `read_trades()` (optimized) | 15.06 M/s | -| Python `databento` | 10.05 M/s | -| Julia `write_dbn()` | 2.75 M/s | -| Rust `dbn` CLI | 1.97 M/s | -| Julia `DBNStream()` | 0.44 M/s | -| Julia `read_dbn()` | 0.43 M/s | - ---- - -**Test environment**: Windows 11, Julia 1.12.1, Python databento 0.x, Rust dbn 1.75+ - -**Reproduce**: `julia --project=. benchmark/compare_all_comprehensive.jl` diff --git a/PERFORMANCE_REPORT.md.bak b/PERFORMANCE_REPORT.md.bak deleted file mode 100644 index 1f555b3..0000000 --- a/PERFORMANCE_REPORT.md.bak +++ /dev/null @@ -1,220 +0,0 @@ -# DBN.jl Performance Report - -## Executive Summary - -After systematic profiling and optimization, **DBN.jl now matches or exceeds Rust performance** on production workloads while maintaining a clean, Julian API. - -### Key Results - -| Workload | Performance | vs Rust | Improvement | -|----------|-------------|---------|-------------| -| **trades.1m.dbn** (Union API) | **7.97 M rec/s** | ~0.8x | **+184%** from baseline | -| **trades.1m.dbn** (Typed API) | **17.5 M rec/s** | **1.75x faster** | **+942%** from baseline | -| **trades.10m.dbn** (Streaming) | **10.8 M rec/s** | **~1.1x faster** | **+367%** from baseline | - ---- - -## Performance Evolution - -### Baseline → Optimized (Union API) - -| File | Baseline | After Opts | Improvement | -|------|----------|------------|-------------| -| trades.1m.dbn | 2.81 M rec/s | **7.97 M rec/s** | **+184%** | -| trades.10m.dbn | 2.42 M rec/s | **4.88 M rec/s** | **+102%** | -| mbo.1m.dbn | 2.61 M rec/s | **6.16 M rec/s** | **+136%** | - -### Type-Specialized API (New!) - -| Function | Throughput | Speedup vs Union | -|----------|------------|------------------| -| `read_trades()` | **17.5 M rec/s** | **5.73x** | -| `read_mbo()` | **17+ M rec/s** | **5.73x** | -| `read_mbp1()` | **17+ M rec/s** | **5.73x** | - ---- - -## Optimization Breakdown - -### 1. Buffered I/O -- **Impact**: +67% throughput -- **Approach**: 64KB buffer reduces syscalls from thousands to dozens -- **Code**: `BufferedReader` wrapper with fast path for primitive types - -### 2. String Handling -- **Impact**: 9x faster on string operations -- **Before**: `String(strip(String(bytes), '\0'))` - double allocation -- **After**: `read_null_terminated_string()` - find null, create once - -### 3. Unsafe Enum Conversions -- **Impact**: 1000x+ faster enum conversions -- **Before**: `safe_action()` with try-catch (22,500 ns) -- **After**: `unsafe_action()` direct conversion (~2 ns) -- **Safety**: Only used when reading well-formed DBN files - -### 4. Type-Specialized Readers (Breakthrough!) -- **Impact**: 5.73x faster by eliminating Union GC overhead -- **Profiling insight**: 80% of time in GC tracking Union objects -- **Solution**: `Vector{TradeMsg}` instead of `Vector{Union{...}}` - ---- - -## API Design - -### Flexible API (Default) -```julia -# Works for all cases, automatic type handling -records = read_dbn("mixed_types.dbn") # 7.97 M rec/s -``` - -### Performance API (When Schema Known) -```julia -# 5.7x faster - no Union overhead -trades = read_trades("trades.dbn") # 17.5 M rec/s -mbos = read_mbo("mbo.dbn") # 17+ M rec/s -depth = read_mbp1("depth.dbn") # 17+ M rec/s -``` - -### Streaming API (Large Files) -```julia -# Memory-efficient, excellent throughput -for record in DBNStream("huge.dbn") # 10.8 M rec/s - process(record) -end -``` - ---- - -## Comparison with Other Implementations - -### Julia vs Rust (Official Implementation) - -| Benchmark | Julia (Union) | Julia (Typed) | Rust CLI | Winner | -|-----------|---------------|---------------|----------|--------| -| trades.1m.dbn | 7.97 M rec/s | **17.5 M rec/s** | 10-12 M rec/s | **Julia** | -| trades.10m.dbn | 4.88 M rec/s | **10.8 M rec/s** | ~10 M rec/s | **Julia** | -| Small files (<100k) | 0.2-1.3 M rec/s | 1-17 M rec/s | 2-5 M rec/s | Rust (JIT overhead) | - -**Conclusion**: -- **Large files (production)**: Julia 1.1-1.75x faster than Rust -- **Small files**: Rust faster due to Julia JIT compilation overhead (~220ms) - -### Why Julia Wins on Large Files - -1. **LLVM optimizations**: Julia's JIT produces highly optimized machine code -2. **Inlining**: `@inline` macros eliminate function call overhead -3. **SIMD**: Julia's array operations can vectorize -4. **Type stability**: Monomorphic code paths (in typed API) -5. **Memory layout**: isbitstype structs stored inline in arrays - -### Why Rust Wins on Small Files - -1. **No JIT**: Ahead-of-time compilation means zero startup cost -2. **For <100k records**: 220ms JIT overhead dominates total time - ---- - -## Memory & Allocation Analysis - -### BenchmarkTools Results (100k records) - -| Operation | Time | Memory | Allocations | -|-----------|------|--------|-------------| -| Read (medium, uncompressed) | 56.3 ms | 6.93 MB | 100,047 | -| Stream (medium) | 50.1 ms | 12.27 MB | 200,044 | -| Write (medium) | 32.9 ms | 19.82 MB | 1,199,009 | - -### Allocation Breakdown -- **Per record**: ~69 bytes (includes Union tag + padding) -- **Typed API**: ~48 bytes (no Union overhead) -- **GC pressure**: Significantly reduced with typed readers - ---- - -## Profiling Insights - -### Hot Path Analysis (1M records) - -| Component | Time % | Samples | -|-----------|--------|---------| -| **Garbage Collection** | 80% | 468/585 | -| Array operations (_setindex!) | 6% | 35/585 | -| Enum conversions | 3.6% | 21/585 | -| I/O operations | 2.7% | 16/585 | -| Record parsing | <2% | <12/585 | - -**Key Finding**: GC overhead from Union tracking was the bottleneck, hence the 5.7x speedup from type-specialized readers. - ---- - -## Performance Tuning Guidelines - -### When to Use Each API - -1. **`read_dbn()`** - Flexible, works for all cases - - Mixed record types in same file - - Unknown schema at compile time - - Acceptable 8 M rec/s performance - -2. **`read_trades()` / `read_mbo()`** - Maximum performance - - Known single record type - - Need 17+ M rec/s throughput - - Production data processing - -3. **`DBNStream()`** - Large files - - Files too large for RAM - - Streaming processing pipeline - - 10+ M rec/s with minimal memory - -### Optimization Checklist - -- ✅ Use typed readers when schema is known -- ✅ Use streaming for files >1GB -- ✅ Let Julia JIT compile before benchmarking (warmup) -- ✅ Process in batches to reduce GC frequency -- ⚠️ Avoid `read_dbn()` in tight loops (use streaming) -- ⚠️ Small files (<100k) have JIT overhead - ---- - -## Technical Details - -### Test Environment -- **Platform**: Windows 11 -- **CPU**: (varies by system) -- **Julia**: 1.12.1 -- **Rust**: 1.75+ (dbn crate) -- **Method**: Mean of 5 runs after warmup - -### Reproducibility -```bash -# Run full benchmark suite -julia --project=. benchmark/run_benchmarks.jl - -# Test typed readers -julia --project=. -e 'using DBN; @time read_trades("benchmark/data/trades.1m.dbn")' - -# Compare with Rust -./dbn-cli decode benchmark/data/trades.1m.dbn > /dev/null -``` - ---- - -## Conclusion - -DBN.jl achieves **competitive or superior performance** to the official Rust implementation through: - -1. **Systematic profiling** to identify bottlenecks (GC overhead) -2. **Targeted optimizations** (buffered I/O, string handling, enum conversions) -3. **API innovation** (type-specialized readers for zero GC overhead) -4. **Julia-native patterns** (no manual GC management, type stability) - -The result is a **fast, flexible, and Julian** implementation that gives users both convenience and performance options. - -**Performance achieved**: Up to **17.5 M records/second** - faster than Rust while maintaining clean, idiomatic Julia code. - ---- - -*Report generated: 2025-10-26* -*Julia version: 1.12.1* -*DBN.jl version: 0.1.0* diff --git a/Project.toml b/Project.toml index 57750e3..73a19fa 100644 --- a/Project.toml +++ b/Project.toml @@ -4,7 +4,6 @@ authors = ["Tyler Beason "] version = "0.1.0" [deps] -BenchmarkTools = "6e4b80f9-dd63-53aa-95a3-0cdb28fa8baf" CSV = "336ed68f-0bac-5ca0-87d4-7b16caf5d00b" CodecZstd = "6b39b394-51ab-5f42-8807-6242bab2b4c2" DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0" diff --git a/STREAMING_OPTIMIZATION_SUMMARY.md b/STREAMING_OPTIMIZATION_SUMMARY.md deleted file mode 100644 index 8f2e65e..0000000 --- a/STREAMING_OPTIMIZATION_SUMMARY.md +++ /dev/null @@ -1,112 +0,0 @@ -# Streaming Optimization Summary - -## Overview -Investigation into benchmark performance issues led to discovery and resolution of a critical performance bottleneck in `DBNStream` iterator implementation. - -## Issues Discovered - -### 1. Python Benchmark Bugs (Fixed) -- **TypeError**: `DBNStore` object doesn't support `len()` - fixed by using `sum(1 for _ in data)` or `list(data)` -- **Materialization issue**: Python reads were not materializing data for fair comparison with Julia -- **Write benchmark incomparability**: Python's `to_file()` does optimized byte copying, not deserialization+reserialization like Julia - -### 2. Benchmark Configuration (Optimized) -- Reduced benchmark time: 5s → 2s per benchmark -- Reduced samples: 100 → 20 (sufficient for stable median) -- Removed unnecessary 0.05s sleep delays in Python benchmarks -- Total benchmark runtime: ~30min → ~8min - -### 3. Streaming Performance Bottleneck (Fixed) -**Root cause**: Passing `DBNDecoder` as iterator state caused massive allocation overhead -- Iterator protocol allocated ~10k extra objects per 100k records (2x overhead) -- Each `iterate()` call was boxing/unboxing the decoder state - -**Solution**: Store decoder in `DBNStream` struct instead of passing as state -```julia -# Before: Decoder passed as state -Base.iterate(stream::DBNStream) = begin - decoder = DBNDecoder(stream.filename) - return iterate(stream, decoder) # decoder becomes state -end - -# After: Decoder stored in struct -mutable struct DBNStream - decoder::DBNDecoder - cleanup::Ref{Bool} -end -``` - -## Performance Results - -### Streaming Performance Improvement (100k records) -| Metric | Before | After | Improvement | -|--------|--------|-------|-------------| -| Speed | 0.42 M/s | 7.42 M/s | **18x faster** | -| Time | 236 ms | 13 ms | **18x faster** | - -### Complete Performance Comparison (10M records) - -| Method | Speed | Time | Memory | Use Case | -|--------|-------|------|--------|----------| -| **Streaming** | 7.16 M/s | 1.397s | 1221 MB | Process data on-the-fly, larger-than-memory | -| Eager read | 5.32 M/s | 1.879s | 687 MB | Load all data, standard use | -| **Optimized eager** | 17.89 M/s | 0.559s | 458 MB | Known schema, load all data | - -**Key takeaway**: Streaming is now **35% faster** than regular eager read for large files! - -### Performance by File Size - -| Size | Streaming | Eager Read | Optimized | Winner | -|------|-----------|------------|-----------|--------| -| 1k | 9.59 M/s | 0.00 M/s* | 15.09 M/s | Optimized | -| 10k | 11.35 M/s | 0.04 M/s* | 20.38 M/s | Optimized | -| 100k | 7.42 M/s | 0.41 M/s | 19.60 M/s | Optimized | -| 1M | 6.70 M/s | 2.70 M/s | 18.31 M/s | Optimized | -| 10M | 7.16 M/s | 5.32 M/s | 17.89 M/s | Optimized | - -*Compilation overhead dominates small files - -## Recommendations - -### For Users - -1. **Larger-than-memory datasets**: Use `DBNStream` (now optimized!) - ```julia - for record in DBNStream("large_file.dbn") - process(record) # Constant memory usage - end - ``` - -2. **Known schema, need all data**: Use optimized readers - ```julia - trades = read_trades("data.dbn") # 18 M/s - mbos = read_mbo("data.dbn") # Similar performance - ``` - -3. **Unknown/mixed schema**: Use `read_dbn()` - ```julia - records = read_dbn("data.dbn") # Works with any schema - ``` - -### For Package Development - -1. **Streaming is now the priority** for large-file workflows -2. **Optimized readers provide best eager performance** when schema is known -3. **Python comparison considerations**: - - Read benchmarks are now comparable (both materialize) - - Write benchmarks are NOT comparable (different operations) - -## Testing -- All 3503 tests pass ✓ -- Streaming functionality verified correct -- Performance improvements confirmed across all file sizes - -## Files Changed -- `src/streaming.jl`: Refactored `DBNStream` to store decoder in struct -- `benchmark/compare_all_comprehensive.jl`: Fixed Python benchmarks, optimized timing -- Added profiling and testing scripts - -## Next Steps (Potential) -1. Investigate why streaming uses more memory than expected in BenchmarkTools -2. Consider adding a fast-path copy function for file transformation use cases -3. Document Python write benchmark limitations in main README diff --git a/add_instdef_helpers.py b/add_instdef_helpers.py deleted file mode 100644 index aedce17..0000000 --- a/add_instdef_helpers.py +++ /dev/null @@ -1,75 +0,0 @@ -#!/usr/bin/env python3 -"""Add the missing InstrumentDef v2/v3 helper functions""" - -# Read the current file -with open("src/decode.jl", "r", encoding="utf-8") as f: - lines = f.readlines() - -print(f"Current file: {len(lines)} lines") - -# Find the insertion point (after read_instrument_def_msg, line 808) -insert_at = 808 # 0-indexed - -# Extract v2 and v3 code from the backup -with open("src/decode.jl.backup_before_refactor", "r", encoding="utf-8") as f: - backup_lines = f.readlines() - -# Find v2 block in backup (starts at line 514, ends before 605) -v2_start = 513 # Line 514 in 1-indexed -v2_end = 604 # Line 605 - -# Find v3 block in backup (starts at line 608, ends at 728) -v3_start = 607 # Line 608 -v3_end = 728 # Line 729 - -# Build v2 function -v2_func = ["@inline function read_instrument_def_v2(decoder::DBNDecoder, hd::RecordHeader)\n"] -v2_func.append(" # ===== DBN V2 InstrumentDefMsg =====\n") -v2_func.append(" # CRITICAL: In v2, encode_order attributes are COMPLETELY IGNORED!\n") -v2_func.append(" # Read ALL fields in exact Rust struct declaration order\n") - -# Add v2 body (remove 12 spaces of indentation, add 4) -for i in range(v2_start, v2_end): - line = backup_lines[i] - if line.strip(): # Skip empty lines at start/end - # Remove 12 spaces of indentation, add 4 - if line.startswith(" "): - line = " " + line[12:] - v2_func.append(line) - -v2_func.append("end\n") -v2_func.append("\n") - -# Build v3 function -v3_func = ["@inline function read_instrument_def_v3(decoder::DBNDecoder, hd::RecordHeader)\n"] -v3_func.append(" # ===== DBN V3 InstrumentDefMsg =====\n") - -# Add v3 body -for i in range(v3_start, v3_end): - line = backup_lines[i] - if line.strip(): - # Remove 12 spaces of indentation, add 4 - if line.startswith(" "): - line = " " + line[12:] - v3_func.append(line) - -v3_func.append("end\n") -v3_func.append("\n") - -print(f"v2 function: {len(v2_func)} lines") -print(f"v3 function: {len(v3_func)} lines") - -# Insert into file -new_lines = ( - lines[:insert_at] + - v2_func + - v3_func + - lines[insert_at:] -) - -# Write back -with open("src/decode.jl", "w", encoding="utf-8", newline='\n') as f: - f.writelines(new_lines) - -print(f"New file: {len(new_lines)} lines (was {len(lines)})") -print("Added v2 and v3 helper functions successfully!") diff --git a/add_missing_helpers.jl b/add_missing_helpers.jl deleted file mode 100644 index 037c370..0000000 --- a/add_missing_helpers.jl +++ /dev/null @@ -1,198 +0,0 @@ -#!/usr/bin/env julia -# Add the missing type-stable helper functions to decode.jl - -println("Adding missing helper functions to decode.jl...") - -# Read current refactored file -lines = readlines("src/decode.jl") - -# Find insertion point (after read_record_dispatch ends, before read_imbalance_msg) -insert_line = findfirst(l -> occursin("# Remaining type-stable reader functions", l), lines) -if insert_line === nothing - error("Could not find insertion point") -end - -println("Inserting at line $insert_line") - -# Read the backup to extract original implementations -backup_lines = readlines("src/decode.jl.backup_before_refactor") - -# Helper functions to add (extracted from original mega-function) -new_functions = String[] - -# Add header comment -push!(new_functions, "# Type-stable reader functions for each record type - eliminates boxing/unboxing") -push!(new_functions, "") - -# 1. MBO Message -push!(new_functions, "@inline function read_mbo_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " ts_recv = read(decoder.io, Int64)") -push!(new_functions, " order_id = read(decoder.io, UInt64)") -push!(new_functions, " size = read(decoder.io, UInt32)") -push!(new_functions, " flags = read(decoder.io, UInt8)") -push!(new_functions, " channel_id = read(decoder.io, UInt8)") -push!(new_functions, " action = safe_action(read(decoder.io, UInt8))") -push!(new_functions, " side = safe_side(read(decoder.io, UInt8))") -push!(new_functions, " price = read(decoder.io, Int64)") -push!(new_functions, " ts_in_delta = read(decoder.io, Int32)") -push!(new_functions, " sequence = read(decoder.io, UInt32)") -push!(new_functions, " return MBOMsg(hd, order_id, price, size, flags, channel_id, action, side, ts_recv, ts_in_delta, sequence)") -push!(new_functions, "end") -push!(new_functions, "") - -# 2. Trade Message (MBP_0) -push!(new_functions, "@inline function read_trade_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " price = read(decoder.io, Int64)") -push!(new_functions, " size = read(decoder.io, UInt32)") -push!(new_functions, " action = safe_action(read(decoder.io, UInt8))") -push!(new_functions, " side = safe_side(read(decoder.io, UInt8))") -push!(new_functions, " flags = read(decoder.io, UInt8)") -push!(new_functions, " depth = read(decoder.io, UInt8)") -push!(new_functions, " ts_recv = read(decoder.io, Int64)") -push!(new_functions, " ts_in_delta = read(decoder.io, Int32)") -push!(new_functions, " sequence = read(decoder.io, UInt32)") -push!(new_functions, " return TradeMsg(hd, price, size, action, side, flags, depth, ts_recv, ts_in_delta, sequence)") -push!(new_functions, "end") -push!(new_functions, "") - -# 3. MBP1 Message -push!(new_functions, "@inline function read_mbp1_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " price = read(decoder.io, Int64)") -push!(new_functions, " size = read(decoder.io, UInt32)") -push!(new_functions, " action = safe_action(read(decoder.io, UInt8))") -push!(new_functions, " side = safe_side(read(decoder.io, UInt8))") -push!(new_functions, " flags = read(decoder.io, UInt8)") -push!(new_functions, " depth = read(decoder.io, UInt8)") -push!(new_functions, " ts_recv = read(decoder.io, Int64)") -push!(new_functions, " ts_in_delta = read(decoder.io, Int32)") -push!(new_functions, " sequence = read(decoder.io, UInt32)") -push!(new_functions, " bid_px = read(decoder.io, Int64)") -push!(new_functions, " ask_px = read(decoder.io, Int64)") -push!(new_functions, " bid_sz = read(decoder.io, UInt32)") -push!(new_functions, " ask_sz = read(decoder.io, UInt32)") -push!(new_functions, " bid_ct = read(decoder.io, UInt32)") -push!(new_functions, " ask_ct = read(decoder.io, UInt32)") -push!(new_functions, " levels = BidAskPair(bid_px, ask_px, bid_sz, ask_sz, bid_ct, ask_ct)") -push!(new_functions, " return MBP1Msg(hd, price, size, action, side, flags, depth, ts_recv, ts_in_delta, sequence, levels)") -push!(new_functions, "end") -push!(new_functions, "") - -# 4. MBP10 Message -push!(new_functions, "@inline function read_mbp10_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " price = read(decoder.io, Int64)") -push!(new_functions, " size = read(decoder.io, UInt32)") -push!(new_functions, " action = safe_action(read(decoder.io, UInt8))") -push!(new_functions, " side = safe_side(read(decoder.io, UInt8))") -push!(new_functions, " flags = read(decoder.io, UInt8)") -push!(new_functions, " depth = read(decoder.io, UInt8)") -push!(new_functions, " ts_recv = read(decoder.io, Int64)") -push!(new_functions, " ts_in_delta = read(decoder.io, Int32)") -push!(new_functions, " sequence = read(decoder.io, UInt32)") -push!(new_functions, " levels = ntuple(10) do _") -push!(new_functions, " bid_px = read(decoder.io, Int64)") -push!(new_functions, " ask_px = read(decoder.io, Int64)") -push!(new_functions, " bid_sz = read(decoder.io, UInt32)") -push!(new_functions, " ask_sz = read(decoder.io, UInt32)") -push!(new_functions, " bid_ct = read(decoder.io, UInt32)") -push!(new_functions, " ask_ct = read(decoder.io, UInt32)") -push!(new_functions, " BidAskPair(bid_px, ask_px, bid_sz, ask_sz, bid_ct, ask_ct)") -push!(new_functions, " end") -push!(new_functions, " return MBP10Msg(hd, price, size, action, side, flags, depth, ts_recv, ts_in_delta, sequence, levels)") -push!(new_functions, "end") -push!(new_functions, "") - -# 5. OHLCV Message -push!(new_functions, "@inline function read_ohlcv_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " open = read(decoder.io, Int64)") -push!(new_functions, " high = read(decoder.io, Int64)") -push!(new_functions, " low = read(decoder.io, Int64)") -push!(new_functions, " close = read(decoder.io, Int64)") -push!(new_functions, " volume = read(decoder.io, UInt64)") -push!(new_functions, " return OHLCVMsg(hd, open, high, low, close, volume)") -push!(new_functions, "end") -push!(new_functions, "") - -# 6. Status Message -push!(new_functions, "@inline function read_status_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " ts_recv = read(decoder.io, UInt64)") -push!(new_functions, " action = read(decoder.io, UInt16)") -push!(new_functions, " reason = read(decoder.io, UInt16)") -push!(new_functions, " trading_event = read(decoder.io, UInt16)") -push!(new_functions, " is_trading = read(decoder.io, UInt8)") -push!(new_functions, " is_quoting = read(decoder.io, UInt8)") -push!(new_functions, " is_short_sell_restricted = read(decoder.io, UInt8)") -push!(new_functions, " _ = read(decoder.io, 7)") -push!(new_functions, " return StatusMsg(hd, ts_recv, action, reason, trading_event, is_trading, is_quoting, is_short_sell_restricted)") -push!(new_functions, "end") -push!(new_functions, "") - -# 7. InstrumentDef dispatcher -push!(new_functions, "@inline function read_instrument_def_msg(decoder::DBNDecoder, hd::RecordHeader)") -push!(new_functions, " start_pos = position(decoder.io)") -push!(new_functions, " record_size_bytes = hd.length * LENGTH_MULTIPLIER") -push!(new_functions, " body_size = record_size_bytes - 16") -push!(new_functions, " if body_size == 384") -push!(new_functions, " return read_instrument_def_v2(decoder, hd)") -push!(new_functions, " else") -push!(new_functions, " return read_instrument_def_v3(decoder, hd)") -push!(new_functions, " end") -push!(new_functions, "end") -push!(new_functions, "") - -# Note: The full v2 and v3 implementations are too long to include inline here -# They should already be in the backup file. Let's extract them. -println("Extracting InstrumentDef v2 and v3 implementations...") - -# Find the InstrumentDef v2 block (lines with body_size == 384) -v2_start = findfirst(l -> occursin("if body_size == 384", l), backup_lines) -v2_end = findnext(l -> occursin("else", l) && occursin("DBN V3", backup_lines[findnext(x->x==l, backup_lines, 1)+1]), backup_lines, v2_start) - -if v2_start !== nothing && v2_end !== nothing - println("Extracting V2: lines $v2_start to $(v2_end-1)") - # Convert the v2 block into a function - push!(new_functions, "@inline function read_instrument_def_v2(decoder::DBNDecoder, hd::RecordHeader)") - # Extract v2 body (skip the if statement, take everything until else) - for i in (v2_start+2):(v2_end-2) - line = backup_lines[i] - # Adjust indentation - push!(new_functions, replace(line, r"^ " => " ")) - end - push!(new_functions, "end") - push!(new_functions, "") -end - -# Extract V3 -v3_start = v2_end -v3_end = findnext(l -> occursin("return InstrumentDefMsg", l), backup_lines, v3_start+10) -if v3_end !== nothing - v3_end = findnext(l -> strip(l) == ")", backup_lines, v3_end) -end - -if v3_start !== nothing && v3_end !== nothing - println("Extracting V3: lines $v3_start to $v3_end") - push!(new_functions, "@inline function read_instrument_def_v3(decoder::DBNDecoder, hd::RecordHeader)") - for i in (v3_start+2):(v3_end) - line = backup_lines[i] - # Adjust indentation - push!(new_functions, replace(line, r"^ " => " ")) - end - push!(new_functions, "end") - push!(new_functions, "") -end - -# Now insert the new functions into the file -new_lines = vcat( - lines[1:insert_line-1], - new_functions, - lines[insert_line:end] -) - -# Write the updated file -open("src/decode.jl", "w") do f - for line in new_lines - println(f, line) - end -end - -println("✓ Added $(length(new_functions)) lines of helper functions") -println("New file: $(length(new_lines)) lines (was $(length(lines)))") diff --git a/refactor.py b/refactor.py deleted file mode 100644 index 6e44961..0000000 --- a/refactor.py +++ /dev/null @@ -1,274 +0,0 @@ -#!/usr/bin/env python3 -""" -Function barrier refactoring for read_record() -Converts the 616-line type-unstable mega-function into type-stable helpers -""" - -def main(): - print("Function Barrier Refactoring") - print("=" * 70) - - # Read the original file - with open("src/decode.jl", "r", encoding="utf-8") as f: - lines = f.readlines() - - print(f"Original file: {len(lines)} lines") - - # read_record is at lines 382-998 (1-indexed, so 381-997 in 0-indexed) - read_record_start = 381 # Line 382 in 1-indexed - read_record_end = 997 # Line 998 in 1-indexed - - # Verify - assert "function read_record(decoder::DBNDecoder)" in lines[read_record_start] - assert lines[read_record_end].strip() == "end" - - print(f"Found read_record: lines {read_record_start+1} to {read_record_end+1}") - print(f"Function size: {read_record_end - read_record_start + 1} lines") - print() - - # Build the new code - new_code = [] - - # Main read_record function (compact) - new_code.extend([ - "function read_record(decoder::DBNDecoder)\n", - " if eof(decoder.io)\n", - " return nothing\n", - " end\n", - " \n", - " hd_result = read_record_header(decoder.io)\n", - " \n", - " # Handle unknown record types\n", - " if hd_result isa Tuple\n", - " _, rtype_raw, record_length = hd_result\n", - " skip(decoder.io, record_length - 2)\n", - " return nothing\n", - " end\n", - " \n", - " hd = hd_result\n", - " \n", - " # Type-stable dispatch - function barrier eliminates type instability\n", - " return read_record_dispatch(decoder, hd, hd.rtype)\n", - "end\n", - "\n", - ]) - - # Dispatch function - new_code.extend([ - "# Dispatch to type-stable helpers - each helper has concrete types\n", - "@inline function read_record_dispatch(decoder::DBNDecoder, hd::RecordHeader, rtype::RType.T)\n", - " if rtype == RType.MBO_MSG\n", - " return read_mbo_msg(decoder, hd)\n", - " elseif rtype == RType.MBP_0_MSG\n", - " return read_trade_msg(decoder, hd)\n", - " elseif rtype == RType.MBP_1_MSG\n", - " return read_mbp1_msg(decoder, hd)\n", - " elseif rtype == RType.MBP_10_MSG\n", - " return read_mbp10_msg(decoder, hd)\n", - " elseif rtype in (RType.OHLCV_1S_MSG, RType.OHLCV_1M_MSG, RType.OHLCV_1H_MSG, RType.OHLCV_1D_MSG)\n", - " return read_ohlcv_msg(decoder, hd)\n", - " elseif rtype == RType.STATUS_MSG\n", - " return read_status_msg(decoder, hd)\n", - " elseif rtype == RType.INSTRUMENT_DEF_MSG\n", - " return read_instrument_def_msg(decoder, hd)\n", - " elseif rtype == RType.IMBALANCE_MSG\n", - " return read_imbalance_msg(decoder, hd)\n", - " elseif rtype == RType.STAT_MSG\n", - " return read_stat_msg(decoder, hd)\n", - " elseif rtype == RType.ERROR_MSG\n", - " return read_error_msg(decoder, hd)\n", - " elseif rtype == RType.SYMBOL_MAPPING_MSG\n", - " return read_symbol_mapping_msg(decoder, hd)\n", - " elseif rtype == RType.SYSTEM_MSG\n", - " return read_system_msg(decoder, hd)\n", - " elseif rtype == RType.CMBP_1_MSG\n", - " return read_cmbp1_msg(decoder, hd)\n", - " elseif rtype == RType.CBBO_1S_MSG\n", - " return read_cbbo1s_msg(decoder, hd)\n", - " elseif rtype == RType.CBBO_1M_MSG\n", - " return read_cbbo1m_msg(decoder, hd)\n", - " elseif rtype == RType.TCBBO_MSG\n", - " return read_tcbbo_msg(decoder, hd)\n", - " elseif rtype == RType.BBO_1S_MSG\n", - " return read_bbo1s_msg(decoder, hd)\n", - " elseif rtype == RType.BBO_1M_MSG\n", - " return read_bbo1m_msg(decoder, hd)\n", - " else\n", - " skip(decoder.io, hd.length - 16)\n", - " return nothing\n", - " end\n", - "end\n", - "\n", - ]) - - # Now extract and convert each handler from the original function - old_function = lines[read_record_start:read_record_end+1] - - # Helper to convert an if/elseif block to a function - def extract_handler(start_marker, func_name): - # Find start - start_idx = None - for i, line in enumerate(old_function): - if start_marker in line: - start_idx = i - break - - if start_idx is None: - print(f"Warning: Could not find {func_name}") - return [] - - # Find the return statement (end of this handler) - end_idx = None - for i in range(start_idx + 1, len(old_function)): - if "return " in old_function[i] and ("Msg(" in old_function[i] or "nothing" in old_function[i]): - end_idx = i - break - - if end_idx is None: - print(f"Warning: Could not find end for {func_name}") - return [] - - # Build the function - func_lines = [f"@inline function {func_name}(decoder::DBNDecoder, hd::RecordHeader)\n"] - - # Add the body (skip the if/elseif line, include up to and including return) - for i in range(start_idx + 1, end_idx + 1): - line = old_function[i] - # Remove one level of indentation (8 spaces) - if line.startswith(" "): - line = " " + line[8:] - func_lines.append(line) - - func_lines.append("end\n\n") - return func_lines - - # Extract all handlers - print("Extracting handlers...") - - # Simple handlers - handlers = [ - ("if hd.rtype == RType.MBO_MSG", "read_mbo_msg"), - ("elseif hd.rtype == RType.MBP_0_MSG", "read_trade_msg"), - ("elseif hd.rtype == RType.MBP_1_MSG", "read_mbp1_msg"), - ("elseif hd.rtype == RType.MBP_10_MSG", "read_mbp10_msg"), - ("elseif hd.rtype in [RType.OHLCV", "read_ohlcv_msg"), - ("elseif hd.rtype == RType.STATUS_MSG", "read_status_msg"), - ("elseif hd.rtype == RType.IMBALANCE_MSG", "read_imbalance_msg"), - ("elseif hd.rtype == RType.STAT_MSG", "read_stat_msg"), - ("elseif hd.rtype == RType.ERROR_MSG", "read_error_msg"), - ("elseif hd.rtype == RType.SYMBOL_MAPPING_MSG", "read_symbol_mapping_msg"), - ("elseif hd.rtype == RType.SYSTEM_MSG", "read_system_msg"), - ("elseif hd.rtype == RType.CMBP_1_MSG", "read_cmbp1_msg"), - ("elseif hd.rtype == RType.CBBO_1S_MSG", "read_cbbo1s_msg"), - ("elseif hd.rtype == RType.CBBO_1M_MSG", "read_cbbo1m_msg"), - ("elseif hd.rtype == RType.TCBBO_MSG", "read_tcbbo_msg"), - ("elseif hd.rtype == RType.BBO_1S_MSG", "read_bbo1s_msg"), - ("elseif hd.rtype == RType.BBO_1M_MSG", "read_bbo1m_msg"), - ] - - for marker, func_name in handlers: - print(f" - {func_name}") - handler_code = extract_handler(marker, func_name) - new_code.extend(handler_code) - - # InstrumentDef is special - it's huge and has v2/v3 variants - # Handle it separately - print(" - read_instrument_def_msg (complex)") - - # Find the InstrumentDef block - inst_start = None - for i, line in enumerate(old_function): - if "elseif hd.rtype == RType.INSTRUMENT_DEF_MSG" in line: - inst_start = i - break - - if inst_start: - # Add the main dispatcher - new_code.extend([ - "@inline function read_instrument_def_msg(decoder::DBNDecoder, hd::RecordHeader)\n", - " start_pos = position(decoder.io)\n", - " record_size_bytes = hd.length * LENGTH_MULTIPLIER\n", - " body_size = record_size_bytes - 16\n", - " if body_size == 384\n", - " return read_instrument_def_v2(decoder, hd)\n", - " else\n", - " return read_instrument_def_v3(decoder, hd)\n", - " end\n", - "end\n", - "\n", - ]) - - # Find v2 block (starts at "if body_size == 384", ends before "else") - v2_start = None - for i in range(inst_start, len(old_function)): - if "if body_size == 384" in old_function[i]: - v2_start = i - break - - v2_end = None - if v2_start: - for i in range(v2_start + 1, len(old_function)): - if old_function[i].strip().startswith("else") and "DBN V3" in old_function[min(i+2, len(old_function)-1)]: - v2_end = i - break - - if v2_start and v2_end: - print(" - read_instrument_def_v2") - new_code.append("@inline function read_instrument_def_v2(decoder::DBNDecoder, hd::RecordHeader)\n") - # Skip comments and "if body_size" line, take body - for i in range(v2_start + 11, v2_end - 1): # Skip header comments - line = old_function[i] - if line.startswith(" "): - line = " " + line[12:] - new_code.append(line) - new_code.append("end\n\n") - - # Find v3 block - v3_start = v2_end - v3_end = None - if v3_start: - # Find the big return statement for InstrumentDefMsg - for i in range(v3_start, len(old_function)): - if "return InstrumentDefMsg(" in old_function[i]: - # Find the closing paren - for j in range(i, len(old_function)): - if old_function[j].strip() == ")": - v3_end = j - break - break - - if v3_start and v3_end: - print(" - read_instrument_def_v3") - new_code.append("@inline function read_instrument_def_v3(decoder::DBNDecoder, hd::RecordHeader)\n") - for i in range(v3_start + 7, v3_end + 1): # Skip "else" and comments - line = old_function[i] - if line.startswith(" "): - line = " " + line[12:] - new_code.append(line) - new_code.append("end\n\n") - - print() - print(f"Generated {len(new_code)} lines of refactored code") - - # Assemble the final file - final_lines = ( - lines[:read_record_start] + # Before - new_code + # New refactored code - lines[read_record_end+1:] # After - ) - - print(f"Final file: {len(final_lines)} lines (was {len(lines)})") - print(f"Change: {len(final_lines) - len(lines):+d} lines") - print() - - # Write the refactored file - with open("src/decode.jl", "w", encoding="utf-8", newline='\n') as f: - f.writelines(final_lines) - - print("✓ Refactoring complete!") - print(" - Replaced 616-line mega-function") - print(" - Created type-stable helper functions") - print(" - All helpers marked @inline") - -if __name__ == "__main__": - main() diff --git a/refactor_complete.jl b/refactor_complete.jl deleted file mode 100644 index 2ed68d0..0000000 --- a/refactor_complete.jl +++ /dev/null @@ -1,297 +0,0 @@ -#!/usr/bin/env julia -# -# Complete function barrier refactoring for read_record() -# This splits the 616-line type-unstable mega-function into type-stable helpers -# - -println("Starting complete function barrier refactoring...") -println("=" ^ 70) - -# Read the original file -original_lines = readlines("src/decode.jl") -println("Original file: $(length(original_lines)) lines") - -# Locate the read_record function (lines 382-998) -read_record_start = findfirst(l -> occursin("function read_record(decoder::DBNDecoder)", l), original_lines) -read_record_end = findnext(l -> strip(l) == "end" && occursin("# Convenience", original_lines[min(length(original_lines), findnext(x->x==l, original_lines, 1)+1)]), original_lines, read_record_start) - -if read_record_start === nothing || read_record_end === nothing - error("Could not locate read_record function") -end - -println("Found read_record: lines $read_record_start to $read_record_end") -println("Function size: $(read_record_end - read_record_start + 1) lines") -println() - -# Build the refactored code -new_code = String[] - -# Part 1: New compact read_record function -push!(new_code, "function read_record(decoder::DBNDecoder)") -push!(new_code, " if eof(decoder.io)") -push!(new_code, " return nothing") -push!(new_code, " end") -push!(new_code, " ") -push!(new_code, " hd_result = read_record_header(decoder.io)") -push!(new_code, " ") -push!(new_code, " # Handle unknown record types") -push!(new_code, " if hd_result isa Tuple") -push!(new_code, " _, rtype_raw, record_length = hd_result") -push!(new_code, " skip(decoder.io, record_length - 2)") -push!(new_code, " return nothing") -push!(new_code, " end") -push!(new_code, " ") -push!(new_code, " hd = hd_result") -push!(new_code, " ") -push!(new_code, " # Type-stable dispatch using function barriers - eliminates 1.1M allocations") -push!(new_code, " return read_record_dispatch(decoder, hd, hd.rtype)") -push!(new_code, "end") -push!(new_code, "") - -# Part 2: Dispatch function -push!(new_code, "# Small dispatch function - type inference barrier") -push!(new_code, "@inline function read_record_dispatch(decoder::DBNDecoder, hd::RecordHeader, rtype::RType.T)") -push!(new_code, " if rtype == RType.MBO_MSG") -push!(new_code, " return read_mbo_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_0_MSG") -push!(new_code, " return read_trade_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_1_MSG") -push!(new_code, " return read_mbp1_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_10_MSG") -push!(new_code, " return read_mbp10_msg(decoder, hd)") -push!(new_code, " elseif rtype in (RType.OHLCV_1S_MSG, RType.OHLCV_1M_MSG, RType.OHLCV_1H_MSG, RType.OHLCV_1D_MSG)") -push!(new_code, " return read_ohlcv_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.STATUS_MSG") -push!(new_code, " return read_status_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.INSTRUMENT_DEF_MSG") -push!(new_code, " return read_instrument_def_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.IMBALANCE_MSG") -push!(new_code, " return read_imbalance_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.STAT_MSG") -push!(new_code, " return read_stat_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.ERROR_MSG") -push!(new_code, " return read_error_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.SYMBOL_MAPPING_MSG") -push!(new_code, " return read_symbol_mapping_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.SYSTEM_MSG") -push!(new_code, " return read_system_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CMBP_1_MSG") -push!(new_code, " return read_cmbp1_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CBBO_1S_MSG") -push!(new_code, " return read_cbbo1s_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CBBO_1M_MSG") -push!(new_code, " return read_cbbo1m_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.TCBBO_MSG") -push!(new_code, " return read_tcbbo_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.BBO_1S_MSG") -push!(new_code, " return read_bbo1s_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.BBO_1M_MSG") -push!(new_code, " return read_bbo1m_msg(decoder, hd)") -push!(new_code, " else") -push!(new_code, " skip(decoder.io, hd.length - 16)") -push!(new_code, " return nothing") -push!(new_code, " end") -push!(new_code, "end") -push!(new_code, "") - -# Part 3: Extract and convert each record type handler to a function -# This is the critical part - extracting the body of each if/elseif block - -println("Extracting type-stable helper functions...") - -# Helper function to extract code between markers and convert to function -function extract_handler(lines, start_pattern, end_pattern, func_name) - start_idx = findfirst(l -> occursin(start_pattern, l), lines) - if start_idx === nothing - return nothing - end - - # Find the return statement - end_idx = findnext(l -> occursin(end_pattern, l), lines, start_idx) - if end_idx === nothing - return nothing - end - - # Extract the body (skip the if/elseif line) - body_lines = String[] - - # Add function signature - push!(body_lines, "@inline function $func_name(decoder::DBNDecoder, hd::RecordHeader)") - - # Add body (with adjusted indentation) - for i in (start_idx+1):(end_idx) - line = lines[i] - # Skip blank comment lines at start - if isempty(strip(line)) && isempty(body_lines[2:end]) - continue - end - # Remove one level of indentation (8 spaces or 2 tabs) - adjusted = replace(line, r"^ " => " ", count=1) - push!(body_lines, adjusted) - end - - push!(body_lines, "end") - - return body_lines -end - -# Extract all handlers from the original mega-function -old_function = original_lines[read_record_start:read_record_end] - -# 1. MBO -println(" - read_mbo_msg") -mbo_code = extract_handler(old_function, "if hd.rtype == RType.MBO_MSG", "return MBOMsg", "read_mbo_msg") -if mbo_code !== nothing - append!(new_code, mbo_code) - push!(new_code, "") -end - -# 2. Trade (MBP_0) -println(" - read_trade_msg") -trade_code = extract_handler(old_function, "elseif hd.rtype == RType.MBP_0_MSG", "return TradeMsg", "read_trade_msg") -if trade_code !== nothing - append!(new_code, trade_code) - push!(new_code, "") -end - -# 3. MBP1 -println(" - read_mbp1_msg") -mbp1_code = extract_handler(old_function, "elseif hd.rtype == RType.MBP_1_MSG", "return MBP1Msg", "read_mbp1_msg") -if mbp1_code !== nothing - append!(new_code, mbp1_code) - push!(new_code, "") -end - -# 4. MBP10 -println(" - read_mbp10_msg") -mbp10_code = extract_handler(old_function, "elseif hd.rtype == RType.MBP_10_MSG", "return MBP10Msg", "read_mbp10_msg") -if mbp10_code !== nothing - append!(new_code, mbp10_code) - push!(new_code, "") -end - -# 5. OHLCV -println(" - read_ohlcv_msg") -ohlcv_code = extract_handler(old_function, "elseif hd.rtype in [RType.OHLCV", "return OHLCVMsg", "read_ohlcv_msg") -if ohlcv_code !== nothing - append!(new_code, ohlcv_code) - push!(new_code, "") -end - -# 6. Status -println(" - read_status_msg") -status_code = extract_handler(old_function, "elseif hd.rtype == RType.STATUS_MSG", "return StatusMsg", "read_status_msg") -if status_code !== nothing - append!(new_code, status_code) - push!(new_code, "") -end - -# 7. InstrumentDef (complex - needs special handling) -println(" - read_instrument_def_msg (+ v2/v3 helpers)") -# Find the InstrumentDef block -inst_start = findfirst(l -> occursin("elseif hd.rtype == RType.INSTRUMENT_DEF_MSG", l), old_function) -inst_end = findnext(l -> occursin("return InstrumentDefMsg", l), old_function, inst_start) -if inst_end !== nothing - # Find the closing paren of the return statement - inst_end = findnext(l -> strip(l) == ")", old_function, inst_end) -end - -if inst_start !== nothing && inst_end !== nothing - # Main dispatcher - push!(new_code, "@inline function read_instrument_def_msg(decoder::DBNDecoder, hd::RecordHeader)") - push!(new_code, " start_pos = position(decoder.io)") - push!(new_code, " record_size_bytes = hd.length * LENGTH_MULTIPLIER") - push!(new_code, " body_size = record_size_bytes - 16") - push!(new_code, " if body_size == 384") - push!(new_code, " return read_instrument_def_v2(decoder, hd)") - push!(new_code, " else") - push!(new_code, " return read_instrument_def_v3(decoder, hd)") - push!(new_code, " end") - push!(new_code, "end") - push!(new_code, "") - - # Find V2 block - v2_start = findnext(l -> occursin("if body_size == 384", l), old_function, inst_start) - v2_comment_end = findnext(l -> occursin("# Read ALL fields", l), old_function, v2_start) - v2_end = findnext(l -> occursin("else", l) && findnext(x -> occursin("DBN V3", x), old_function, findnext(y->y==l, old_function, 1)) !== nothing, old_function, v2_start) - - if v2_start !== nothing && v2_end !== nothing - push!(new_code, "@inline function read_instrument_def_v2(decoder::DBNDecoder, hd::RecordHeader)") - for i in (v2_comment_end):(v2_end-2) - line = old_function[i] - adjusted = replace(line, r"^ " => " ", count=1) - push!(new_code, adjusted) - end - push!(new_code, "end") - push!(new_code, "") - end - - # Find V3 block - v3_start = v2_end - v3_comment = findnext(l -> occursin("encode_order 0: ts_recv", l), old_function, v3_start) - v3_end = findnext(l -> occursin("return InstrumentDefMsg", l), old_function, v3_start) - v3_end = findnext(l -> strip(l) == ")", old_function, v3_end) - - if v3_start !== nothing && v3_end !== nothing - push!(new_code, "@inline function read_instrument_def_v3(decoder::DBNDecoder, hd::RecordHeader)") - for i in (v3_comment):(v3_end) - line = old_function[i] - adjusted = replace(line, r"^ " => " ", count=1) - push!(new_code, adjusted) - end - push!(new_code, "end") - push!(new_code, "") - end -end - -# 8-18. Remaining message types -handlers = [ - ("IMBALANCE_MSG", "ImbalanceMsg", "read_imbalance_msg"), - ("STAT_MSG", "StatMsg", "read_stat_msg"), - ("ERROR_MSG", "ErrorMsg", "read_error_msg"), - ("SYMBOL_MAPPING_MSG", "SymbolMappingMsg", "read_symbol_mapping_msg"), - ("SYSTEM_MSG", "SystemMsg", "read_system_msg"), - ("CMBP_1_MSG", "CMBP1Msg", "read_cmbp1_msg"), - ("CBBO_1S_MSG", "CBBO1sMsg", "read_cbbo1s_msg"), - ("CBBO_1M_MSG", "CBBO1mMsg", "read_cbbo1m_msg"), - ("TCBBO_MSG", "TCBBOMsg", "read_tcbbo_msg"), - ("BBO_1S_MSG", "BBO1sMsg", "read_bbo1s_msg"), - ("BBO_1M_MSG", "BBO1mMsg", "read_bbo1m_msg"), -] - -for (rtype, msgtype, funcname) in handlers - println(" - $funcname") - code = extract_handler(old_function, "elseif hd.rtype == RType.$rtype", "return $msgtype", funcname) - if code !== nothing - append!(new_code, code) - push!(new_code, "") - end -end - -println() -println("Generated $(length(new_code)) lines of refactored code") - -# Combine: before + new_code + after -new_file = vcat( - original_lines[1:(read_record_start-1)], - new_code, - original_lines[(read_record_end+1):end] -) - -println("New file: $(length(new_file)) lines (was $(length(original_lines)))") -println("Change: $(length(new_file) - length(original_lines)) lines") -println() - -# Write the refactored file -open("src/decode.jl", "w") do f - for line in new_file - println(f, line) - end -end - -println("✓ Refactoring complete!") -println(" Summary:") -println(" - Replaced 616-line mega-function with compact dispatch") -println(" - Created $(length(handlers) + 7) type-stable helper functions") -println(" - All helpers marked @inline for zero overhead") -println(" - Expected: Eliminate ~1M allocations, 3x performance improvement") diff --git a/refactor_read_record.jl b/refactor_read_record.jl deleted file mode 100644 index 22778b0..0000000 --- a/refactor_read_record.jl +++ /dev/null @@ -1,111 +0,0 @@ -#!/usr/bin/env julia -# -# Refactor read_record() to use function barriers for type stability -# This eliminates ~1.1M allocations by splitting the mega-function into type-stable helpers -# - -println("Refactoring read_record() in src/decode.jl...") - -# Read current file -lines = readlines("src/decode.jl") -println("Original file: $(length(lines)) lines") - -# The old read_record function spans lines 382-998 (617 lines) -# We'll replace it with the new refactored version - -# Part 1: Keep everything before read_record (lines 1-381) -before = lines[1:381] - -# Part 2: Keep everything after read_record (lines 999-end) -after = lines[999:end] - -# Part 3: New refactored read_record code -new_code = String[] - -push!(new_code, """function read_record(decoder::DBNDecoder) - if eof(decoder.io) - return nothing - end - - hd_result = read_record_header(decoder.io) - - # Handle unknown record types - if hd_result isa Tuple - _, rtype_raw, record_length = hd_result - skip(decoder.io, record_length - 2) - return nothing - end - - hd = hd_result - - # Dispatch to type-stable reader functions (function barrier pattern) - return read_record_dispatch(decoder, hd, hd.rtype) -end""") - -push!(new_code, "") -push!(new_code, "# Type-stable dispatch function - eliminates type instability from mega-function") -push!(new_code, "@inline function read_record_dispatch(decoder::DBNDecoder, hd::RecordHeader, rtype::RType.T)") -push!(new_code, " if rtype == RType.MBO_MSG") -push!(new_code, " return read_mbo_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_0_MSG") -push!(new_code, " return read_trade_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_1_MSG") -push!(new_code, " return read_mbp1_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.MBP_10_MSG") -push!(new_code, " return read_mbp10_msg(decoder, hd)") -push!(new_code, " elseif rtype in (RType.OHLCV_1S_MSG, RType.OHLCV_1M_MSG, RType.OHLCV_1H_MSG, RType.OHLCV_1D_MSG)") -push!(new_code, " return read_ohlcv_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.STATUS_MSG") -push!(new_code, " return read_status_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.INSTRUMENT_DEF_MSG") -push!(new_code, " return read_instrument_def_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.IMBALANCE_MSG") -push!(new_code, " return read_imbalance_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.STAT_MSG") -push!(new_code, " return read_stat_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.ERROR_MSG") -push!(new_code, " return read_error_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.SYMBOL_MAPPING_MSG") -push!(new_code, " return read_symbol_mapping_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.SYSTEM_MSG") -push!(new_code, " return read_system_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CMBP_1_MSG") -push!(new_code, " return read_cmbp1_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CBBO_1S_MSG") -push!(new_code, " return read_cbbo1s_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.CBBO_1M_MSG") -push!(new_code, " return read_cbbo1m_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.TCBBO_MSG") -push!(new_code, " return read_tcbbo_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.BBO_1S_MSG") -push!(new_code, " return read_bbo1s_msg(decoder, hd)") -push!(new_code, " elseif rtype == RType.BBO_1M_MSG") -push!(new_code, " return read_bbo1m_msg(decoder, hd)") -push!(new_code, " else") -push!(new_code, " skip(decoder.io, hd.length - 16)") -push!(new_code, " return nothing") -push!(new_code, " end") -push!(new_code, "end") -push!(new_code, "") - -# Now add all the type-stable helper functions -# Include the helpers from the separate file we created -helper_code = read("src/decode_refactored.jl", String) -append!(new_code, split(helper_code, '\n')) - -# Combine all parts -new_lines = vcat(before, new_code, after) - -println("New file: $(length(new_lines)) lines") -println("Difference: $(length(new_lines) - length(lines)) lines") - -# Write the refactored file -open("src/decode.jl", "w") do f - for line in new_lines - println(f, line) - end -end - -println("✓ Refactoring complete!") -println(" - Old read_record: 617 lines (type-unstable mega-function)") -println(" - New read_record + helpers: ~$(length(new_code)) lines (type-stable dispatch)")