[C++] Advanced HTS Real-Time Market Data Processing: Handling Multiple PB Service Feeds and Per-Symbol Parallelism (CAS vs Shared Mutex)

In the previous post, I designed the skeleton of a lock-free architecture that combines a custom deleter for std::shared_ptr with an object pool (Object Pool) to resolve bottlenecks in the HTS data pipeline.

However, in an actual brokerage's production environment, you run into far more complicated edge cases than that. The two most representative ones are "a situation where multiple PB (push) services update a single data structure piece by piece" and "a situation where thousands of symbols' data must be managed in parallel at the same time."

Today I want to take the previous architecture a step further and walk through how I built a fully parallel, multi-symbol real-time processing system using a CAS (Compare-And-Swap) loop with std::atomic and std::shared_mutex.

1. The Dilemma of Receiving Multiple PB Services and Partial Updates

HTS real-time market data doesn't arrive from a single server bundled together all at once. For example, it arrives in a distributed fashion, where Service A pushes the "current price" and Service B pushes the "trading volume."

So what happens when both services try to update the same symbol's data just 0.001 seconds apart? If you use a simple overwrite (store) approach, B ends up overwriting A's earlier update with stale data, causing a lost update problem.

2. The Solution: CAS (Compare-And-Swap) Loop and the RCU Pattern

The ultimate technique for resolving this dilemma without a lock is the CAS (Compare-And-Swap) operation. Using C++'s std::atomic::compare_exchange_strong function, we can implement the Read-Copy-Update (RCU) pattern.

void OnReceiveServiceA(const std::string& code, int p1, long v1) {
    std::shared_ptr<MarketData> currentSnapshot = m_latestData.load();

    while (true) { // Retry indefinitely until it succeeds (CAS loop)
        MarketData* rawPtr = m_pool.Acquire();

        if (currentSnapshot) {
            *rawPtr = *currentSnapshot; // 1. Read-Copy (preserve the latest state)
        }

        rawPtr->code = code;
        rawPtr->price1 = p1; // 2. Update (only the field this service owns)
        rawPtr->volume1 = v1;

        auto newData = std::shared_ptr<MarketData>(rawPtr, [this](MarketData* p) {
            m_pool.Release(p); // custom deleter (automatic return to pool)
        });

        // 3. Attempt the swap
        // If Service B slipped in during that split second, this fails and we re-enter the loop
        if (m_latestData.compare_exchange_strong(currentSnapshot, newData)) {
            break;
        }
    }
}

With this structure, even when multiple threads attempt updates at the same time, the data merges safely with no lock waiting involved, and the temporary data from any failed update attempt is immediately returned to the pool by the custom deleter, so no memory leak occurs either.

3. Architecture for Scaling to Multiple Symbols: std::unordered_map + std::shared_mutex

Since the market data for every symbol doesn't arrive into one single structure, we need to extend this lock-free architecture so that thousands of instances of it can run at once, keyed by symbol code.

To do this, we introduce std::unordered_map, and to prevent structural corruption of the map, we apply a Double-Checked Locking pattern combined with C++14's std::shared_mutex (a read/write lock).

struct StockContext {
    std::atomic<std::shared_ptr<MarketData>> latestData;
    StockContext(std::shared_ptr<MarketData> initData) {
        latestData.store(initData);
    }
};

class DataManager {
private:
    MarketDataPool m_pool;
    std::unordered_map<std::string, std::unique_ptr<StockContext>> m_stockMap;
    std::shared_mutex m_mapMutex;

    StockContext* GetOrCreateContext(const std::string& code) {
        // 1. Shared lock (read lock): dozens of threads can look up the map concurrently at high speed
        {
            std::shared_lock<std::shared_mutex> readLock(m_mapMutex);
            auto it = m_stockMap.find(code);
            if (it != m_stockMap.end()) return it->second.get();
        }

        // 2. Exclusive lock (write lock): only change the map structure for a brand-new symbol
        std::unique_lock<std::shared_mutex> writeLock(m_mapMutex);
        auto it = m_stockMap.find(code);
        if (it == m_stockMap.end()) {
            MarketData* rawPtr = m_pool.Acquire();
            auto initData = std::shared_ptr<MarketData>(rawPtr, [this](MarketData* p){ m_pool.Release(p); });

            auto context = std::make_unique<StockContext>(initData);
            // Move memory ownership fully into the map, constructing in place
            auto result = m_stockMap.emplace(code, std::move(context));
            return result.first->second.get();
        }
        return it->second.get();
    }
};

You can work out the real-time data receive logic for this case pretty easily just by taking the OnReceiveServiceA function from the code block in section 2 and swapping out m_latestData for latestData inside StockContext.

4. Wrapping Up

Working through this advanced design, I could feel both the complexity and the appeal of real-time, high-volume data processing at the same time. The combination of "the bare minimum read-write lock needed to protect the map's structure" and "a 100% lock-free CAS loop for updating the data inside each symbol" may well be the most ideal architecture for an HTS system that has to achieve extreme rendering performance and data consistency at the same time, on a modern multi-core environment.