spark/src/lib/orderman/server/OrderServiceServer.cpp

#include "OrderServiceServer.h"
#include "lib/api/public/ServiceContext.h"
#include "lib/api/services/AdminService.client.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/jetfix.pb.h"
#include "lib/config/protos/orderman_cfg.pb.h"
#include "lib/md/MDConsumer.h"
#include "lib/mercury/actors/Timer.h"
#include "lib/mercury/executor/Executor.h"
#include "lib/metrics/metrics.h"
#include "lib/orderman/actors/AsyncFactory.h"
#include "lib/orderman/actors/ExecutionGateway.h"
#include "lib/orderman/actors/FillStore.h"
#include "lib/orderman/actors/MDSource.h"
#include "lib/orderman/actors/OrderStore.h"
#include "lib/orderman/actors/SecuritySrc.h"
#include "lib/orderman/actors/ServiceBridge.h"
#include "lib/orderman/actors/order_manager/OrderManager.h"
#include "lib/utils/bts_chrono.h"
#include "lib/utils/type_name.h"
#include "lib/utils/thread_manager.h"

#include <boost/algorithm/string/predicate.hpp>

namespace bts {
namespace orderman {

struct ActorThread {
   ActorThread(const config::OrdermanCfg::Executor &cfg,metrics::MetricsContainer &stats)
       : cfg(cfg), thread(), executor(), loop_metric(), done(false) {
      std::array<double, 30> buckets;
      auto i = 0;
      for (auto &b : buckets)
         b = (i += 1000);

      loop_metric = std::make_shared<metrics::HistogramMetric<30>>(
          fmt::format("spark_metrics_{}_thread_loop_ns", cfg.thread_name()), buckets);
      stats.publish(loop_metric);
   }

   void start() {
      thread = threading::make_thread(cfg.thread_name(),[this]() {
         const auto inner_iter = 1000;
         while (!done) {
            auto start = Timer::now();
            for (auto i = 0; i < inner_iter; i++)
               executor.do_work();
            auto stop = Timer::now();
            loop_metric->observe(ns_diff(stop, start) / inner_iter);
            if (!cfg.poll())
               std::this_thread::sleep_for(10ms);
         }
      });
   }

   ~ActorThread() {
      done= true;
      thread.join();
   }

   config::OrdermanCfg::Executor cfg;
   std::thread thread;
   mercury::Executor executor;
   std::shared_ptr<metrics::HistogramMetric<30>> loop_metric;
   std::atomic<bool> done;
};

ActorThread *find_thread_for_actor(
    std::string const &actor_name,
    std::vector<std::unique_ptr<ActorThread>> const &all_executors) {
   for (auto const &ex : all_executors)
      for (auto const &hosted : ex->cfg.hosted_actor())
         if (hosted == "*" || boost::algorithm::to_lower_copy(hosted) ==
             boost::algorithm::to_lower_copy(actor_name))
            return ex.get();
   return nullptr;
}

struct OrderServiceServer::PImpl {
   PImpl(bts::api::ServiceContext &ctx,
         std::shared_ptr<api::ISyntheticSecurityService> &ss,
         metrics::MetricsContainer &stats)
       : service_bridge(), service_ready(false), known_orders(), known_fills(),
         publish_orders(), publish_fills(),
         admin_service(ctx.locate<api::admin::AdminService>()), md_consumer(ss),
         stats(stats) {}

   ServiceBridge* service_bridge;

   std::vector<std::unique_ptr<ActorThread>> actor_threads;

   std::atomic<bool> service_ready;
   std::unordered_set<OrderKey> known_orders;
   std::unordered_set<std::string> known_fills;

