INGESTION_OPTIMIZATION_IDEAS

ThemisDB Ingestion: Optimierungspotenziale und Ideen

Erstellt: 25. Dezember 2025
Version: 1.0
Status: Analyse & Empfehlungen
Kategorie: Performance Optimization

---

📋 Executive Summary

Diese Analyse untersucht die Datenübertragung zur ThemisDB-Datenbank und identifiziert konkrete Optimierungsmöglichkeiten für die Ingestion-Pipeline. Die Untersuchung konzentriert sich auf:

1. RocksDB Write Path Optimierungen
2. HTTP/gRPC Protokoll-Overhead Reduktion
3. Batch- und Buffer-Strategien
4. Kompression und Serialisierung
5. Memory-Mapped I/O und Zero-Copy
6. Client-Side Optimierungen

Kernerkenntnisse

✅ Bestehende Stärken:

• Auto-Buffer für Time Series (TSAutoBuffer) mit 10-50× Throughput-Verbesserung
• VectorAutoBuffer für RAG/Semantic Search implementiert
• Compressed WAL Shipping mit 3-10× Bandbreitenreduktion
• RocksDB WriteBatch für atomare Bulk-Operationen
• Binary Protocol Support für niedrige Latenz

💡 Identifizierte Optimierungspotenziale:

• RocksDB Write Buffer Tuning: +20-40% Schreib-Performance
• HTTP Overhead Reduktion: +30-50% bei binärem Protokoll
• Adaptive Batch Sizing: +15-25% bei variablen Datenmengen
• Payload Compression: 60-90% Bandbreitenreduktion
• Memory-Mapped Bulk Import: +100-300% für große Dateien

---

🎯 Teil 1: RocksDB Write Path Optimierungen

1.1 Write Buffer (Memtable) Tuning

Aktuelle Konfiguration:

// src/storage/rocksdb_wrapper.cpp, Zeile 87-89
options_->write_buffer_size = config_.memtable_size_mb * 1024 * 1024;  // Default: 256 MB
options_->max_write_buffer_number = config_.max_write_buffer_number;   // Default: 3
options_->min_write_buffer_number_to_merge = config_.min_write_buffer_number_to_merge;  // Default: 1

Problem:

• Zu kleine Memtables verursachen häufige Flushes → höhere Latenz
• Zu große Memtables verzögern Flushes → Memory-Druck
• Nicht optimiert für verschiedene Workload-Profile

Empfohlene Optimierungen:

1.1.1 Dynamische Write Buffer Sizing

// Adaptive write buffer configuration basierend auf System-RAM
struct AdaptiveWriteBufferConfig {
    size_t system_ram_gb;
    size_t write_buffer_size_mb;
    int max_write_buffer_number;
    size_t db_write_buffer_size_mb;
};

// Empfohlene Konfigurationen
std::vector<AdaptiveWriteBufferConfig> configs = {
    // RAM,  WB_Size, Max_Num, Total_Size
    {8,      128,     3,        384},      // Low-end systems
    {16,     256,     4,        1024},     // Standard (current)
    {32,     512,     6,        3072},     // Write-heavy workloads
    {64,     1024,    6,        6144},     // High-throughput ingestion
    {128,    2048,    8,        16384}     // Enterprise bulk loading
};

Erwarteter Impact:

• Low-end (8GB RAM): Stabil, keine Out-of-Memory
• Standard (16GB): Baseline (aktuell)
• Write-heavy (32GB): +40-60% Write Throughput
• High-throughput (64GB): +100-150% Write Throughput
• Enterprise (128GB): +200-300% Write Throughput

Risiko: Niedrig (gut getestet in RocksDB)

---

1.1.2 Parallel Memtable Writes

Aktuelle Konfiguration:

// src/storage/rocksdb_wrapper.cpp, Zeile 174-181
options_->allow_concurrent_memtable_write = config_.allow_concurrent_memtable_write;  // Default: true
options_->enable_pipelined_write = config_.enable_pipelined_write;  // Default: true

Problem:

• Concurrent memtable writes nur bei WRITE_PREPARED policy
• Pipelined writes haben immer noch Lock Contention bei >16 Threads

Empfohlene Optimierung:

// Enhanced parallel write configuration
struct ParallelWriteConfig {
    Config::WritePolicy write_policy = Config::WritePolicy::WritePrepared;
    bool allow_concurrent_memtable_write = true;
    bool enable_pipelined_write = true;
    bool two_write_queues = true;           // Dual queues (prepare/commit)
    uint64_t wp_commit_cache_bits = 23;     // 8M commit cache entries
    
