petefisher78/spark
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added .clang-format, applied it to a couple of files.

Browse Source
This commit is contained in:
Pedro Pinto committed 2017-12-24 13:00:05 -05:00
1 parent a68ff5342c
commit b76bec20ad
3 files changed
+165 -139
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.clang-format
+1
View File
@@ -0,0 +1 @@
IndentWidth: 3
src/server/sparkd/main.cpp
+17 -17
View File
@@ -15,23 +15,23 @@
#include "server/sparkd/cmds/TestOrderman.h" #include "server/sparkd/cmds/TestOrderman.h"
int main(int argc, char **argv) { int main(int argc, char **argv) {
using namespace bts; using namespace bts;
os::set_thread_name("main"); os::set_thread_name("main");
utils::init_spark_lib(); utils::init_spark_lib();
BEGIN_CMD; BEGIN_CMD;
REGISTER(Test); REGISTER(Test);
REGISTER(Help); REGISTER(Help);
REGISTER(Run); REGISTER(Run);
REGISTER(Md); REGISTER(Md);
REGISTER(TestOrderman); REGISTER(TestOrderman);
REGISTER(TestCeloxica); REGISTER(TestCeloxica);
REGISTER(ConfigCheck); REGISTER(ConfigCheck);
REGISTER(RunSim); REGISTER(RunSim);
REGISTER(Benchmark); REGISTER(Benchmark);
END_CMD; END_CMD;
utils::finit_spark_lib(); utils::finit_spark_lib();
std::string s; std::string s;
return 0; return 0;
} }
src/server/sparkd/orderman2/actors/order_manager/OrderManager.cpp
+147 -122
View File
@@ -2,15 +2,14 @@
#include "OrderManager.h" #include "OrderManager.h"
#include <unordered_map>
#include "server/sparkd/jetfix/common/OutFIX.h" #include "server/sparkd/jetfix/common/OutFIX.h"
#include "server/sparkd/orderman2/actors/order_manager/OrderServices.h"
#include "server/sparkd/orderman2/events/OrderManagerReadyEvent.h"
#include "server/sparkd/orderman2/order_factory/OrderFactory.h" #include "server/sparkd/orderman2/order_factory/OrderFactory.h"
#include "server/sparkd/orderman2/orders/Order.h" #include "server/sparkd/orderman2/orders/Order.h"
#include "server/sparkd/securities/ISecurityDB.h" #include "server/sparkd/securities/ISecurityDB.h"
#include "server/sparkd/orderman2/actors/order_manager/OrderServices.h"
#include "server/sparkd/orderman2/events/OrderManagerReadyEvent.h" #include <unordered_map>
namespace bts { namespace bts {
namespace orderman2 { namespace orderman2 {
@@ -18,33 +17,27 @@ namespace orderman2 {
using namespace mercury; using namespace mercury;
const auto fix_buffer_pool_size = 10000; const auto fix_buffer_pool_size = 10000;
enum class OrderManager::State {WAIT_SEC, WAIT_ORDERS, READY } ; enum class OrderManager::State { WAIT_SEC, WAIT_ORDERS, READY };
struct OrderManager::OrderInfo { struct OrderManager::OrderInfo {
OrderInfo (Order &o, OrderServicesData &data) OrderInfo(Order &o, OrderServicesData &data)
: order(&o), : order(&o), order_services(std::make_unique<OrderServices>(o, data)) {}
order_services(std::make_unique<OrderServices> (o,data))
{}
OrderInfo() = default; OrderInfo() = default;
OrderInfo(OrderInfo &&src) = default; OrderInfo(OrderInfo &&src) = default;
OrderInfo& operator=(OrderInfo &&src) = default; OrderInfo &operator=(OrderInfo &&src) = default;
Order *order; Order *order;
std::unique_ptr<OrderServices> order_services; std::unique_ptr<OrderServices> order_services;
}; };
struct OrderManager::PImpl struct OrderManager::PImpl {
{ PImpl(config::JetFIX::Connection const &cfg, Actor &owner)
PImpl(config::JetFIX::Connection const &cfg, Actor&owner) : : state(State::WAIT_SEC), cfg(cfg), gw_name(GatewayName(cfg.name())),
state (State::WAIT_SEC), services_data(cfg, owner) {}
cfg(cfg),
gw_name(GatewayName(cfg.name())),
services_data(cfg,owner)
{}
State state; State state;
std::unordered_map<OrderKey,OrderInfo> orders; std::unordered_map<OrderKey, OrderInfo> orders;
mercury::Actor::SubHandle stored_order_sub; mercury::Actor::SubHandle stored_order_sub;
const config::JetFIX::Connection cfg; const config::JetFIX::Connection cfg;
const GatewayName gw_name; const GatewayName gw_name;
@@ -52,48 +45,50 @@ struct OrderManager::PImpl
mercury::Stream<OrderManagerReadyEvent> pub_ready; mercury::Stream<OrderManagerReadyEvent> pub_ready;
}; };
OrderManager::OrderManager (config::JetFIX::Connection const&cfg) : me (new PImpl(cfg,*this)) {} OrderManager::OrderManager(config::JetFIX::Connection const &cfg)
OrderManager::~OrderManager () {} : me(new PImpl(cfg, *this)) {}
OrderManager::~OrderManager() {}
void OrderManager::init() void OrderManager::init() {
{ subscribe<NewSecuritiesEvent>(
subscribe<NewSecuritiesEvent>([this] (auto const&sec_update) {this->on_securities(sec_update);}); [this](auto const &sec_update) { this->on_securities(sec_update); });
} }
void OrderManager::on_securities(const mercury::Update<NewSecuritiesEvent> &update) void OrderManager::on_securities(
{ const mercury::Update<NewSecuritiesEvent> &update) {
if (update.events) { if (update.events) {
// update our security db // update our security db
for (auto ev : *(update.events)) { for (auto ev : *(update.events)) {
auto sec_ptr = ev.security; auto sec_ptr = ev.security;
me->services_data.sec_db.securities[SecurityKey(sec_ptr->id())] = sec_ptr; me->services_data.sec_db.securities[SecurityKey(sec_ptr->id())] =
sec_ptr;
} }
if (me->state == State::WAIT_SEC) { if (me->state == State::WAIT_SEC) {
// obtain recovered orders // obtain recovered orders
me->state = State::WAIT_ORDERS; me->state = State::WAIT_ORDERS;
me->stored_order_sub = me->stored_order_sub = subscribe<StoredOrderEvent>(
subscribe<StoredOrderEvent> (me->gw_name, me->gw_name,
[this] (auto const&orders) [this](auto const &orders) { this->on_order_recovery(orders); });
{this->on_order_recovery(orders);});
} }
} }
} }
void OrderManager::on_order_recovery(const mercury::Update<StoredOrderEvent> &update) void OrderManager::on_order_recovery(
{ const mercury::Update<StoredOrderEvent> &update) {
if (update.events) if (update.events)
if (me->state == State::WAIT_ORDERS) { if (me->state == State::WAIT_ORDERS) {
// Got the first batch of persisted orders which has to match // Got the first batch of persisted orders which has to match
// the orders stored on disk (since we are the only ones // the orders stored on disk (since we are the only ones
// producing new orders) // producing new orders)
for (auto const &ev : *(update.events)) { for (auto const &ev : *(update.events)) {
auto &order = me->services_data.order_factory.recover(*ev.order_record); auto &order =
auto const&ostate = ev.order_record->state(); me->services_data.order_factory.recover(*ev.order_record);
OrderInfo oi {order,me->services_data}; auto const &ostate = ev.order_record->state();
OrderInfo oi{order, me->services_data};
auto serv_ptr = oi.order_services.get(); auto serv_ptr = oi.order_services.get();
me->orders[order.key()] = std::move(oi); me->orders[order.key()] = std::move(oi);
order.on_activate(Price(ostate.price()), order.on_activate(Price(ostate.price()),
Quantity(ostate.quantity()), *serv_ptr); Quantity(ostate.quantity()), *serv_ptr);
} }
@@ -115,34 +110,45 @@ void OrderManager::on_order_recovery(const mercury::Update<StoredOrderEvent> &up
// it. In that case we will do our best to recover the order // it. In that case we will do our best to recover the order
// from the execution report. That is the only purpose of // from the execution report. That is the only purpose of
// this subscription. // this subscription.
subscribe<NewExecutionReportEvent> (me->gw_name, subscribe<NewExecutionReportEvent>(
[this] (auto const&ev) {this->on_execution_report(ev);}); me->gw_name,
[this](auto const &ev) { this->on_execution_report(ev); });
// we need to know the gateway status because some of our // we need to know the gateway status because some of our
// orders need to know it at activation time. The // orders need to know it at activation time. The
// ContentSubscription mechanism does not provide a synopsis // ContentSubscription mechanism does not provide a synopsis
// so we cache the last known state of the gateway ourselves // so we cache the last known state of the gateway ourselves
// and offer it to the orders. // and offer it to the orders.
subscribe<GatewayStatusChangeEvent> (me->gw_name, subscribe<GatewayStatusChangeEvent>(
[this] (auto const&ev) {this->on_gateway_state(ev);}); me->gw_name,
[this](auto const &ev) { this->on_gateway_state(ev); });
// Order creation and maintenance // Order creation and maintenance
subscribe<ActivateOrderEvent>(me->gw_name, [this] (auto const&ev) {this->on_activate_order(ev);}); subscribe<ActivateOrderEvent>(me->gw_name, [this](auto const &ev) {
subscribe<UpdateOrderEvent>(me->gw_name, [this] (auto const&ev) {this->on_update_order(ev);}); this->on_activate_order(ev);
subscribe<CancelOrderEvent>(me->gw_name, [this] (auto const&ev) {this->on_cancel_order(ev);}); });
subscribe<CrackOrderEvent>(me->gw_name, [this] (auto const&ev) {this->on_crack_order(ev);}); subscribe<UpdateOrderEvent>(me->gw_name, [this](auto const &ev) {
this->on_update_order(ev);
});
subscribe<CancelOrderEvent>(me->gw_name, [this](auto const &ev) {
this->on_cancel_order(ev);
});
subscribe<CrackOrderEvent>(
me->gw_name, [this](auto const &ev) { this->on_crack_order(ev); });
// clock // clock
subscribe<mercury::TimerEvent>([this] (auto const &ev) {this->on_timer(ev);}); subscribe<mercury::TimerEvent>(
[this](auto const &ev) { this->on_timer(ev); });
// Allocating FIX buffers is relatively expensive so we try // Allocating FIX buffers is relatively expensive so we try
// to keep a pool of pre-allocated buffers ready. To // to keep a pool of pre-allocated buffers ready. To
// replinish the pool we issue CreateFIXBufferRequestEvent // replinish the pool we issue CreateFIXBufferRequestEvent
// and listen for replies // and listen for replies
while (free_buffers().size() != fix_buffer_pool_size) while (free_buffers().size() != fix_buffer_pool_size)
free_buffers().push_back(std::make_shared<jetfix::OutFIX>(me->cfg)); free_buffers().push_back(std::make_shared<jetfix::OutFIX>(me->cfg));
subscribe<CreateFIXBufferReplyEvent>(me->gw_name, subscribe<CreateFIXBufferReplyEvent>(
[this] (auto const &ev) {this->on_new_fix_buffer(ev);}); me->gw_name,
[this](auto const &ev) { this->on_new_fix_buffer(ev); });
// Subscription setup/teardown is a relatively heavy // Subscription setup/teardown is a relatively heavy
// operation. For that reason we use coarse-grained topics to // operation. For that reason we use coarse-grained topics to
@@ -150,142 +156,161 @@ void OrderManager::on_order_recovery(const mercury::Update<StoredOrderEvent> &up
// can impose more fine-grained and light weight filtering on // can impose more fine-grained and light weight filtering on
// top of those topic subscriptions through the // top of those topic subscriptions through the
// ContentSubscription helper class. // ContentSubscription helper class.
me->services_data.md_subs = ContentSubscription<BookUpdateEvent> (*this); me->services_data.md_subs =
me->services_data.child_order_subs = ContentSubscription<OrderStateUpdateEvent,ParentKey> (*this); ContentSubscription<BookUpdateEvent>(*this);
me->services_data.md_subs = ContentSubscription<BookUpdateEvent> (*this); me->services_data.child_order_subs =
me->services_data.exe_report_subs = ContentSubscription<NewExecutionReportEvent> (*this,me->gw_name); ContentSubscription<OrderStateUpdateEvent, ParentKey>(*this);
me->services_data.cancel_reject_subs = ContentSubscription<CancelRejectEvent> (*this,me->gw_name); me->services_data.md_subs =
me->services_data.gateway_status_subs = ContentSubscription<GatewayStatusChangeEvent,NoKey> (*this); ContentSubscription<BookUpdateEvent>(*this);
me->services_data.time_subs = ContentSubscription<TimerEvent,NoKey> (*this); me->services_data.exe_report_subs =
me->services_data.order_creation_subs = ContentSubscription<CreateOrderReplyEvent> (*this,me->gw_name); ContentSubscription<NewExecutionReportEvent>(*this, me->gw_name);
me->services_data.cancel_reject_subs =
ContentSubscription<CancelRejectEvent>(*this, me->gw_name);
me->services_data.gateway_status_subs =
ContentSubscription<GatewayStatusChangeEvent, NoKey>(*this);
me->services_data.time_subs =
ContentSubscription<TimerEvent, NoKey>(*this);
me->services_data.order_creation_subs =
ContentSubscription<CreateOrderReplyEvent>(*this, me->gw_name);
// Outputs // Outputs
me->services_data.order_state_pub = make_stream<OrderStateUpdateEvent>(); me->services_data.order_state_pub =
me->services_data.out_fix_pub = make_stream<OutFIXMsgEvent>(me->gw_name); make_stream<OrderStateUpdateEvent>();
me->services_data.activate_order_pub = make_stream<ActivateOrderEvent>(me->gw_name); me->services_data.out_fix_pub =
me->services_data.update_order_pub = make_stream<UpdateOrderEvent>(me->gw_name); make_stream<OutFIXMsgEvent>(me->gw_name);
me->services_data.cancel_order_pub = make_stream<CancelOrderEvent>(me->gw_name); me->services_data.activate_order_pub =
me->services_data.crack_order_pub = make_stream<CrackOrderEvent>(me->gw_name); make_stream<ActivateOrderEvent>(me->gw_name);
me->services_data.create_order_req_pub = make_stream<CreateOrderRequestEvent>(me->gw_name); me->services_data.update_order_pub =
me->services_data.create_fix_req_pub = make_stream<CreateFIXBufferRequestEvent>(me->gw_name); make_stream<UpdateOrderEvent>(me->gw_name);
me->services_data.cancel_order_pub =
make_stream<CancelOrderEvent>(me->gw_name);
me->services_data.crack_order_pub =
make_stream<CrackOrderEvent>(me->gw_name);
me->services_data.create_order_req_pub =
make_stream<CreateOrderRequestEvent>(me->gw_name);
me->services_data.create_fix_req_pub =
make_stream<CreateFIXBufferRequestEvent>(me->gw_name);
me->services_data.md_request_pub = make_stream<MDRequestEvent>(); me->services_data.md_request_pub = make_stream<MDRequestEvent>();
me->services_data.report_order_state_pub = make_stream<OrderStateUpdateEvent>(); me->services_data.report_order_state_pub =
me->pub_ready = make_stream<OrderManagerReadyEvent> (me->gw_name); make_stream<OrderStateUpdateEvent>();
me->pub_ready = make_stream<OrderManagerReadyEvent>(me->gw_name);
// signal that we are ready // signal that we are ready
publish(me->pub_ready); publish(me->pub_ready);
} }
} }
void OrderManager::on_execution_report(const mercury::Update<NewExecutionReportEvent> &m) void OrderManager::on_execution_report(
{ const mercury::Update<NewExecutionReportEvent> &m) {
if (m.events) if (m.events)
for (auto const &report: *m.events) { for (auto const &report : *m.events) {
auto k = report.data->order_key; auto k = report.data->order_key;
auto it = me->orders.find(k); auto it = me->orders.find(k);
if (it == me->orders.end()) { if (it == me->orders.end()) {
LOG(OrderMan,Error, LOG(OrderMan, Error,
"Received an execution report for an unknown order with key {}. " "Received an execution report for an unknown order with key "
"Order state will be reconstructed from this report but some information " "{}. "
"Order state will be reconstructed from this report but some "
"information "
"will be lost, including CustomerOrFirm,CTI code and " "will be lost, including CustomerOrFirm,CTI code and "
"parent order id", k.value()); "parent order id",
auto order = &me->services_data.order_factory.recreate_from_exe_report(*report.data); k.value());
OrderInfo oi {*order,me->services_data}; auto order =
&me->services_data.order_factory.recreate_from_exe_report(
*report.data);
OrderInfo oi{*order, me->services_data};
auto serv_ptr = oi.order_services.get(); auto serv_ptr = oi.order_services.get();
me->orders[order->key()] = std::move(oi); me->orders[order->key()] = std::move(oi);
order->on_activate(report.data->price,report.data->quantity,*serv_ptr); order->on_activate(report.data->price, report.data->quantity,
*serv_ptr);
} }
} }
} }
void OrderManager::on_activate_order(const mercury::Update<ActivateOrderEvent> &m) void OrderManager::on_activate_order(
{ const mercury::Update<ActivateOrderEvent> &m) {
if (m.events) if (m.events)
for (auto const &ev: *m.events) { for (auto const &ev : *m.events) {
auto order = ev.order; auto order = ev.order;
auto k = order->key(); auto k = order->key();
bts_assert(me->orders.find(k) != me->orders.end()); bts_assert(me->orders.find(k) != me->orders.end());
OrderInfo oi {*order,me->services_data}; OrderInfo oi{*order, me->services_data};
auto serv_ptr = oi.order_services.get(); auto serv_ptr = oi.order_services.get();
me->orders[k] = std::move(oi); me->orders[k] = std::move(oi);
order->on_activate(ev.price,ev.quantity,*serv_ptr); order->on_activate(ev.price, ev.quantity, *serv_ptr);
} }
} }
void OrderManager::on_update_order(const mercury::Update<UpdateOrderEvent> &m) void OrderManager::on_update_order(const mercury::Update<UpdateOrderEvent> &m) {
{
if (m.events) if (m.events)
for (auto const &ev: *m.events) { for (auto const &ev : *m.events) {
auto k = ev.key(); auto k = ev.key();
auto it = me->orders.find(k); auto it = me->orders.find(k);
if (it != me->orders.end()) if (it != me->orders.end())
it->second.order->on_update(ev.price,ev.quantity); it->second.order->on_update(ev.price, ev.quantity);
else { else {
LOG(OrderMan,Error, LOG(OrderMan, Error,
"Received an update request for an unknown order {}. " "Received an update request for an unknown order {}. "
"This should not happen and likely indicates a serious bug. ", "This should not happen and likely indicates a serious bug. ",
ev.key().value()); ev.key().value());
} }
} }
} }
void OrderManager::on_cancel_order(const mercury::Update<CancelOrderEvent> &m) void OrderManager::on_cancel_order(const mercury::Update<CancelOrderEvent> &m) {
{
if (m.events) if (m.events)
for (auto const &ev: *m.events) { for (auto const &ev : *m.events) {
auto k = ev.key(); auto k = ev.key();
auto it = me->orders.find(k); auto it = me->orders.find(k);
if (it != me->orders.end()) if (it != me->orders.end())
it->second.order->on_cancel(); it->second.order->on_cancel();
else { else {
LOG(OrderMan,Error, LOG(OrderMan, Error,
"Received a cancel request for an unknown order {}. " "Received a cancel request for an unknown order {}. "
"This should not happen and likely indicates a serious bug. ", "This should not happen and likely indicates a serious bug. ",
ev.key().value()); ev.key().value());
} }
} }
} }
void OrderManager::on_crack_order(const mercury::Update<CrackOrderEvent> &m) void OrderManager::on_crack_order(const mercury::Update<CrackOrderEvent> &m) {
{
if (m.events) if (m.events)
for (auto const &ev: *m.events) { for (auto const &ev : *m.events) {
auto k = ev.key(); auto k = ev.key();
auto it = me->orders.find(k); auto it = me->orders.find(k);
if (it != me->orders.end()) if (it != me->orders.end())
it->second.order->on_crack(); it->second.order->on_crack();
else { else {
LOG(OrderMan,Error, LOG(OrderMan, Error,
"Received a crack request for an unknown order {}. " "Received a crack request for an unknown order {}. "
"This should not happen and likely indicates a serious bug. ", "This should not happen and likely indicates a serious bug. ",
ev.key().value()); ev.key().value());
} }
} }
} }
void OrderManager::on_gateway_state(const mercury::Update<GatewayStatusChangeEvent> &m) void OrderManager::on_gateway_state(
{ const mercury::Update<GatewayStatusChangeEvent> &m) {
if (m.events) if (m.events)
for (auto ev : *m.events) for (auto ev : *m.events)
me->services_data.gw_connected = ev.is_connected; me->services_data.gw_connected = ev.is_connected;
} }
void OrderManager::on_new_fix_buffer(const mercury::Update<CreateFIXBufferReplyEvent> &m) void OrderManager::on_new_fix_buffer(
{ const mercury::Update<CreateFIXBufferReplyEvent> &m) {
if (m.events) if (m.events)
for (auto ev : *m.events) for (auto ev : *m.events)
free_buffers().insert(free_buffers().end(), free_buffers().insert(free_buffers().end(), ev.buffers.begin(),
ev.buffers.begin(), ev.buffers.end()); ev.buffers.end());
} }
void OrderManager::on_timer(const mercury::Update<mercury::TimerEvent> &m) void OrderManager::on_timer(const mercury::Update<mercury::TimerEvent> &m) {
{
if (m.events) if (m.events)
for (auto ev : *m.events) for (auto ev : *m.events)
me->services_data.last_timer_ev = ev; me->services_data.last_timer_ev = ev;
} }
std::vector<jetfix::OutFIXPtr> &OrderManager::free_buffers() {
std::vector<jetfix::OutFIXPtr> &OrderManager::free_buffers() return me->services_data.free_buffers;
{return me->services_data.free_buffers;} }
}
}} }