Pedro Pinto
f4dc550f1a
This slowness was actually causing us to take too long to complete the FIX connection logon process, resulting in a timeout and failure to establish the session. It took almost 1ms per log line, with logs commonly having 20K lines. k
317 lines
9.6 KiB
C++
317 lines
9.6 KiB
C++
#include "FIXFileStore.h"
|
|
|
|
#include "lib/config/config.h"
|
|
#include "lib/jetfix/codec/FIXEncodingStream.h"
|
|
#include "lib/jetfix/common/InMsgBuffer.h"
|
|
#include "lib/jetfix/common/OutFIX.h"
|
|
#include "lib/jetfix/store/session_name.h"
|
|
#include "lib/os/os.h"
|
|
#include "lib/utils/bts_chrono.h"
|
|
#include "lib/utils/datetime.h"
|
|
#include "lib/utils/file_utils.h"
|
|
#include "lib/utils/thread_manager.h"
|
|
|
|
#include <boost/date_time.hpp>
|
|
#include <boost/date_time/gregorian/gregorian.hpp>
|
|
#include <boost/filesystem.hpp>
|
|
#include <tbb/concurrent_queue.h>
|
|
|
|
#include <fmt/format.h>
|
|
#include <fstream>
|
|
#include <regex>
|
|
#include <thread>
|
|
|
|
namespace bts {
|
|
namespace jetfix {
|
|
|
|
struct FIXFileStore::Event {
|
|
Event()
|
|
: out_msgs(), in_msg(), quit(false), has_last_msg_processed(false),
|
|
last_msg_processed(0), has_seq_num(false), seq_num(0) {}
|
|
|
|
Event(Event const &src)
|
|
: out_msgs(src.out_msgs), in_msg(src.in_msg), quit(src.quit),
|
|
has_last_msg_processed(src.has_last_msg_processed),
|
|
last_msg_processed(src.last_msg_processed),
|
|
has_seq_num(src.has_seq_num), seq_num(src.seq_num) {}
|
|
|
|
Event(Event &&src)
|
|
: out_msgs(std::move(src.out_msgs)), in_msg(std::move(src.in_msg)),
|
|
quit(src.quit), has_last_msg_processed(src.has_last_msg_processed),
|
|
last_msg_processed(src.last_msg_processed),
|
|
has_seq_num(src.has_seq_num), seq_num(src.seq_num) {}
|
|
|
|
Event &operator=(Event const &src) {
|
|
out_msgs = src.out_msgs;
|
|
in_msg = src.in_msg;
|
|
quit = src.quit;
|
|
has_last_msg_processed = src.has_last_msg_processed;
|
|
last_msg_processed = src.last_msg_processed;
|
|
has_seq_num = src.has_seq_num;
|
|
seq_num = src.seq_num;
|
|
return *this;
|
|
}
|
|
|
|
Event &operator=(Event const &&src) {
|
|
out_msgs = std::move(src.out_msgs);
|
|
in_msg = std::move(src.in_msg);
|
|
quit = src.quit;
|
|
has_last_msg_processed = src.has_last_msg_processed;
|
|
last_msg_processed = src.last_msg_processed;
|
|
has_seq_num = src.has_seq_num;
|
|
seq_num = src.seq_num;
|
|
return *this;
|
|
}
|
|
|
|
OutFIXPtr out_msgs;
|
|
InMsgBuffer in_msg;
|
|
bool quit;
|
|
bool has_last_msg_processed;
|
|
uint32_t last_msg_processed;
|
|
bool has_seq_num;
|
|
uint32_t seq_num;
|
|
};
|
|
|
|
struct FIXFileStore::PImpl {
|
|
PImpl(config::JetFIX::Connection const &cfg)
|
|
: active(false), store_called(false), replay_messages(), last_seq_num(0),
|
|
last_seq_num_processed(0), seq_num_path(), fix_store(),
|
|
writer_thread(), event_queue(), cfg(cfg) {}
|
|
|
|
bool active;
|
|
bool store_called;
|
|
std::vector<InMsgBuffer> replay_messages;
|
|
uint32_t last_seq_num;
|
|
uint32_t last_seq_num_processed;
|
|
std::string seq_num_path;
|
|
std::ofstream fix_store;
|
|
std::thread writer_thread;
|
|
tbb::concurrent_queue<Event> event_queue;
|
|
config::JetFIX::Connection const cfg;
|
|
};
|
|
|
|
std::string fix_log_path(std::string const &base_path) {
|
|
boost::filesystem::path result = base_path;
|
|
result /= "fix";
|
|
return result.string();
|
|
}
|
|
|
|
std::string seqnum_path(std::string const &base_path) {
|
|
boost::filesystem::path result = base_path;
|
|
result /= "seqnum";
|
|
return result.string();
|
|
}
|
|
|
|
std::string store_path(std::string base_path,
|
|
std::string const &connection_name,
|
|
std::string const &session_name) {
|
|
if (base_path.empty())
|
|
base_path = os::bts_data_path();
|
|
boost::filesystem::path result = utils::normalize_data_path(base_path);
|
|
result /= "jetfix";
|
|
result /= connection_name;
|
|
result /= session_name;
|
|
return result.string();
|
|
}
|
|
|
|
static std::regex store_line_rx("\\d\\d\\d\\d\\d\\d\\d\\d-"
|
|
"\\d\\d:\\d\\d:\\d\\d\\."
|
|
"\\d\\d\\d . 8=FIX\\.4\\.2.*");
|
|
|
|
bool is_valid_store_line(std::string const &line) {
|
|
return std::regex_match(line, store_line_rx);
|
|
}
|
|
|
|
std::map<uint32_t, InMsgBuffer>
|
|
load_replay(config::JetFIX::Connection const &cfg, std::string const &path,
|
|
uint32_t) {
|
|
std::map<uint32_t, InMsgBuffer> result;
|
|
if (boost::filesystem::exists(path)) {
|
|
|
|
std::ifstream f(path);
|
|
std::string line_buffer;
|
|
line_buffer.reserve(10 * 1024);
|
|
std::string date_buffer("20150427-19:05:17.032");
|
|
|
|
for (;;) {
|
|
std::getline(f, line_buffer);
|
|
if (!f.eof()) {
|
|
if (!is_valid_store_line(line_buffer))
|
|
continue;
|
|
|
|
if (line_buffer[22] == '<') {
|
|
auto ts =
|
|
datetime::parse_FIX_timestamp(line_buffer.substr(0, 21));
|
|
if (!datetime::is_at_least_secs_old(ts,
|
|
cfg.seconds_to_replay())) {
|
|
InMsgBuffer msg(line_buffer.substr(24));
|
|
auto sn = msg.get_seq_num();
|
|
result[sn] = msg;
|
|
}
|
|
}
|
|
} else
|
|
break;
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
void save_fix(std::ostream &out, const char *data, size_t size,
|
|
bool is_outgoing) {
|
|
out << datetime::FIX_timestamp(datetime::now()) << " "
|
|
<< (is_outgoing ? ">" : "<") << " ";
|
|
out.write(data, size);
|
|
out << std::endl;
|
|
}
|
|
|
|
void save_seq_nums(std::string const &path, uint32_t last_seq_num,
|
|
uint32_t last_seq_num_processed) {
|
|
std::ofstream f(path);
|
|
f << last_seq_num << " " << last_seq_num_processed << std::endl;
|
|
}
|
|
|
|
void load_seq_nums(std::string const &path, uint32_t *last_seq_num,
|
|
uint32_t *last_seq_num_processed) {
|
|
if (boost::filesystem::exists(path)) {
|
|
std::ifstream f(path);
|
|
f >> *last_seq_num >> *last_seq_num_processed;
|
|
} else {
|
|
*last_seq_num = 0;
|
|
*last_seq_num_processed = 0;
|
|
save_seq_nums(path, *last_seq_num, *last_seq_num_processed);
|
|
}
|
|
}
|
|
|
|
FIXFileStore::FIXFileStore(config::JetFIX::Connection const &cfg,
|
|
std::string base_path)
|
|
: me(new PImpl(cfg)) {
|
|
auto sn = session_name(cfg);
|
|
me->active = sn.size() > 0;
|
|
if (me->active) {
|
|
|
|
auto path = store_path(base_path, cfg.name(), sn);
|
|
if (!boost::filesystem::exists(path))
|
|
bts_assert(boost::filesystem::create_directories(path));
|
|
|
|
// load seq numbers
|
|
auto seq_num_p = seqnum_path(path);
|
|
load_seq_nums(seq_num_p, &me->last_seq_num, &me->last_seq_num_processed);
|
|
me->seq_num_path = seq_num_p;
|
|
|
|
// load replay
|
|
auto fix_log_p = fix_log_path(path);
|
|
auto replay_messages =
|
|
load_replay(cfg, fix_log_p, me->last_seq_num_processed);
|
|
auto max_sn = 0u;
|
|
for (auto &p : replay_messages) {
|
|
auto sn = p.first;
|
|
auto &m = p.second;
|
|
if (sn > max_sn) {
|
|
me->replay_messages.emplace_back(m);
|
|
max_sn = sn;
|
|
}
|
|
}
|
|
me->fix_store.open(fix_log_p, std::ofstream::app);
|
|
auto thread_name = fmt::format("jf_store_{}", me->cfg.name().c_str());
|
|
me->writer_thread = threading::make_thread(
|
|
thread_name, std::bind(&FIXFileStore::write, this));
|
|
}
|
|
}
|
|
|
|
FIXFileStore::~FIXFileStore() {
|
|
if (me->active) {
|
|
Event ev;
|
|
ev.quit = true;
|
|
me->event_queue.push(ev);
|
|
me->writer_thread.join();
|
|
}
|
|
}
|
|
|
|
bool FIXFileStore::session_active() const { return me->active; }
|
|
|
|
std::vector<InMsgBuffer> FIXFileStore::replay() const {
|
|
bts_assert(me->active && !me->store_called);
|
|
return me->replay_messages;
|
|
}
|
|
|
|
uint32_t FIXFileStore::last_seq_num() const {
|
|
bts_assert(me->active && !me->store_called);
|
|
return me->last_seq_num;
|
|
}
|
|
|
|
uint32_t FIXFileStore::last_seq_num_processed() const {
|
|
bts_assert(me->active && !me->store_called);
|
|
return me->last_seq_num_processed;
|
|
}
|
|
|
|
void FIXFileStore::store(OutFIXPtr const &msgs) {
|
|
bts_assert(me->active);
|
|
me->store_called = true;
|
|
Event ev;
|
|
ev.out_msgs = msgs;
|
|
me->event_queue.push(ev);
|
|
}
|
|
|
|
void FIXFileStore::store(InMsgBuffer const &msg) {
|
|
bts_assert(me->active);
|
|
me->store_called = true;
|
|
Event ev;
|
|
ev.in_msg = msg;
|
|
me->event_queue.push(ev);
|
|
}
|
|
|
|
void FIXFileStore::set_last_msg_processed(uint32_t last_msg_processed) {
|
|
bts_assert(me->active);
|
|
me->store_called = true;
|
|
Event ev;
|
|
ev.has_last_msg_processed = true;
|
|
ev.last_msg_processed = last_msg_processed;
|
|
me->event_queue.push(ev);
|
|
}
|
|
|
|
void FIXFileStore::set_last_seq_num(uint32_t last_seq_num) {
|
|
bts_assert(me->active);
|
|
me->store_called = true;
|
|
Event ev;
|
|
ev.has_seq_num = true;
|
|
ev.seq_num = last_seq_num;
|
|
me->event_queue.push(ev);
|
|
}
|
|
|
|
void FIXFileStore::write() {
|
|
Event ev;
|
|
|
|
while (true)
|
|
if (me->event_queue.try_pop(ev)) {
|
|
if (ev.quit) {
|
|
return;
|
|
} else if (ev.out_msgs) {
|
|
for (auto &encoded_stream : ev.out_msgs->fix)
|
|
for (auto &mm : encoded_stream.msg_map()) {
|
|
me->last_seq_num = std::max(me->last_seq_num, mm.seq_num);
|
|
save_seq_nums(me->seq_num_path, me->last_seq_num,
|
|
me->last_seq_num_processed);
|
|
save_fix(me->fix_store, encoded_stream.data() + mm.offset,
|
|
mm.size, true);
|
|
}
|
|
me->fix_store.flush();
|
|
} else if (ev.in_msg.size()) {
|
|
save_fix(me->fix_store, ev.in_msg.data(), ev.in_msg.size(), false);
|
|
me->fix_store.flush();
|
|
} else if (ev.has_last_msg_processed) {
|
|
me->last_seq_num_processed = ev.last_msg_processed;
|
|
save_seq_nums(me->seq_num_path, me->last_seq_num,
|
|
me->last_seq_num_processed);
|
|
} else if (ev.has_seq_num) {
|
|
me->last_seq_num = ev.seq_num;
|
|
save_seq_nums(me->seq_num_path, me->last_seq_num,
|
|
me->last_seq_num_processed);
|
|
} else
|
|
bts_assert(false);
|
|
} else
|
|
std::this_thread::sleep_for(100ms);
|
|
}
|
|
|
|
} // namespace jetfix
|
|
} // namespace bts
|