    // NEW: Per-thread memtable assignment
    bool enable_per_thread_memtable = true;  // Reduced lock contention
    int num_memtable_shards = 16;            // Shard count (power of 2)
};

Erwarteter Impact:

• 8 Threads: +10-15% (wenig Contention)
• 16 Threads: +25-40% (moderate Contention)
• 32 Threads: +60-100% (hohe Contention)
• 64+ Threads: +150-250% (extreme Contention)

Implementierungsaufwand: 3-5 Tage
Risiko: Mittel (erfordert RocksDB-Patch oder neueste Version)

---

1.2 Level0 Compaction Tuning

Aktuelle Konfiguration:

// src/storage/rocksdb_wrapper.cpp, Zeile 140-142
options_->level0_file_num_compaction_trigger = config_.level0_file_num_compaction_trigger;  // Default: 4
options_->level0_slowdown_writes_trigger = config_.level0_slowdown_writes_trigger;  // Default: 20
options_->level0_stop_writes_trigger = config_.level0_stop_writes_trigger;  // Default: 36

Problem:

• Bei hohem Write-Throughput können L0-Files schneller entstehen als Compaction mithalten kann
• Write-Stalls treten auf → Latenz-Spitzen

Empfohlene Optimierung:

// Aggressive Level0 Compaction for High-Throughput Ingestion
struct AggressiveCompactionConfig {
    int level0_file_num_compaction_trigger = 2;   // Start früher (default: 4)
    int level0_slowdown_writes_trigger = 8;       // Frühere Warnung (default: 20)
    int level0_stop_writes_trigger = 16;          // Frühere Blockierung (default: 36)
    
    int max_background_compactions = 8;           // Mehr Compaction-Threads
    int max_subcompactions = 2;                   // Parallel sub-compactions
    
    // NEW: Prioritize L0 compactions
    bool prioritize_level0_compaction = true;
    double level0_compaction_priority_multiplier = 2.0;
};

Trade-off:

• ✅ Vermeidet Write-Stalls → niedrigere P99-Latenz
• ⚠️ Höherer CPU-Verbrauch für Compaction
• ⚠️ Höherer I/O durch häufigere Compactions

Erwarteter Impact:

• P99 Write Latency: -50-70% (eliminiert Stalls)
• CPU-Auslastung: +15-25%
• Sustained Write Throughput: +20-40%

Use Case: Bulk-Import, Streaming-Ingestion mit hohen Raten

---

1.3 Write-Ahead Log (WAL) Optimization

Aktuelle Konfiguration:

// src/storage/rocksdb_wrapper.cpp, Zeile 184-188
write_options_->sync = config_.enable_wal;  // Default: true
write_options_->disableWAL = config_.disable_wal_for_benchmark;  // Default: false

Problem:

• WAL fsync bei jedem Schreibvorgang → hohe Latenz
• Keine Group Commit für kleine Writes

Empfohlene Optimierungen:

1.3.1 Asynchronous WAL mit Group Commit

// Async WAL Configuration
struct AsyncWALConfig {
    bool sync = false;                          // Async writes (keine fsync)
    bool disableWAL = false;                    // WAL bleibt aktiv
    
    // NEW: Group Commit Settings
    size_t wal_bytes_per_sync = 1024 * 1024;   // 1 MB (batch fsync)
    std::chrono::microseconds max_write_delay{100};  // Max 100µs delay für group commit
    size_t max_writes_in_group = 100;          // Max 100 writes pro group
};

Sicherheits-Trade-off:

• ❌ Datenverlust-Risiko: Bis zu 100µs Daten bei Crash
• ✅ Performance: +200-500% Write Throughput
• ✅ Anwendbar: Bei Read-Replicas, Entwicklungsumgebungen

Anwendungsbeispiel:

// Konfiguration für verschiedene Szenarien
AsyncWALConfig production = {
    .sync = true,                    // Volle Durability
    .wal_bytes_per_sync = 0          // Jeder Write synced
};

AsyncWALConfig read_replica = {
    .sync = false,                   // Async WAL
    .wal_bytes_per_sync = 1MB,       // Batch fsync
    .max_write_delay = 100µs         // Niedrige Latenz
};

AsyncWALConfig bulk_import = {
    .sync = false,                   // Async WAL
    .wal_bytes_per_sync = 64MB,      // Große Batches
    .max_write_delay = 1000µs        // Höhere Latenz akzeptabel
};

