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/broker/AsyncEventBroker.h"
#include "lib/mercury/mercury_utils/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"

namespace bts {
namespace orderman {

struct OrderServiceServer::PImpl {
   PImpl(bts::api::ServiceContext &ctx,
         std::shared_ptr<api::ISyntheticSecurityService> &ss,
         metrics::MetricsContainer &stats)
       : fast_broker(std::make_shared<mercury::AsyncEventBroker>()),
         slow_broker(std::make_shared<mercury::AsyncEventBroker>()),
         service_bridge(), fast_exe(*fast_broker), slow_exe(*slow_broker),
         fast_thread(), slow_thread(), service_ready(false), known_orders(),
         known_fills(), publish_orders(), publish_fills(),
         admin_service(ctx.locate<api::admin::AdminService>()), md_consumer(ss),
         stats(stats) {}

   std::shared_ptr<mercury::AsyncEventBroker> fast_broker;
   std::shared_ptr<mercury::AsyncEventBroker> slow_broker;
   std::shared_ptr<ServiceBridge> service_bridge;
   mercury::Executor fast_exe;
   mercury::Executor slow_exe;
   std::thread fast_thread;
   std::thread slow_thread;
   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();

   me->service_bridge = std::make_shared<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); });
       });

   me->fast_broker->init_neighbours({me->slow_broker});
   me->slow_broker->init_neighbours({me->fast_broker});

   // fast actors
   me->fast_exe.spawn<MDSource>(me->md_consumer, me->stats);
   me->fast_exe.spawn<OrderManager>();

   auto &all_connections = config::get<config::JetFIX>().connection();
   for (auto const &c : all_connections)
      me->fast_exe.spawn<ExecutionGateway>(c, stats);

   me->fast_thread = std::thread([this]() {
      os::set_thread_name("fast_exe");

      std::array<double, 30> buckets;
      auto i = 1;
      for (auto &b : buckets)
         b = (i *= 2);

      auto loop_timer = std::make_shared<metrics::HistogramMetric<30>>(
          "spark_metrics_fast_thread_loop_ns", buckets);

      me->stats.publish(loop_timer);
      for (;;) {
         auto start = Timer::now();
         me->fast_exe.do_work();
         auto stop = Timer::now();
         loop_timer->observe(ns_diff(stop, start));
      }
   });

   auto fast_thread_core =
       config::get<config::OrdermanCfg>().fast_thread_affinity();
   if (fast_thread_core >= 0) {
      LOG(OrderMan, Info, "Setting fast thread core affinity to {}",
          fast_thread_core);
      cpu_set_t cpuset;
      CPU_ZERO(&cpuset);
      CPU_SET(fast_thread_core, &cpuset);
      int rc = pthread_setaffinity_np(me->fast_thread.native_handle(),
                                      sizeof(cpu_set_t), &cpuset);
      if (rc != 0)
         LOG(OrderMan, Error,
             "Could not set core affinity to {} for fast thread. Error code={}",
             fast_thread_core, rc);
   }

   // slow actors
   me->slow_exe.activate(me->service_bridge);
   me->slow_exe.spawn<mercury::Timer>();
   me->slow_exe.spawn<AsyncFactory>();
   me->slow_exe.spawn<OrderStore>();
   me->slow_exe.spawn<FillStore>();
   me->slow_exe.spawn<SecuritySrc>(me->md_consumer);
   me->slow_thread = std::thread([this]() {
      os::set_thread_name("slow_exe");
      for (;;)
         if (!me->slow_exe.do_work())
            std::this_thread::sleep_for(10ms);
   });

   // 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