Drop folly use tbb instead

Author: dsirotkin256

.gitignore

@@ -1,5 +1,6 @@
 build
-builder
+core
+Debug
 
 # Prerequisites
 *.d

CMakeLists.txt

@@ -26,9 +26,7 @@ option(ENABLE_BENCHMARKS "Build benchmarks" YES)
 
 # Compiler configuration
 set(CMAKE_CXX_STANDARD 17)
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fexceptions -Wdeprecated -Wextra -Wno-switch-enum -Wno-float-equal -Wno-unused-parameter -Wno-unused-member-function -Werror -fno-omit-frame-pointer -fno-builtin-malloc -fno-builtin-calloc -fno-builtin-realloc -fno-builtin-free -lpthread -lm")
-
-#set(CMAKE_BUILD_TYPE Release)
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fexceptions -Wdeprecated -Wextra -Wno-switch-enum -Wno-float-equal -Wno-unused-parameter -Wno-unused-member-function -Werror -fno-omit-frame-pointer -fno-builtin-malloc -fno-builtin-calloc -fno-builtin-realloc -fno-builtin-free -lpthread -lm -march=native")
 
 set(CMAKE_THREAD_PREFER_PTHREAD TRUE)
 set(CMAKE_HAVE_THREADS_LIBRARY TRUE)

Matching/CMakeLists.txt

@@ -4,17 +4,16 @@ target_sources(matching INTERFACE ${CMAKE_CURRENT_SOURCE_DIR})
 target_link_libraries(
   matching
   INTERFACE
-    CONAN_PKG::jemalloc
-    CONAN_PKG::boost
-    CONAN_PKG::spdlog
-    CONAN_PKG::tbb
-    CONAN_PKG::folly
-    CONAN_PKG::nlohmann_json
+  CONAN_PKG::jemalloc
+  CONAN_PKG::boost
+  CONAN_PKG::spdlog
+  CONAN_PKG::tbb
+  CONAN_PKG::nlohmann_json
 )
 
 target_include_directories(
     matching
     INTERFACE
-      $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/src>
-      $<INSTALL_INTERFACE:include/matching>
+    $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/src>
+    $<INSTALL_INTERFACE:include/matching>
 )

Matching/src/order_router.hpp

@@ -3,15 +3,13 @@
 #include <orderbook.hpp>
 #include "influxdb.hpp"
 #include "spdlog/spdlog.h"
-#include <future>
 #include <sys/types.h>
 #include <unistd.h>
 #include <pthread.h>
 #include <map>
-#include <folly/concurrency/UnboundedQueue.h>
 #include <boost/asio.hpp>
 #include <boost/asio/thread_pool.hpp>
-//#include <boost/lockfree/queue.hpp>
+#include <tbb/concurrent_queue.h>
 
 namespace matching_engine
 {
@@ -32,21 +30,22 @@ class consumer
     }
     void push(OrderPtr order)
     {
-        queue_.enqueue(std::move(order));
+        queue_.push(std::move(order));
     }
     void register_market(std::string_view market)
     {
         markets.emplace(market, market);
     }
     void listen()
     {
-        for (const auto& [name, _] : markets) {
-            console_->info("Consumer of {} started @{}", name, (pid_t) syscall (SYS_gettid));
-        }
+        if (console_ != nullptr)
+            for (const auto& [name, _] : markets) {
+                console_->info("Consumer of {} started @{}", name, (pid_t) syscall (SYS_gettid));
+            }
         OrderPtr order;
         auto last_log = Time::now();
         while(should_consume_()) {
-            queue_.dequeue(order);
+            queue_.pop(order);
             auto &ob = markets.at(order->market_name());
             const auto start = Time::now();
             ob.match(std::move(order));
@@ -66,6 +65,7 @@ class consumer
                 });
                 last_log = start;
             }
+
         }
     }
 private:
@@ -74,8 +74,7 @@ class consumer
         return !should_exit_ or !queue_.empty();
     }
     std::unordered_map<std::string_view, OrderBook> markets;
-    folly::UnboundedQueue<OrderPtr, true, true, true, 16> queue_;
-    //boost::lockfree::queue<OrderPtr> queue_;
+    tbb::concurrent_bounded_queue<OrderPtr> queue_;
     std::atomic_bool should_exit_;
     std::shared_ptr<spdlog::logger> console_;
 };
@@ -85,8 +84,8 @@ class dispatcher
 public:
     dispatcher() = default;
     dispatcher(const dispatcher&) = delete;
-    dispatcher(std::shared_ptr<spdlog::logger> console,
-               std::vector<std::string_view> markets,
+    dispatcher(std::vector<std::string_view> markets,
+               std::shared_ptr<spdlog::logger> console = nullptr,
                const uint64_t available_cores = std::max(1u, std::thread::hardware_concurrency() - 1)):
         pool_{available_cores},
         console_{console}
@@ -121,10 +120,17 @@ class dispatcher
         auto& market_consumer = market_registry_.at(order->market_name());
         market_consumer->push(std::move(order));
     }
-    std::string_view registered_market_name(std::string_view market) const
+    std::string_view registered_market_name(const std::string_view market) const
     {
         return market_registry_.find(market)->first;
     }