---

1.3.2 Separate WAL Directory auf schnellem Storage

Empfehlung:

# config/rocksdb.yaml
storage:
  db_path: /data/rocksdb        # HDD/SATA SSD (bulk storage)
  wal_dir: /nvme/rocksdb/wal    # NVMe SSD (low latency)
  
  # Multi-path SSTable placement
  db_paths:
    - path: /nvme/rocksdb/hot
      target_size: 100GB          # Hot data auf NVMe
    - path: /ssd/rocksdb/warm
      target_size: 500GB          # Warm data auf SATA SSD
    - path: /data/rocksdb/cold
      target_size: unlimited      # Cold data auf HDD

Erwarteter Impact:

• Write Latency: -30-50% (NVMe für WAL)
• Read Performance: +20-40% (Tiered Storage)
• Kosten: Optimal (nur Hot Data auf teurem Storage)

---

🌐 Teil 2: HTTP/gRPC Protokoll-Overhead Reduktion

2.1 Binary Protocol vs. HTTP/JSON

Aktuelle Situation:

ThemisDB unterstützt mehrere Protokolle:

1. HTTP/REST mit JSON (Standard)
2. Binary Protocol mit MessagePack (buffer_binary_protocol.h)
3. gRPC (optional, via Plugin)

Overhead-Analyse:

Protokoll	Payload Size	Parse Time	Latency	Throughput
HTTP/JSON	100% (Baseline)	100%	100%	100%
HTTP/MessagePack	60%	30%	70%	140%
Binary/MessagePack	60%	20%	50%	200%
gRPC/Protobuf	40%	15%	40%	250%

Beispiel: Time Series Insert

// HTTP/JSON (229 Bytes)
POST /ts/put
{
  "metric": "cpu.usage",
  "entity": "server-01",
  "timestamp": 1703520000,
  "value": 75.5,
  "tags": {
    "datacenter": "us-east-1",
    "app": "frontend"
  }
}

# Binary/MessagePack (87 Bytes, -62%)
# Opcode: 0x70 (TS_PUT_BUFFERED)
# Payload: {
#   'm': 'cpu.usage',
#   'e': 'server-01',
#   't': 1703520000,
#   'v': 75.5,
#   'tags': {'dc': 'us-east-1', 'app': 'fe'}
# }

Empfehlung:

Für Bulk-Ingestion und High-Throughput-Szenarien sollte das Binary Protocol bevorzugt werden:

# Python Client - Binary Protocol
from themisdb import BufferedClient

client = BufferedClient('localhost', 9090, protocol='binary')
client.put_ts_buffered('cpu.usage', 'server-01', timestamp, 75.5)

---

2.2 HTTP/2 und HTTP/3 Multiplexing

Aktuelle Unterstützung:

// include/server/http2_session.h
// include/server/http3_session.h
#ifdef THEMIS_ENABLE_HTTP2
#ifdef THEMIS_ENABLE_HTTP3

Problem:

• HTTP/1.1: Head-of-Line Blocking
• Eine langsame Request blockiert alle folgenden Requests
• Kein Request-Pipelining

Empfehlung: HTTP/2 aktivieren

# config/server.yaml
server:
  enable_http2: true
  http2_max_concurrent_streams: 1000   # Parallel requests
  http2_initial_window_size: 65535     # Flow control

Erwarteter Impact:

• Parallele Requests: 10-100× mehr concurrent requests
• Latenz: -30-50% (kein Head-of-Line Blocking)
• Throughput: +50-100% bei vielen kleinen Requests

---

2.3 Payload Compression

Aktuelle Situation: Keine automatische HTTP-Payload-Kompression implementiert.

Empfohlene Implementierung:

// Content-Encoding Support
enum class CompressionType {
    None,
    Gzip,      // Weit verbreitet, moderate Kompression
    Zstd,      // Bessere Ratio, schneller
    LZ4        // Sehr schnell, moderate Ratio
};

// HTTP Request Handler
http::response<http::string_body> handleRequest(
    const http::request<http::string_body>& req) {
    
    auto response = processRequest(req);
    
    // Check Accept-Encoding header
    if (req["Accept-Encoding"].find("zstd") != std::string::npos) {
        response.set(http::field::content_encoding, "zstd");
        response.body() = compress_zstd(response.body());
    } else if (req["Accept-Encoding"].find("gzip") != std::string::npos) {
        response.set(http::field::content_encoding, "gzip");
        response.body() = compress_gzip(response.body());
    }
    
    return response;
}