   OrderServiceServer::OrdersPublisher publish_orders;
   OrderServiceServer::FillsPublisher publish_fills;
   api::admin::AdminService admin_service;
   bts::md::MDConsumer md_consumer;
   metrics::MetricsContainer &stats;
};

OrderServiceServer::OrderServiceServer(
    bts::api::ServiceContext &ctx,
    std::shared_ptr<api::ISyntheticSecurityService> &ss,
    metrics::MetricsContainer &stats)
    : api::orderman::OrderServiceBase(ctx), me(new PImpl(ctx, ss, stats)) {

   me->publish_orders = make_orders_publisher();
   me->publish_fills = make_fills_publisher();

   std::vector<mercury::Executor*> all_executors;
   for (auto const &ex_cfg : config::get<config::OrdermanCfg>().executor()) {
      auto at = std::make_unique<ActorThread>(ex_cfg,stats);
      all_executors.push_back(&at->executor);
      me->actor_threads.push_back(std::move(at));
   }

   for (auto &t : me->actor_threads)
      t->executor.init_neighbours(all_executors);
   
   // MDSource
   auto thread = find_thread_for_actor("MDSource", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for MDSource. See the "
                     "OrdermanCfg key");
   thread->executor.spawn<MDSource>(me->md_consumer, me->stats);

   // OrderManager
   thread = find_thread_for_actor("OrderManager", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for OrderManager. See "
                     "the OrdermanCfg key");
   thread->executor.spawn<OrderManager>();

   // ExecutionGateway's
   thread = find_thread_for_actor("ExecutionGateway", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for ExecutionGateway "
                     "instances. See the OrdermanCfg key");
   auto &all_connections = config::get<config::JetFIX>().connection();
   for (auto const &c : all_connections)
      thread->executor.spawn<ExecutionGateway>(c, stats);

   // Timer
   thread = find_thread_for_actor("Timer", me->actor_threads);
   bts_assert_msg(
       thread,
       "Could not find a configured thread for Timer. See the OrdermanCfg key");
   thread->executor.spawn<mercury::Timer>();

   // AsyncFactory
   thread = find_thread_for_actor("AsyncFactory", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for AsyncFactory. See "
                     "the OrdermanCfg key");
   thread->executor.spawn<AsyncFactory>();

   // OrderStore
   thread = find_thread_for_actor("OrderStore", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for OrderStore. See "
                     "the OrdermanCfg key");
   thread->executor.spawn<OrderStore>();

   // FillStore
   thread = find_thread_for_actor("FillStore", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for FilllStore. See "
                     "the OrdermanCfg key");
   thread->executor.spawn<FillStore>();

   // SecuritySrc
   thread = find_thread_for_actor("SecuritySrc", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for SecuritySrc. See "
                     "the OrdermanCfg key");
   thread->executor.spawn<SecuritySrc>(me->md_consumer);

   // ServiceBridge
   thread = find_thread_for_actor("ServiceBridge", me->actor_threads);
   bts_assert_msg(thread, "Could not find a configured thread for ServiceBridge. See "
                     "the OrdermanCfg key");
   me->service_bridge = new ServiceBridge(
       [this]() { me->service_ready = true; },

       [this, &ctx](auto const &orders) {
          ctx.post([this, orders] { this->on_orders(orders); });
       },

       [this, &ctx](auto const &fill) {
          ctx.post([this, fill] { this->on_fills(fill); });
       },
       thread->executor.make_proxy() 
      );

   thread->executor.activate(me->service_bridge);

   // spin up threads
   for (auto &t : me->actor_threads)
      t->start();

   // wait for the service bridge to be ready
   me->service_bridge->block_until_ready();

   // subscribe to admin model updates and forward them to the service bridge
   me->admin_service.subscribe_admin_data() |
       rxcpp::operators::subscribe<api::admin::Admin_dataUpdate>(
           [this](auto const &update) {
              bts_assert(update.all().size() == 1);
              me->service_bridge->update_admin_data(update.all()[0]);
           });
}

OrderServiceServer::~OrderServiceServer() {
   bts_assert(false); // this is badly broken at the moment
}

void OrderServiceServer::on_orders(
    std::vector<api::orderman::OrderRecord> const &orders) {
   LOG(OrderMan, Info, "Received order update for publishing");
   std::vector<api::orderman::OrderRecord> created;
   created.reserve(orders.size());
   std::vector<api::orderman::OrderRecord> updated;
   updated.reserve(orders.size());

   for (auto const &rec : orders) {
      auto const &k = OrderKey(rec.key());
      if (me->known_orders.count(k) == 0) {
         me->known_orders.insert(k);
         created.push_back(rec);
      } else
         updated.push_back(rec);
   }

   LOG(OrderMan, Info,
       "Publishing orders to the service API. Created {}, Updated {}",
       created.size(), updated.size());

   me->publish_orders(&created, &updated, nullptr);
}

void OrderServiceServer::on_fills(
    std::vector<api::orderman::Fill> const &fills) {
   LOG(OrderMan, Info, "Received fill update for publishing");
   std::vector<api::orderman::Fill> created;
   created.reserve(fills.size());
   std::vector<api::orderman::Fill> updated;
   updated.reserve(fills.size());

   for (auto const &f : fills) {
      auto const &k = f.fill_key();
      if (me->known_fills.count(k) == 0) {
         me->known_fills.insert(k);
         created.push_back(f);
      } else
         updated.push_back(f);
   }

   me->publish_fills(&created, &updated, nullptr);
}

void OrderServiceServer::create_order(
    const api::orderman::Create &c,
    const api::Response<api::orderman::Response> &r) {
   api::orderman::Response response;
   if (!me->service_ready)
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
   else
      response = me->service_bridge->create_order(c);
   r(response);
}

void OrderServiceServer::cancel_order(
    const api::orderman::Cancel &cancel,
    const api::Response<api::orderman::Response> &r) {
   api::orderman::Response response;
   response.set_order_key(cancel.order_key());

   if (!me->service_ready)
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
   else
      response = me->service_bridge->cancel_order(cancel);
   r(response);
}

void OrderServiceServer::update_order(
    const api::orderman::Update &update,
    const api::Response<api::orderman::Response> &r) {
   api::orderman::Response response;
   response.set_order_key(update.order_key());

   if (!me->service_ready)
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
   else
      response = me->service_bridge->update_order(update);
   r(response);
}

void OrderServiceServer::crack_order(
    const api::orderman::Crack &crack,
    const api::Response<api::orderman::Response> &r) {
   api::orderman::Response response;
   response.set_order_key(crack.order_key());

   if (!me->service_ready)
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
   else
      response = me->service_bridge->crack_order(crack);
   r(response);
}
} // namespace orderman
} // namespace bts