petefisher78/spark
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
484ad656f9dde923d286002ec71439877ca94d33
spark/src/lib/md/relay/QueryServer.cpp
T

472 lines
18 KiB
C++
Raw Blame History

#ifndef NO_CPP0X
#include "QueryServer.h"
#include "lib/config/config.h"
#include "lib/utils/Dispatcher.h"
#include "lib/utils/TimeStamp.h"
#include "lib/utils/helper_macros.h"
#include "../MDConsumer.h"
#include "../PlatformLoader.h"
#include "lib/logging/logging.h"
#include "lib/config/protos/system.pb.h"
#include "lib/md/protos/co2.pb.h"
#include "lib/api/apicore.pb.h"
#include "lib/api/services/MarketDataRelay.server.pb.h"
#include "lib/api/public/ServiceContext.h"
#ifndef _WIN32
#include "server/sparkd/metrics/metrics.h"
#endif
#include <vector>
#include <list>
#include <memory>
#include <set>
#include <google/protobuf/arena.h>
#include <boost/container/flat_map.hpp>
#include <tbb/concurrent_vector.h>
using namespace bts::co2;
using std::unique_ptr;
using std::shared_ptr;
namespace bts {
namespace md {
/*
* NORMAL OPERATION:
* Client sends time=0 to server
* Server sends complete snapshot back (time=502)
* client sends time=502 to server
* Server sends any changes that have happened after time=502 and any securities discovered since time=502
*
* ERROR RECOVERY:
* Client sends time=0 to server
* Server sends complete snapshot back (time=502)
* Server detects heartbeat failure and recovers. All timestamps > 502
* Client sends time=502 to server
* Server sends complete snapshot back (time=605)
* - Since the server shared memory has been recreated the client will automatically overwrite all stale data.
* - If a security has been deleted the client will leave it frozen (OK)
*
* We scan for bad heartbeats on a different time
* If a bad heartbeat is detected we try to construct a new Consumer
* When a good consumer is created we std::move the old consumer and handles to delayed queue for deletion
* This way we don't have to lock anything during the recovery process because all pointers will remain valid
* for the duration of existing calls.
*
*/
struct DatedHandle {
TimeStamp discovered;
MDConsumerHandle handle;
};
struct MDConsumerGroup {
// probably just have to send the consumer in from outside
MDConsumerGroup(const UserSymbol& symbol, const MDConsumer& consumer, metrics::MetricsContainer& stats) :
symbol(symbol),
platform(platform_for_exchange(symbol.exchange())),
channel(consumer.get_channel(symbol)),
handles()
{
set_consumer(consumer);
// create and publish a stat for this group
#ifndef _WIN32
if (symbol.type() != SecurityType::SECURITYTYPE_SYNTHETIC) {
std::vector<metrics::Label> labels;
labels.push_back({ "symbol", symbol.symbol() });
labels.push_back({ "type", SecurityType_Name(symbol.type()).substr(13) /*SECURITYTYPE_*/ });
labels.push_back({ "exch", Venue_Name(symbol.exchange()).substr(6) /*VENUE_*/ });
stats.publish(
std::make_shared<metrics::LazyCounterMetric>("spark_md_book_updates", [=]() { return std::max(channel.bbo_count(), channel.depth_count()); }, labels));
stats.publish(
std::make_shared<metrics::LazyCounterMetric>("spark_md_security_count", [=]() { return channel.security_count(); }, labels));
stats.publish(
std::make_shared<metrics::LazyCounterMetric>("spark_md_trade_count", [=]() { return channel.trade_count(); }, labels));
}
#endif
}
MDConsumerGroup(const MDConsumerGroup& other) = delete;
// recovery has occured and we need to clear our our collection of handles and wait for new ones
void set_consumer(const MDConsumer& consumer) {
// create a new set of handles and swap with the old.
// If the new consumer is bad we still have to swap our handles because the consumer that owns our handles will be scheduled for deletion
auto temphandles = std::make_shared<tbb::concurrent_vector<DatedHandle>>();
handles.swap(temphandles);
channel = consumer.get_channel(symbol);
}
void add_security(MDConsumerHandle& handle) {
bts_assert(handles);
bts_assert_d(handle.is_initialized());
bts_assert(handle.get_usersymbol() == symbol);
DatedHandle h = { TimeStamp::now(), handle };
handles->push_back(h);
}
UserSymbol symbol;
Platform platform;
MDConsumerChannel channel;
// Can add to handles without invalidating readers (no lock needed)
// When we recover we swap a new vector into this unique_ptr and delayed-delete
shared_ptr<tbb::concurrent_vector<DatedHandle>> handles;
};
void convert(PriceSize ps, PriceSizeMessage* pps) {
pps->set_mantissa(ps.mantissa());
pps->set_exponent(ps.exponent());
pps->set_size(ps.size());
}
void set_price(PriceSize ps, co2::TradeMessage* ptrade) {
ptrade->set_mantissa(ps.mantissa());
ptrade->set_exponent(ps.exponent());
ptrade->set_size(ps.size());
}
void copy_misc(MiscData misc, co2::MiscMessage* pmisc) {
pmisc->set_settlement_price(misc.settlement_price);
pmisc->set_settlement_date(misc.settlement_date.to_short());
pmisc->set_previous_settlement_price(misc.previous_settlement_price);
pmisc->set_previous_settlement_date(misc.previous_settlement_date.to_short());
pmisc->set_open_price(misc.open_price);
pmisc->set_open_date(misc.open_date.to_short());
pmisc->set_close_price(misc.close_price);
pmisc->set_close_date(misc.close_date.to_short());
//pmisc->set_fixing_price(misc.fixing_price);
pmisc->set_session_high(misc.session_high);
pmisc->set_session_low(misc.session_low);
//pmisc->set_open_interest(misc.open_interest);
//pmisc->set_session_high_bid(misc.session_high_bid);
//pmisc->set_session_low_offer(misc.session_low_offer);
}
// Handles market data queries and responds
class QueryServerImpl : public IQueryServer {
MDConsumer _consumer;
const std::set<UserSymbol> _symbols;
const std::shared_ptr<api::ISyntheticSecurityService> _ss;
boost::container::flat_map<UserSymbol,shared_ptr<MDConsumerGroup>> _groups;
public:
QueryServerImpl(const std::set<UserSymbol>& symbols, std::shared_ptr<api::ISyntheticSecurityService>& ss, metrics::MetricsContainer& stats) :
_consumer(symbols, ss),
_symbols(symbols),
_ss(ss),
_groups()
{
for (const auto& sym : symbols) {
_groups.emplace(std::make_pair(sym, std::make_unique<MDConsumerGroup>(sym, _consumer, stats)));
}
_groups.shrink_to_fit();
subscribe_securities();
}
void subscribe_securities() {
auto id = _consumer.id();
_consumer.on_security([this,id](const Security&, MDConsumerHandle handle) {
if (_consumer.id() == id) {
auto iter = _groups.find(handle.get_usersymbol());
bts_assert(iter != _groups.end());
const auto& pgroup = iter->second;
bts_assert(pgroup);
pgroup->add_security(handle);
}
});
}
// not-reentrant
void scan_securities() {
if (_consumer.all_heartbeats_ok()) {
TimeStamp start = TimeStamp::now();
size_t count = _consumer.scan_securities(false);
auto delta = (TimeStamp::now() - start);
if (delta.milliseconds() > 10) {
LOG(md,Diag,"Scanned handles: {}ms found: {} securities. " ,
delta.milliseconds(),count);
}
}
}
// not-reentrant
void check_heartbeats(utils::Dispatcher& dispatcher) {
if (!_consumer.all_heartbeats_ok()) {
MDConsumer temp(_symbols, _ss);
if (temp.all_heartbeats_ok()) {
_consumer.swap(temp);
LOG(md,Info, "Recovering from bad MD. Old consumer {} new consumer {}", _consumer.id(), temp.id());
for (auto& pair : _groups) {
pair.second->set_consumer(_consumer);
}
subscribe_securities();
dispatcher.post([temp]() { LOG(md,Info,"QServer destroying consumer {}", temp.id()); }, 60000);
}
}
}
// the hot part
void fill(const SecurityChannel& chan, MDConsumerGroup& group, ChannelReply* reply) const {
bts_assert(reply);
reply->mutable_channel()->CopyFrom(chan);
reply->set_platform(group.platform);
reply->mutable_channel()->set_time(TimeStamp::now().to_uint64());
auto flags = (MDRequest::Flags)chan.flags();
// Get a copy of handles. If recovery happens during this call
// we can safely continue using this pointer for 60 seconds
auto phandles = group.handles;
if (!phandles) return;
ChannelUpdate temp_update;
for (const auto& instrument : *phandles) {
bool snapshot = false;
MDConsumerHandle handle = instrument.handle;
temp_update.Clear();
ChannelUpdate* pupd = &temp_update;
pupd->set_key(instrument.handle.key());
bts_assert_d(handle.is_initialized());
if (!handle.is_initialized()) continue;
// If the security was just discovered try to ensure all data is sent
if (instrument.discovered > TimeStamp(chan.time())) {
pupd->mutable_security()->CopyFrom(handle.get_normalized_security());
snapshot = true;
}
if (HASFLAG(flags, MDRequest::BBO) || HASFLAG(flags, MDRequest::DEPTH)) {
std::array<PriceSize, COCONUT_MAX_DEPTH> bids, asks;
TimeStamp time(snapshot ? 0 : chan.time());
size_t copy_depth = HASFLAG(flags, MDRequest::DEPTH) ? COCONUT_MAX_DEPTH : 1;
auto result = handle.get_depth(time, copy_depth, bids.data(), asks.data());
if (result == HandleReadResult::SUCCESS) {
auto* p = pupd->mutable_book();
for (size_t x = 0; x < COCONUT_MAX_DEPTH; ++x) {
PriceSize b = bids[x];
if (b != PriceSize::NO_MARKET) {
convert(b, p->add_bids());
}
PriceSize a = asks[x];
if (a != PriceSize::NO_MARKET) {
convert(a, p->add_asks());
}
}
}
else if (result == HandleReadResult::BUSY) {
LOG(md,Info, "Busy read on handle {}. Retry on next request", instrument.handle.key());
// If we get a busy reading BBO (which happens 1-2x per day for a single relay client)
// We move the timestamp back so the next request will try to read this value again.
// ALTERNATIVE: We could spin until the value was not busy but it could be that the writer
// was context switched out and we will have to wait multiple ms to unblock the request.
// This fix assumes it is better to send the update now with 1 stale value than to block the call
// and allow all the values to be stale until this one can be read.
reply->mutable_channel()->set_time(chan.time());
}
}
if (HASFLAG(flags, MDRequest::LAST_TRADE)) {
TimeStamp time(snapshot ? 0 : chan.time());
PriceSize price;
datetime::LLDateTime trade_time;
size_t volume;
if (handle.get_last_trade(time, price, trade_time, &volume)) {
auto* trade = pupd->mutable_last_trade();
trade->set_key(handle.key());
trade->set_time(trade_time);
trade->set_total_volume(volume);
set_price(price, trade);
}
}
if (HASFLAG(flags, MDRequest::CUSTOMER_BBO)) {
TimeStamp time(snapshot ? 0 : chan.time());
PriceSize bid, ask;
if (handle.get_customer_bbo(time, bid, ask)) {
auto* p = pupd->mutable_customer_bbo();
convert(bid, p->mutable_bid());
convert(ask, p->mutable_ask());
}
}
if (HASFLAG(flags, MDRequest::MISC)) {
TimeStamp time(snapshot ? 0 : chan.time());
if (handle.misc_timestamp() > time) {
copy_misc(handle.misc(), pupd->mutable_misc());
}
}
if (pupd->has_book() ||
pupd->has_last_trade() ||
pupd->has_customer_bbo() ||
pupd->has_misc() ||
pupd->has_security()) {
reply->add_updates()->CopyFrom(*pupd);
}
}
reply->mutable_channel()->set_segment_id(group.channel.create_time().to_uint64());
bool is_same_segment = (chan.segment_id() == group.channel.create_time().to_uint64());
if (HASFLAG(flags, MDRequest::TRADE_LOG)) {
int64_t request_cursor = is_same_segment ? chan.trade_cursor() : 0;
auto cursor = group.channel.read_trades(*reply->mutable_trade_log(), request_cursor);
if (cursor != chan.trade_cursor()) {
reply->mutable_channel()->set_trade_cursor(cursor);
}
}
if (HASFLAG(flags, MDRequest::RFQ)) {
int64_t request_cursor = is_same_segment ? chan.rfq_cursor() : 0;
auto cursor = group.channel.read_rfqs(*reply->mutable_rfq_log(), request_cursor);
if (cursor != chan.rfq_cursor()) {
reply->mutable_channel()->set_rfq_cursor(cursor);
}
}
}
void query(const MDRequest& request, MDReply* reply) const {
bts_assert(reply != nullptr);
for (auto& ch : request.channels()) {
auto groupiter = _groups.find(UserSymbol(ch.symbol(), ch.exchange(), ch.type()));
if (groupiter != _groups.end()) {
auto& group = groupiter->second;
fill(ch, *group, reply->add_channels());
}
}
}
void get_names(ExpirationNameListReply* reply) const {
bts_assert(reply != nullptr);
Security security;
for (auto& groupiter : _groups) {
const auto& group = groupiter.second;
ExpirationNameList* psec = reply->add_securities();
psec->mutable_product()->CopyFrom(group->symbol.to_product());
for(const auto& h : *(group->handles)) {
security.CopyFrom(h.handle.get_normalized_security());
auto x = psec->mutable_names();
if (security.ExchangeData_case() == Security::kGlbx) {
google::protobuf::MapPair<uint32_t,std::string> v(h.handle.key(), security.glbx().security_desc());
x->insert(v);
}
}
}
}
};
struct MDRelayServer : bts::co2::MarketDataRelayBase {
const QueryServerImpl& _db;
utils::Dispatcher& _threadpool;
#ifndef _WIN32
std::shared_ptr<metrics::HistogramMetric<5>> _time,_size;
#endif
MDRelayServer(const QueryServerImpl& db, utils::Dispatcher& threadpool, bts::api::ServiceContext ctx, metrics::MetricsContainer& stats) :
bts::co2::MarketDataRelayBase(ctx),
_db(db), _threadpool(threadpool)
#ifndef _WIN32
, _time(std::make_shared<metrics::HistogramMetric<5>>("spark_qserver_response_time_ms", std::array<double,5>({ 10.0, 50.0, 100.0, 500.0, 1000.0 }))),
_size(std::make_shared<metrics::HistogramMetric<5>>("spark_qserver_size_bytes", std::array<double,5>({ 1024 /*kb*/, 1024*1024 /*mb*/, (1024*1024)*4, (1024*1024)*16, (1024*1024)*32 })))
{
stats.publish(_time);
stats.publish(_size);
}
#else
{}
#endif
virtual void requestmd(const MDRequest & request, const std::function<void(const MDReply&)>& reply) override {
TimeStamp start = TimeStamp::now();
_threadpool.post([=]() {
google::protobuf::Arena arena;
MDReply* preply = google::protobuf::Arena::CreateMessage<MDReply>(&arena);
_db.query(request, preply);
reply(*preply);
auto elapsed = (TimeStamp::now() - start);
#ifndef _WIN32
_time->observe(elapsed.milliseconds());
_size->observe(preply->ByteSize());
#endif
});
}
virtual void request_products(const api::VoidMessage&, const std::function<void(const ProductListReply&)>& reply) override {
ProductListReply list;
std::set<UserSymbol> syms;
get_system_md_symbols(syms);
for(const auto& i : syms)
{
auto prod = list.add_products();
prod->set_exchange(i.exchange());
prod->set_type(i.type());
prod->set_symbol(i.symbol());
}
reply(list);
}
virtual void request_names(const api::VoidMessage&, const std::function<void(const ExpirationNameListReply&)>& reply) override {
_threadpool.post([=]() {
google::protobuf::Arena arena;
ExpirationNameListReply* preply = google::protobuf::Arena::CreateMessage<ExpirationNameListReply>(&arena);
_db.get_names(preply);
reply(*preply);
});
}
};
struct QueryServer {
QueryServer(std::set<UserSymbol> symbols,
utils::Dispatcher& threadpool,
const bts::api::ServiceContext& ctx,
std::shared_ptr<api::ISyntheticSecurityService>& ss,
metrics::MetricsContainer& stats):
db(symbols, ss, stats),
relay(db, threadpool, ctx, stats),
security_timer(),
health_timer()
{ }
QueryServerImpl db;
MDRelayServer relay;
std::unique_ptr<utils::DispatcherTimer> security_timer;
std::unique_ptr<utils::DispatcherTimer> health_timer;
};
QueryServerPtr run_query_server(const bts::api::ServiceContext& ctx, std::shared_ptr<api::ISyntheticSecurityService>& ss, metrics::MetricsContainer& stats) {
auto& threadpool = utils::Dispatcher::get_named("MDQuery");
std::set<UserSymbol> all_symbols;
Venue venue;
bts_assert(api::Venue_Parse(config::get<PlatformConfig>().exchange(), &venue));
SecurityType type;
bts_assert(api::SecurityType_Parse(config::get<PlatformConfig>().type(), &type));
get_system_md_symbols(venue, type, all_symbols);
QueryServerPtr container = std::make_shared<QueryServer>(all_symbols, threadpool, ctx, ss, stats);
auto& db = container->db;
// Scan for new securities every half second
container->security_timer = std::make_unique<utils::DispatcherTimer>(threadpool, config::get<RelayOptions>().security_scan_interval(), [&db]() {
db.scan_securities();
});
container->security_timer->start();
// Scan for dead producers every 30 seconds
container->health_timer = std::make_unique<utils::DispatcherTimer>(threadpool, config::get<RelayOptions>().recovery_interval(), [&db, &threadpool]() {
db.check_heartbeats(threadpool);
});
container->health_timer->start();
// Add a few threads to keep md latency low
for (int x = 1; x < std::min(4, os::num_cores()); ++x) {
threadpool.add_thread();
}
return container;
}
std::unique_ptr<IQueryServer> create_query_server(const std::set<UserSymbol>& symbols, std::shared_ptr<api::ISyntheticSecurityService>& ss, metrics::MetricsContainer& stats) {
return std::make_unique<QueryServerImpl>(symbols, ss, stats);
}
}}//namespace
#endif
Reference in new issue View Git Blame Copy Permalink