Kompressionsraten:

Datentyp	Gzip	Zstd	LZ4
JSON Metadata	70-80%	75-85%	60-70%
Time Series (Doubles)	40-60%	50-70%	30-50%
Text/Logs	80-90%	85-92%	70-80%
Embeddings (Float32)	10-20%	15-25%	5-15%

Empfehlung:

• Standard: Zstd Level 3 (bestes Trade-off)
• Low-Latency: LZ4 (minimal CPU)
• Maximum Compression: Zstd Level 19 (für Archivierung)

---

🔄 Teil 3: Batch- und Buffer-Strategien

3.1 Adaptive Batch Sizing

Problem: Feste Batch-Größen sind nicht optimal für variable Datenmengen:

• Kleine Batches: Overhead durch häufige Commits
• Große Batches: Höhere Latenz, mehr Memory

Empfohlene Implementierung:

// Adaptive Batch Controller
class AdaptiveBatchController {
public:
    struct Config {
        size_t min_batch_size = 10;
        size_t max_batch_size = 10000;
        size_t target_batch_latency_ms = 50;    // Target: 50ms pro Batch
        size_t memory_limit_mb = 100;
        
        // Learning parameters
        double alpha = 0.1;                      // Learning rate
        std::chrono::milliseconds sample_window{10000};  // 10s window
    };
    
    // Dynamisch angepasste Batch-Größe
    size_t getCurrentBatchSize() const {
        return current_batch_size_;
    }
    
    // Feedback nach Batch-Commit
    void recordBatchMetrics(size_t batch_size, 
                           std::chrono::milliseconds latency,
                           size_t memory_used) {
        // Update moving averages
        avg_latency_ = (1 - alpha_) * avg_latency_ + alpha_ * latency.count();
        
        // Adjust batch size
        if (avg_latency_ > config_.target_batch_latency_ms) {
            // Latenz zu hoch -> kleinere Batches
            current_batch_size_ = std::max(
                config_.min_batch_size,
                current_batch_size_ * 0.9
            );
        } else if (avg_latency_ < config_.target_batch_latency_ms * 0.5) {
            // Latenz niedrig -> größere Batches
            current_batch_size_ = std::min(
                config_.max_batch_size,
                current_batch_size_ * 1.1
            );
        }
    }
    
private:
    Config config_;
    size_t current_batch_size_;
    double avg_latency_;
    double alpha_;
};

Erwarteter Impact:

• Variable Workloads: +15-25% Durchsatz
• Latenz-Stabilität: -20-30% P99 Latency
• Memory-Effizienz: +10-20%

---

3.2 Multi-Level Buffering

Konzept:

Client Buffer (100ms)
    ↓
Server Buffer (1s)
    ↓
RocksDB WriteBatch (sync)

Implementierung:

// Three-tier buffering architecture
class ThreeTierBuffer {
public:
    // Tier 1: Client-side buffer (lowest latency)
    class ClientBuffer {
        std::deque<Record> buffer_;
        std::chrono::milliseconds flush_interval_{100};  // 100ms
        
        void add(const Record& record) {
            buffer_.push_back(record);
            if (shouldFlush()) flush();
        }
    };
    
    // Tier 2: Server-side buffer (medium latency)
    class ServerBuffer {
        std::deque<Record> buffer_;
        std::chrono::milliseconds flush_interval_{1000};  // 1s
        size_t max_size_ = 10000;
        
        void add(const Record& record) {
            buffer_.push_back(record);
            if (shouldFlush()) flush();
        }
    };
    
    // Tier 3: RocksDB WriteBatch (atomicity)
    void flushToRocksDB(const std::vector<Record>& records) {
        auto batch = storage_->createWriteBatch();
        for (const auto& r : records) {
            batch->put(r.key, r.value);
        }
        batch->commit();
    }
};

Vorteile:

• Client: Niedrige Latenz (in-memory)
• Server: Aggregation & Kompression
• RocksDB: Atomare Commits mit WriteBatch

---

3.3 Prioritätsbasierte Buffer-Queues

Problem: Alle Daten werden gleich behandelt, unabhängig von Wichtigkeit oder Latenz-Anforderungen.

Empfohlene Implementierung:

// Priority-based buffer queues
enum class Priority {
    Critical = 0,    // Real-time alerts, high-value transactions
    High = 1,        // User-facing operations
    Normal = 2,      // Standard ingestion
    Low = 3,         // Batch imports, backfills
    Background = 4   // Analytics, archival
};

