391 lines
14 KiB
C++
391 lines
14 KiB
C++
#include "OrderServiceServer.h"
|
|
#include "lib/api/public/ServiceContext.h"
|
|
#include "lib/api/services/AdminService.client.pb.h"
|
|
#include "lib/api/services/MarginService.client.pb.h"
|
|
#include "lib/api/services/PositionService.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/MDClient.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/FillStore.h"
|
|
#include "lib/orderman/actors/MDSource.h"
|
|
#include "lib/orderman/actors/MDSupplementarySource.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/orderman/events/OrderOpResultEvent.h"
|
|
#include "lib/orderman/order_factory/DynamicOrderRegistry.h"
|
|
#include "lib/plugin/SparkServices.h"
|
|
#include "lib/plugin/plugin.h"
|
|
#include "lib/utils/bts_chrono.h"
|
|
#include "lib/utils/thread_manager.h"
|
|
#include "lib/utils/type_name.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(trace::PerfLog &log) {
|
|
if (cfg.poll())
|
|
thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
|
|
for (;;)
|
|
executor.do_work(log);
|
|
});
|
|
else {
|
|
thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
|
|
const auto inner_iter = 1000;
|
|
while (!done) {
|
|
auto start = Timer::now();
|
|
for (auto i = 0; i < inner_iter; i++)
|
|
executor.do_work(log);
|
|
auto stop = Timer::now();
|
|
loop_metric->observe(ns_diff(stop, start) / inner_iter);
|
|
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, md::MDClient &mdc,
|
|
DynamicOrderRegistry const &d, metrics::MetricsContainer &stats,
|
|
trace::PerfLog &log)
|
|
: service_bridge(), service_ready(false), known_orders(), known_fills(),
|
|
publish_orders(), publish_fills(),
|
|
admin_service(ctx.locate<api::admin::AdminService>()),
|
|
position_service(ctx.locate<api::position::PositionService>()),
|
|
margin_service(ctx.locate<api::position::MarginService>()),
|
|
md_consumer(mdc.current_consumer()), stats(stats),
|
|
dyn_order_registry(d), log(log), in_flight_order_ops(),
|
|
current_req_op_id(0) {}
|
|
|
|
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;
|
|
api::position::PositionService position_service;
|
|
api::position::MarginService margin_service;
|
|
bts::md::MDConsumer &md_consumer;
|
|
metrics::MetricsContainer &stats;
|
|
DynamicOrderRegistry const &dyn_order_registry;
|
|
trace::PerfLog &log;
|
|
std::unordered_map<uint32_t, api::Response<api::orderman::Response>>
|
|
in_flight_order_ops;
|
|
uint32_t current_req_op_id;
|
|
};
|
|
|
|
OrderServiceServer::OrderServiceServer(bts::api::ServiceContext &ctx,
|
|
md::MDClient &mdc,
|
|
DynamicOrderRegistry const &d,
|
|
metrics::MetricsContainer &stats,
|
|
trace::PerfLog &log)
|
|
: api::orderman::OrderServiceBase(ctx),
|
|
me(new PImpl(ctx, mdc, d, stats, log)) {
|
|
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, log);
|
|
|
|
// 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>(d, me->log, mdc.oracle());
|
|
|
|
// 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>(d, mdc.oracle());
|
|
|
|
// 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>(mdc.oracle());
|
|
|
|
// 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>(mdc.oracle());
|
|
|
|
// 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);
|
|
|
|
// MDSupplementarySource
|
|
thread = find_thread_for_actor("MDSupplementarySource", me->actor_threads);
|
|
bts_assert_msg(
|
|
thread,
|
|
"Could not find a configured thread for MDSupplementarySource. See "
|
|
"the OrdermanCfg key");
|
|
thread->executor.spawn<MDSupplementarySource>(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); });
|
|
},
|
|
[this, &ctx](auto const &order_op_result) {
|
|
ctx.post([this, order_op_result] {
|
|
this->on_order_op_result(order_op_result);
|
|
});
|
|
},
|
|
me->dyn_order_registry, thread->executor.make_proxy(), mdc.oracle());
|
|
|
|
thread->executor.activate(me->service_bridge);
|
|
|
|
// spin up threads
|
|
for (auto &t : me->actor_threads)
|
|
t->start(log);
|
|
|
|
// 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]);
|
|
});
|
|
|
|
// subscribe to position updates and foward them to the service bridge
|
|
me->position_service.subscribe_pnl_account_data() |
|
|
rxcpp::operators::subscribe<api::position::Pnl_account_dataUpdate>(
|
|
[this](auto const &update) {
|
|
me->service_bridge->update_account_pnl(update);
|
|
});
|
|
|
|
me->margin_service.subscribe_excess_liquidity_report() |
|
|
rxcpp::operators::subscribe<api::position::Excess_liquidity_reportUpdate>(
|
|
[this](auto const &update) {
|
|
me->service_bridge->update_excess_liquidity_report(update);
|
|
});
|
|
}
|
|
|
|
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) {
|
|
|
|
if (!me->service_ready) {
|
|
api::orderman::Response response;
|
|
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
|
|
r(response);
|
|
} else {
|
|
me->in_flight_order_ops[++me->current_req_op_id] = r;
|
|
me->service_bridge->create_order(me->current_req_op_id, c);
|
|
}
|
|
}
|
|
|
|
void OrderServiceServer::cancel_order(
|
|
const api::orderman::Cancel &cancel,
|
|
const api::Response<api::orderman::Response> &r) {
|
|
|
|
if (!me->service_ready) {
|
|
api::orderman::Response response;
|
|
response.set_order_key(cancel.order_key());
|
|
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
|
|
r(response);
|
|
} else {
|
|
me->in_flight_order_ops[++me->current_req_op_id] = r;
|
|
me->service_bridge->cancel_order(me->current_req_op_id, cancel);
|
|
}
|
|
}
|
|
|
|
void OrderServiceServer::update_order(
|
|
const api::orderman::Update &update,
|
|
const api::Response<api::orderman::Response> &r) {
|
|
|
|
if (!me->service_ready) {
|
|
api::orderman::Response response;
|
|
response.set_order_key(update.order_key());
|
|
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
|
|
r(response);
|
|
} else {
|
|
me->in_flight_order_ops[++me->current_req_op_id] = r;
|
|
me->service_bridge->update_order(me->current_req_op_id, update);
|
|
}
|
|
}
|
|
|
|
void OrderServiceServer::detach_order(
|
|
const api::orderman::Detach &detach,
|
|
const api::Response<api::orderman::Response> &r) {
|
|
|
|
if (!me->service_ready) {
|
|
api::orderman::Response response;
|
|
response.set_order_key(detach.order_key());
|
|
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
|
|
r(response);
|
|
} else {
|
|
me->in_flight_order_ops[++me->current_req_op_id] = r;
|
|
me->service_bridge->detach_order(me->current_req_op_id, detach);
|
|
}
|
|
}
|
|
|
|
void OrderServiceServer::get_registered_dynamic_order_types(
|
|
const api::VoidMessage &,
|
|
const api::Response<api::orderman::RegisteredDynamicOrderTypes> &r) {
|
|
api::orderman::RegisteredDynamicOrderTypes response;
|
|
|
|
auto all_orders = me->dyn_order_registry.get_all_dynamic_order_types();
|
|
for (auto const &desc : all_orders)
|
|
*response.add_dynamic_order_desc() = desc;
|
|
r(response);
|
|
}
|
|
|
|
void OrderServiceServer::on_order_op_result(
|
|
OrderOpResultEvent const &op_result) {
|
|
auto it = me->in_flight_order_ops.find(op_result.req_id);
|
|
bts_assert(it != me->in_flight_order_ops.end());
|
|
auto &responder = it->second;
|
|
api::orderman::Response r;
|
|
r.set_order_key(op_result.ok.value());
|
|
r.set_error_code(op_result.result.error);
|
|
r.set_msg(op_result.result.error_msg);
|
|
responder(r);
|
|
me->in_flight_order_ops.erase(it);
|
|
}
|
|
|
|
} // namespace orderman
|
|
} // namespace bts
|