+    void shutdown()
+    {
+        for (auto const& [_, c] : market_registry_) {
+            c->shutdown();
+        }
+        pool_.join();
+    }
 private:
     //tbb::concurrent_unordered_map<std::string, std::shared_ptr<consumer>> market_registry_;
     std::unordered_map<std::string_view, std::shared_ptr<consumer>> market_registry_;

Matching/src/orderbook.hpp

@@ -92,7 +92,7 @@ class Order
     bool operator==(const Order &rhs) const;
 };
 
-using OrderPtr = std::unique_ptr<Order>;
+using OrderPtr = std::shared_ptr<Order>;
 
 using queue_allocator = boost::container::allocator<OrderPtr>;
 using order_queue_type = boost::container::deque<OrderPtr, queue_allocator>;

Service/CMakeLists.txt

@@ -9,8 +9,8 @@ add_executable(
 target_link_libraries(
   matching_service
   PRIVATE
-    matching
-    CONAN_PKG::jemalloc
+  matching
+  CONAN_PKG::jemalloc
 )
 
 set_target_properties(service PROPERTIES LINKER_LANGUAGE CXX)

Service/src/main.cpp

@@ -41,7 +41,7 @@ int main(int argc, char *argv[])
         u8"EUR_AUD",
         u8"GBP_JPY", u8"USD_JPY"
     };
-    auto dispatcher = std::make_shared<me::router::dispatcher>(console, markets);
+    auto dispatcher = std::make_shared<me::router::dispatcher>(markets, console);
 
     /* Initialise TCP transport layer */
     boost::asio::io_context ioc{(int)std::thread::hardware_concurrency()};

Testing/Benchmark/main.cpp

@@ -23,10 +23,17 @@ static void OrderCreation(benchmark::State& state)
     Price price = rand() % 100 + 1;
     Quantity quantity = double(rand() % 100 + 1) / (rand() % 20 + 1);
     for (auto _ : state) {
+        auto start = std::chrono::high_resolution_clock::now();
         auto order = std::make_unique<Order>(market, side, price, quantity);
+        auto end = std::chrono::high_resolution_clock::now();
+        auto elapsed_seconds =
+            std::chrono::duration_cast<std::chrono::duration<double>>(
+                end - start);
+        state.SetIterationTime(elapsed_seconds.count());
     }
+    state.SetComplexityN(state.range(0));
 }
-BENCHMARK(OrderCreation)->DenseRange(0, 1000, 250);
+BENCHMARK(OrderCreation)->DenseRange(1, 1000, 250)->UseManualTime()->Complexity(benchmark::oN);
 
 static void OrderMatching(benchmark::State& state)
 {
@@ -39,11 +46,38 @@ static void OrderMatching(benchmark::State& state)
             auto side = rand() % 2 ? SIDE::BUY : SIDE::SELL;
             auto quantity = double(rand() % 10 + 1) / (rand() % 20 + 1);
             auto order = std::make_unique<Order>(market, side, price, quantity);
-            //dispatcher->send(std::move(order));
+            auto start = std::chrono::high_resolution_clock::now();
             ob.match(std::move(order));
+            auto end = std::chrono::high_resolution_clock::now();
+            auto elapsed_seconds =
+                std::chrono::duration_cast<std::chrono::duration<double>>(
+                    end - start);
+            state.SetIterationTime(elapsed_seconds.count());
         }
     }
+    state.SetComplexityN(state.range(0));
 }
-BENCHMARK(OrderMatching)->DenseRange(0, 1000, 250);
-// Run the benchmark
+BENCHMARK(OrderMatching)->DenseRange(1, 1000, 250)->UseManualTime()->Complexity(benchmark::oLogN);
+
+// TODO Starts and hangs, issue with thread pool. Fix it.
+static void OrderDispatching(benchmark::State& state)
+{
+    // Perform setup here
+    const std::vector<std::string_view> markets = {u8"USD_JPY"};
+    auto dispatcher = std::make_shared<router::dispatcher>(markets);
+    auto prices = SimulateMarket(state.range(0));
+    for(auto _ : state) {
+        for (auto price : prices) {
+            auto side = rand() % 2 ? SIDE::BUY : SIDE::SELL;
+            auto quantity = double(rand() % 10 + 1) / (rand() % 20 + 1);
+            auto order = std::make_unique<Order>(markets[0], side, price, quantity);
+            dispatcher->send(std::move(order));
+        }
+    }
+    dispatcher->shutdown();
+}
+
+//BENCHMARK(OrderDispatching)->DenseRange(0, 1000, 250)->MeasureProcessCPUTime();
+
+/* Run the benchmark */
 BENCHMARK_MAIN();

Testing/UnitTest/CMakeLists.txt

@@ -6,9 +6,9 @@ add_executable(
 target_link_libraries(
   unit_tests
   PRIVATE
-    matching
-    Threads::Threads
-    CONAN_PKG::jemalloc
-    CONAN_PKG::gtest
+  matching
+  Threads::Threads
+  CONAN_PKG::jemalloc
+  CONAN_PKG::gtest
 )
 

conanfile.txt

@@ -6,7 +6,6 @@ spdlog/1.5.0
 nlohmann_json/3.7.3
 jemalloc/5.2.1
 tbb/2020.0
-folly/2018.11.12.00@bincrafters/stable
 
 [generators]
 txt