class PriorityBufferQueue {
public:
    void add(const Record& record, Priority priority) {
        queues_[static_cast<int>(priority)].push_back(record);
    }
    
    std::vector<Record> getNextBatch(size_t max_size) {
        // Priority-based selection: Critical > High > Normal > Low > Background
        for (auto& queue : queues_) {
            if (!queue.empty()) {
                size_t batch_size = std::min(max_size, queue.size());
                std::vector<Record> batch(queue.begin(), queue.begin() + batch_size);
                queue.erase(queue.begin(), queue.begin() + batch_size);
                return batch;
            }
        }
        return {};
    }
    
private:
    std::array<std::deque<Record>, 5> queues_;  // 5 priority levels
};

Use Cases:

• Critical: Financial transactions, alerts
• High: User API requests
• Normal: Standard ingestion
• Low: Batch imports
• Background: Analytics processing

---

🗜️ Teil 4: Kompression und Serialisierung

4.1 Embedding-Kompression (Product Quantization)

Aktuelle Situation: Embeddings werden als Float32 gespeichert (4 Bytes pro Dimension).

Problem:

• 768D Embedding = 3072 Bytes
• 1 Million Embeddings = 3 GB
• Keine Kompression → hoher Memory/Disk-Verbrauch

Empfohlene Optimierung: Product Quantization

// Product Quantization für Embeddings
class ProductQuantizer {
public:
    struct Config {
        int dimension = 768;
        int num_subvectors = 96;         // 768/8 = 96 subvectors
        int codebook_bits = 8;           // 256 codes per subvector
        bool use_opq = true;             // Optimized Product Quantization
    };
    
    // Trainiere Codebooks
    void train(const std::vector<std::vector<float>>& embeddings) {
        // 1. Optional: Rotate (OPQ)
        if (config_.use_opq) {
            rotation_matrix_ = computeOptimalRotation(embeddings);
            embeddings = rotate(embeddings, rotation_matrix_);
        }
        
        // 2. Split in subvectors
        auto subvectors = splitIntoSubvectors(embeddings);
        
        // 3. Train k-means für jeden Subvektor (256 clusters)
        for (int i = 0; i < config_.num_subvectors; i++) {
            codebooks_[i] = trainKMeans(subvectors[i], 256);
        }
    }
    
    // Komprimiere Embedding
    std::vector<uint8_t> encode(const std::vector<float>& embedding) {
        std::vector<uint8_t> codes(config_.num_subvectors);
        for (int i = 0; i < config_.num_subvectors; i++) {
            codes[i] = findNearestCode(embedding, i);
        }
        return codes;  // 768 floats → 96 bytes (32x Kompression!)
    }
    
    // Dekomprimiere Embedding (approximativ)
    std::vector<float> decode(const std::vector<uint8_t>& codes) {
        std::vector<float> embedding(config_.dimension);
        for (int i = 0; i < config_.num_subvectors; i++) {
            auto centroid = codebooks_[i][codes[i]];
            std::copy(centroid.begin(), centroid.end(), 
                     embedding.begin() + i * (config_.dimension / config_.num_subvectors));
        }
        return embedding;
    }
};

Kompressionsraten:

Methode	Größe	Recall@10	Suche
Float32 (Original)	100%	100%	100%
Int8 Quantization	25%	95-98%	90%
Int16 Quantization	50%	98-99%	95%
Product Quantization (PQ)	3-6%	90-95%	80%
OPQ (Optimized PQ)	3-6%	93-97%	85%

Erwarteter Impact:

• Speicher: -90-97% (32× Kompression mit PQ)
• Disk I/O: -90-97%
• Recall: -3-5% (akzeptabel für viele Use Cases)
• Suchgeschwindigkeit: +20-40% (weniger Daten zu laden)

Use Cases:

• RAG/Semantic Search mit Millionen Dokumenten
• Image/Video Embeddings
• Knowledge Graph Embeddings

---

4.2 Time Series Gorilla Compression

Bereits implementiert in TSAutoBuffer! ✅

// include/timeseries/ts_auto_buffer.h
// Gorilla compression: 10-20x compression für Double-Werte
class TSAutoBuffer {
    // XOR-based delta encoding
    // Timestamp delta-of-delta encoding
    // Leading/trailing zero suppression
};

Performance:

• Kompression: 10-20× für typische Time Series
• CPU-Overhead: Minimal (<5%)
• Bereits produktiv im Einsatz ✅

---

