DO_NOT_MERGE: Add diagnostic logging

src/client/sparkcmd/Cmd/MBOCommand.cs

@@ -24,6 +24,7 @@ namespace bts {
       }
       
       static void dump_mbo(uint key, api.Side side) {
+         // todo: fix me
          using (var ctx = new ServiceContext()) {
             var mbo = ctx.locate<MBOService>();
             var data = mbo.get_orders_for_security(new MBOSecurityQuery { SecurityId = key });

src/lib/md/MBOProcessor.cpp

@@ -3,6 +3,7 @@
 #include "lib/config/config.h"
 #include "lib/config/protos/toggles.pb.h"
 #include "lib/utils/thread_manager.h"
+#include "lib/utils/LockBox.h"
 
 #include <map>
 #include <functional>
@@ -63,6 +64,7 @@ void print_pool_stats(const utils::ConcurrentPool<MBOSnapshot>& pool) {
 }
 
 typedef tbb::concurrent_hash_map<SecurityKey, MBOSnapshot*, SecurityKeyHashCompare> MapType;
+// typedef std::unordered_map<SecurityKey, MBOSnapshot*> MapType;
 struct Incremental {
    MBOWorkingState* state;
    inc::Update update;
@@ -71,6 +73,7 @@ struct MBOProcessorImpl {
    utils::ConcurrentPool<MBOSnapshot> pool;
    tbb::concurrent_queue<Incremental> incremental_input;
    MapType snapshots;
+   // utils::LockBox<MapType> snapshots;
    bool running;
    utils::TimeStamp last_pool_print;
    std::thread thread;
@@ -85,7 +88,6 @@ struct MBOProcessorImpl {
       if (config::get<config::toggles::MBOStats>().enabled()) {
          last_pool_print = utils::TimeStamp::now();
       }
-
    }
 
    ~MBOProcessorImpl() {
@@ -94,14 +96,25 @@ struct MBOProcessorImpl {
    }
 
    void publish_snapshot(SecurityKey key, MBOSnapshot** snapshot) {
+      LOG(md,Diag, "publish_snapshot/begin/{}", key.value());
       bts_assert(*snapshot != nullptr);
       MapType::accessor accessor;
       snapshots.insert(accessor, key);
       std::swap(accessor->second, *snapshot);
       if (*snapshot != nullptr) {
          // Note: relase() must happen inside accessor lifetime to ensure correctness
+         LOG(md,Diag, "publish_snapshot/release/{}", key.value());
          pool.release(snapshot);
       }
+      LOG(md,Diag, "publish_snapshot/end/{}", key.value());
+
+      // snapshots.lock([&](MapType& map) {
+      //    std::swap(map[key], *snapshot);
+      //    // Note: relase() must happen inside accessor lifetime to ensure correctness
+      //    if (*snapshot != nullptr) {
+      //       pool.release(snapshot);
+      //    }
+      // });
    }
 
    void publish_incremental(MBOWorkingState& state) {
@@ -139,6 +152,12 @@ struct MBOProcessorImpl {
          }
 
          // publish a small batch of snapshots (then check queue again)
+
+         auto sz = dirty.size();
+         if (sz > 0) {
+            LOG(md,Diag, "process_incremental/publish/sz={}", sz);
+         }
+
          for (auto iter = dirty.begin(); iter != dirty.end(); ++iter) {
             publish_incremental(**iter);
             clean.push_back(*iter);
@@ -276,6 +295,7 @@ void apply_incremental(uint32_t security_key, C& collection, std::unordered_map<
    }
 }
 void apply_incremental(MBOWorkingState& state, const inc::Update& update) {
+   LOG(md,Diag, "apply_incremental/begin/{}", state.handle.key());
    // apply the update to the working state
    if (update.type == inc::UpdateType::Bid) {
       apply_incremental(state.handle.key(), state.bid, state.orderid_index, update);
@@ -300,6 +320,15 @@ MBOSnapshot* MBOReader::acquire_snapshot(SecurityKey key) const {
    else {
       release_snapshot(&result);
    }
+   // _me->snapshots.lock([&](const MapType& map) {
+   //    auto iter = map.find(key);
+   //    if (iter != map.end()) {
+   //       result->copy_from(*iter->second);
+   //    }
+   //    else {
+   //       release_snapshot(&result);
+   //    }
+   // });
    return result;
 }
 void MBOReader::release_snapshot(MBOSnapshot**s) const {

src/lib/md/MBOProcessor.h

@@ -136,9 +136,9 @@ class MBOWriter {
    std::shared_ptr<MBOProcessorImpl> _me;
    friend class MBOProcessor;
    MBOWriter(std::shared_ptr<MBOProcessorImpl>&);
-   ~MBOWriter();
 
    public:
+   ~MBOWriter();
    //
    // Snapshot api
    MBOSnapshot* allocate_snapshot(size_t bid_count=0, size_t ask_count=0);
@@ -158,8 +158,8 @@ class MBOReader {
    friend class MBOProcessor;
    friend class MBOWriter;
    MBOReader(std::shared_ptr<MBOProcessorImpl>&);
-   ~MBOReader();
    public:
+   ~MBOReader();
    MBOSnapshot* acquire_snapshot(SecurityKey key) const;
    void release_snapshot(MBOSnapshot**) const;
 };

src/lib/md/MBOServiceServer.cpp

@@ -309,8 +309,8 @@ void MBOServiceServer::get_orders_for_security(const MBOSecurityQuery& request,
    }
 
    size_t count = 0;
-   auto tick = handle.get_minimum_tick();
    if(handle.is_initialized()) {
+      auto tick = handle.get_minimum_tick();
       auto mbo = handle.get_mbo();
       if (mbo) {
          for (const auto& o : mbo->bids) {
@@ -346,7 +346,15 @@ void MBOServiceServer::get_orders_for_security(const MBOSecurityQuery& request,
             }
          }
       }
+      else {
+         LOG(md,Diag,"get_orders_for_security/fail/null/{}", handle.key());
+      }
    }
+   else {
+      LOG(md,Diag,"get_orders_for_security/fail/uninitialized/{}", handle.key());
+   }
+   
+   LOG(md,Diag,"get_orders_for_security/{}/{}b/{}a/{}", handle.key(), response.bid_size(), response.ask_size(), request.price_level());
    
    reply(response);
 }

src/lib/md/MDConsumerHandle.cpp

@@ -379,6 +379,9 @@ std::unique_ptr<MBOSnapshot,std::function<void(MBOSnapshot*)>> MDConsumerHandle:
          if (snapshot != nullptr) {
             return {snapshot,[mbo=book.mbo](auto* s) { mbo->release_snapshot(&s); }};
          }
+         else {
+            LOG(md,Diag,"get_mbo/fail/{}", key());
+         }
       }
       return nullptr;
    }

src/lib/md/test/mbo_tests.cpp

@@ -2,10 +2,12 @@
 #include "catch/catch.hpp"
 #include "../IntPrice.h"
 #include "../MBOProcessor.h"
+#include "../MBOServiceServer.h"
 #include "Co2TestInstrument.h"
 
 #include <random>
 #include <algorithm>
+#include <thread>
 
 namespace bts {
 namespace md {
@@ -316,5 +318,92 @@ TEST_CASE("MBO change price and size", "[md][mbo]") {
    REQUIRE(!not_ask);
 }
 
+TEST_CASE("MBO stress", "[md][mbo]") {
+   /*
+      This test attempts to shake out concurrency issues with MBO processing
+      If it fails DO NOT ignore it
+   */
+
+   MBOProcessor p;
+   bool running = true;
+
+   std::vector<SecurityKey> keys;
+   for (auto x = 0; x < 157; ++x) {
+      keys.push_back(SecurityKey(x));
+   }
+
+   auto writer = p.writer();
+   size_t write_count = 0;
+   size_t wtime = 0;
+   size_t wcount = 0;
+   std::thread twriter([&]() {
+      size_t count = 0;
+      while(running) {
+         auto start = utils::TimeStamp::now();
+         for (const auto& k : keys) {
+            ++count;
+            auto s = writer.allocate_snapshot();
+            REQUIRE(s);
+            for (int t = 0; t < 15000; ++t) {
+               s->bids.push_back({ ExchangeOrderKey(count%1024 + t), { 101, -2 }, 42 });
+               s->asks.push_back({ ExchangeOrderKey(count%1024 + 10000 + t), { 111, -2 }, 32 });
+               ++write_count;
+            }
+            writer.publish_snapshot(k, &s);
+         }
+         auto end = utils::TimeStamp::now();
+         ++wcount;
+         wtime += (end - start).milliseconds();
+      }
+   });
+   // Ensure writer has written each once
+   while(wcount < 1)
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+   std::cout << "wcount: " << wcount << " - time: " << (wtime/wcount) << std::endl;
+
+   auto reader = p.reader();
+   std::vector<std::thread> readers;
+   size_t time = 0;
+   size_t count = 0;
+   size_t nullcount = 0;
+
+   for (int x = 0; x<1; ++x) {
+      readers.push_back(std::thread([&]() {
+         while(running) {
+            auto start = utils::TimeStamp::now();
+            for (const auto& k : keys) {
+               auto s = reader.acquire_snapshot(k);
+               if (s == nullptr) nullcount++;
+               (void)s;
+               reader.release_snapshot(&s);
+               // REQUIRE(s);
+            }
+            auto end = utils::TimeStamp::now();
+            ++count;
+            time += (end - start).microseconds();
+         }
+      }));
+   }
+
+   std::this_thread::sleep_for(std::chrono::seconds(10));
+   running = false;
+   twriter.join();
+   for (auto& t : readers) t.join();
+
+   std::cout << "Write time: " << wtime/wcount << std::endl;
+   std::cout << "Read time: " << time/count << std::endl;
+   std::cout << "nullcount: " << nullcount << std::endl;
+   REQUIRE(nullcount == 0);
+   /*
+
+   Next step:
+
+   - Replace concurrent_hash with Lockbox<Unordered> and see if we can still make it happen
+
+   */
+}
+
+
+
 }}}
 #endif

src/lib/utils/ConcurrentPool.h

@@ -24,6 +24,18 @@ class ConcurrentPool {
          delete [] b;
    }
 
+#if 0
+   T* allocate() {
+      return new T();
+   }
+   void reserve(size_t ) {
+   }
+   void release(T** item) {
+      if (*item) {
+         delete *item;
+      }
+   }
+#else
    T* allocate() {
       T* result = nullptr;
       ++_outstanding;
@@ -46,6 +58,7 @@ class ConcurrentPool {
       _all.push_back(buffer);
    }
 
+
    void release(T** item) {
       if (*item) {
          (*item)->reset();
@@ -55,6 +68,8 @@ class ConcurrentPool {
       }
    }
 
+#endif
+
    template <class CB>
    void unsafe_iterate(CB cb) const {
       for (auto iter = _all.begin(); iter != _all.end(); ++iter) {