4.3 JSON Payload Pre-Compression

Problem: Große JSON-Payloads werden uncompressed übertragen.

Empfohlene Client-Side Implementierung:

# Python Client mit Payload-Kompression
import zstandard as zstd
import requests

class CompressedClient:
    def __init__(self, url):
        self.url = url
        self.compressor = zstd.ZstdCompressor(level=3)
    
    def post(self, endpoint, data):
        # Serialisiere zu JSON
        json_data = json.dumps(data).encode('utf-8')
        
        # Komprimiere mit Zstd
        compressed = self.compressor.compress(json_data)
        
        # Sende mit Content-Encoding header
        response = requests.post(
            f"{self.url}{endpoint}",
            data=compressed,
            headers={
                'Content-Type': 'application/json',
                'Content-Encoding': 'zstd'
            }
        )
        return response

Server-Side Handler:

// Automatische Dekompression im HTTP Handler
http::response<http::string_body> handleRequest(
    const http::request<http::string_body>& req) {
    
    std::string body = req.body();
    
    // Check Content-Encoding
    auto encoding = req["Content-Encoding"];
    if (encoding == "zstd") {
        body = decompress_zstd(body);
    } else if (encoding == "gzip") {
        body = decompress_gzip(body);
    }
    
    // Parse JSON
    auto payload = json::parse(body);
    // ... process request
}

Erwarteter Impact:

• JSON Metadata: 70-85% Reduktion
• Network Traffic: -70-85%
• Client CPU: +5-10%
• Server CPU: +5-10%

Trade-off: Akzeptabel für WAN, weniger sinnvoll für lokale Netzwerke.

---

💾 Teil 5: Memory-Mapped I/O und Zero-Copy

5.1 Memory-Mapped File Import

Problem: Große Dateien werden vollständig in Memory geladen vor dem Import.

Empfohlene Implementierung:

// Memory-mapped bulk import
class MemoryMappedImporter {
public:
    Status importFromFile(const std::string& filepath) {
        // 1. Memory-map file (kein Memory-Overhead)
        int fd = open(filepath.c_str(), O_RDONLY);
        struct stat st;
        fstat(fd, &st);
        
        void* mapped = mmap(nullptr, st.st_size, PROT_READ, MAP_PRIVATE, fd, 0);
        madvise(mapped, st.st_size, MADV_SEQUENTIAL);  // Hint: sequential read
        
        // 2. Parse in chunks (keine Kopie)
        size_t offset = 0;
        while (offset < st.st_size) {
            // Parse chunk direkt aus Memory-Map
            auto records = parseChunk(static_cast<char*>(mapped) + offset, chunk_size);
            
            // Batch-Import
            importBatch(records);
            
            offset += chunk_size;
        }
        
        // 3. Cleanup
        munmap(mapped, st.st_size);
        close(fd);
        
        return Status::OK();
    }
};

Vorteile:

• Kein Memory-Overhead: OS managed page cache
• Lazy Loading: Nur benötigte Pages werden geladen
• Zero-Copy: Direkte Verarbeitung aus Memory-Map

Erwarteter Impact:

• Memory-Verbrauch: -80-95% (keine Duplikation)
• Import-Geschwindigkeit: +100-300% (bei großen Dateien)
• Startup-Zeit: Instant (kein Pre-Loading)

---

5.2 Zero-Copy Network Transfers

Problem: Daten werden mehrfach kopiert: Network Buffer → User Space → RocksDB

Empfohlene Optimierung:

// Zero-copy network receive
class ZeroCopyReceiver {
public:
    Status receiveAndStore(int socket_fd) {
        // 1. Prepare RocksDB WriteBatch
        auto batch = storage_->createWriteBatch();
        
        // 2. Direct socket → WriteBatch (keine Zwischenkopie)
        char buffer[4096];
        while (true) {
            // Recv direkt in Batch-Buffer
            ssize_t bytes = recv(socket_fd, buffer, sizeof(buffer), 0);
            if (bytes <= 0) break;
            
            // Parse und direkt in Batch schreiben
            auto records = parseRecords(buffer, bytes);
            for (const auto& r : records) {
                batch->put(r.key, std::string_view(buffer + r.value_offset, r.value_size));
            }
        }
        
        // 3. Single commit
        return batch->commit();
    }
};

Erwarteter Impact:

• Memory Copies: 3 → 1 (67% Reduktion)
• CPU-Verbrauch: -20-30%
• Latenz: -15-25%

---

5.3 Direct I/O für Bulk Writes

Aktuelle Konfiguration:

// src/storage/rocksdb_wrapper.cpp
options_->use_direct_reads = false;
options_->use_direct_io_for_flush_and_compaction = false;

Empfohlene Aktivierung für Bulk Import:

// Direct I/O configuration for bulk ingestion
struct DirectIOConfig {
    bool use_direct_reads = false;                       // Standard: false
    bool use_direct_io_for_flush_and_compaction = true;  // Bulk: true
    
    // NEW: Conditional direct I/O
    bool enable_direct_io_during_bulk_import = true;
};

Vorteile:

• OS Cache Bypass: Kein Thrashing bei großen Imports
• Predictable Performance: Keine OS-Cache-Effekte
• Memory-Effizienz: Mehr Memory für RocksDB Block Cache

Trade-off:

• ✅ Bulk Import: +30-50% Performance
• ⚠️ Normal Operations: -10-20% Performance (kein OS Cache)

Empfehlung: Nur für Bulk-Import aktivieren, danach deaktivieren.

---

👨‍💻 Teil 6: Client-Side Optimierungen

6.1 Connection Pooling

Problem: Jeder Request öffnet eine neue TCP-Verbindung → hoher Overhead.

Empfohlene Implementierung:

# Python Client mit Connection Pool
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class PooledClient:
    def __init__(self, base_url, pool_size=100):
        self.base_url = base_url
        self.session = requests.Session()
        
        # Configure connection pool
        adapter = HTTPAdapter(
            pool_connections=pool_size,
            pool_maxsize=pool_size,
            max_retries=Retry(
                total=3,
                backoff_factor=0.1,
                status_forcelist=[500, 502, 503, 504]
            )
        )
        
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)
    
    def post(self, endpoint, data):
        return self.session.post(f"{self.base_url}{endpoint}", json=data)

Erwarteter Impact:

• Connection Overhead: -80-90%
• Latenz: -20-40% (keine TCP Handshake)
• Throughput: +50-100%

---

6.2 Request Pipelining und Multiplexing

HTTP/2 Client:

# HTTP/2 Client mit Multiplexing
import httpx

class HTTP2Client:
    def __init__(self, base_url):
        self.base_url = base_url
        self.client = httpx.AsyncClient(http2=True)
    
    async def post_many(self, requests):
        # Send alle Requests parallel über HTTP/2
        tasks = [
            self.client.post(f"{self.base_url}{req['endpoint']}", json=req['data'])
            for req in requests
        ]
        
        # Wait für alle Responses
        responses = await asyncio.gather(*tasks)
        return responses

Erwarteter Impact:

• Parallele Requests: 10-100× mehr
• Latenz: -40-60% (kein Head-of-Line Blocking)
• Throughput: +100-300%

---

6.3 Client-Side Batching

Problem: Viele kleine Requests statt wenige große Batches.

Empfohlene Implementierung:

# Auto-Batching Client
import time
from collections import deque
from threading import Thread, Lock

class AutoBatchingClient:
    def __init__(self, base_url, batch_size=1000, flush_interval=1.0):
        self.base_url = base_url
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        
        self.buffer = deque()
        self.lock = Lock()
        
        # Background flush thread
        self.flush_thread = Thread(target=self._flush_loop, daemon=True)
        self.flush_thread.start()
    
    def add(self, endpoint, data):
        with self.lock:
            self.buffer.append({'endpoint': endpoint, 'data': data})
            
            if len(self.buffer) >= self.batch_size:
                self._flush()
    
    def _flush(self):
        if not self.buffer:
            return
        
        with self.lock:
            batch = list(self.buffer)
            self.buffer.clear()
        
        # Send batch request
        response = requests.post(
            f"{self.base_url}/batch",
            json={'operations': batch}
        )
    
    def _flush_loop(self):
        while True:
            time.sleep(self.flush_interval)
            self._flush()

Erwarteter Impact:

• Request Count: -99% (1000:1 Batching)
• Overhead: -95-99%
• Throughput: +500-1000%

---

📊 Teil 7: Zusammenfassung und Priorisierung

7.1 Quick Wins (1-2 Wochen)

Optimierung	Impact	Aufwand	ROI
Adaptive Write Buffer Sizing	+40-60%	2 Tage	⭐⭐⭐⭐⭐
HTTP/2 Aktivierung	+50-100%	1 Tag	⭐⭐⭐⭐⭐
Client Connection Pooling	+50-100%	2 Tage	⭐⭐⭐⭐⭐
Payload Compression (Zstd)	-70-85% Traffic	3 Tage	⭐⭐⭐⭐
Level0 Compaction Tuning	-50-70% P99 Latency	2 Tage	⭐⭐⭐⭐

Gesamtimpact: +100-200% Throughput, -60-80% Latency

---

7.2 Medium-term (1-2 Monate)

Optimierung	Impact	Aufwand	ROI
Product Quantization für Embeddings	-90-97% Speicher	2 Wochen	⭐⭐⭐⭐
Memory-Mapped File Import	+100-300%	1 Woche	⭐⭐⭐⭐
Per-Thread Memtables	+150-250% (@64 Threads)	1 Monat	⭐⭐⭐
Direct I/O für Bulk Import	+30-50%	1 Woche	⭐⭐⭐
Zero-Copy Network Transfers	-20-30% CPU	2 Wochen	⭐⭐⭐

Gesamtimpact: +200-500% für spezifische Workloads

---

7.3 Long-term (3-6 Monate)

Optimierung	Impact	Aufwand	ROI
Async WAL mit Group Commit	+200-500%	3 Wochen	⭐⭐⭐⭐⭐
Adaptive Batch Sizing	+15-25%	1 Monat	⭐⭐⭐
Priority-based Queues	Bessere QoS	2 Wochen	⭐⭐⭐
Multi-Level Buffering	+20-40%	1 Monat	⭐⭐⭐

---

7.4 Konfigurationsempfehlungen

Standard-Konfiguration (16GB RAM, 4 Cores)

# config/rocksdb.yaml
storage:
  memtable_size_mb: 256
  max_write_buffer_number: 4
  db_write_buffer_size_mb: 1024
  max_background_jobs: 4
  
  level0_file_num_compaction_trigger: 4
  level0_slowdown_writes_trigger: 20
  level0_stop_writes_trigger: 36
  
  enable_http2: true
  compression: "zstd"

High-Throughput-Konfiguration (64GB RAM, 16+ Cores)

storage:
  memtable_size_mb: 1024
  max_write_buffer_number: 6
  db_write_buffer_size_mb: 6144
  max_background_jobs: 16
  max_background_compactions: 8
  max_subcompactions: 2
  
  level0_file_num_compaction_trigger: 2
  level0_slowdown_writes_trigger: 8
  level0_stop_writes_trigger: 16
  
  enable_http2: true
  compression: "zstd"
  
  # Separate WAL auf NVMe
  wal_dir: /nvme/rocksdb/wal

Bulk-Import-Konfiguration

storage:
  memtable_size_mb: 2048
  max_write_buffer_number: 8
  db_write_buffer_size_mb: 16384
  max_background_jobs: 16
  
  # Aggressive compaction
  level0_file_num_compaction_trigger: 2
  max_background_compactions: 12
  
  # Direct I/O
  use_direct_io_for_flush_and_compaction: true
  
  # Async WAL (Trade-off: Datenverlust-Risiko)
  sync: false
  wal_bytes_per_sync: 67108864  # 64 MB

---

🎯 Empfohlener Aktionsplan

Phase 1: Quick Wins (Woche 1-2)

1. ✅ Adaptive Write Buffer Sizing implementieren
2. ✅ HTTP/2 aktivieren und testen
3. ✅ Client Connection Pooling in SDKs
4. ✅ Payload Compression (Zstd) aktivieren
5. ✅ Level0 Compaction tunen

Erwarteter Gewinn: +100-200% Throughput

---

Phase 2: Medium-term (Monat 1-2)

1. ✅ Product Quantization für Embeddings
2. ✅ Memory-Mapped File Import
3. ✅ Direct I/O für Bulk Import
4. ⚠️ Per-Thread Memtables (RocksDB-Patch)

Erwarteter Gewinn: +200-500% für spezifische Workloads

---

Phase 3: Long-term (Monat 3-6)

1. ⚠️ Async WAL mit Group Commit (Trade-off Durability)
2. ✅ Adaptive Batch Sizing
3. ✅ Priority-based Queues
4. ✅ Multi-Level Buffering

---

📚 Referenzen

• RocksDB Tuning Guide
• RocksDB Performance Benchmarks
• HTTP/2 Performance Best Practices
• Product Quantization Paper
• Gorilla Time Series Compression
• Zstandard Compression
• Memory-Mapped I/O Performance

---

Autor: ThemisDB Performance Team
Review: Engineering Lead
Status: Ready for Implementation