186 lines
6.3 KiB
C++
186 lines
6.3 KiB
C++
#include "SocketConnection.h"
|
|
#include "lib/jetfix/store/session_name.h"
|
|
#include "lib/logging/logging.h"
|
|
#include "lib/metrics/metrics.h"
|
|
#include "lib/tracing/bts_trace.h"
|
|
#include "lib/utils/TimeStamp.h"
|
|
#include "lib/utils/bts_assert.h"
|
|
#include "lib/utils/helper_macros.h"
|
|
|
|
#include <arpa/inet.h>
|
|
#include <fstream>
|
|
#include <netinet/tcp.h>
|
|
#include <sys/socket.h>
|
|
#include <sys/types.h>
|
|
#include <sys/uio.h>
|
|
#include <unistd.h>
|
|
|
|
namespace bts {
|
|
namespace jetfix {
|
|
|
|
SocketConnection::SocketConnection(bts::config::JetFIX::Connection const &cfg,
|
|
metrics::MetricsContainer &stats)
|
|
: _fd(-1), _cfg(cfg), _log(cfg.log_fix_stream()), _in_log(), _out_log(),
|
|
_stats(stats), _tick_to_order(), _fill_to_order() {
|
|
if (_log) {
|
|
_in_log.open(std::string("/tmp/fix_raw_read_log_") + session_name(cfg),
|
|
std::ios_base::app);
|
|
_out_log.open(std::string("/tmp/fix_raw_write_log_") + session_name(cfg),
|
|
std::ios_base::app);
|
|
}
|
|
std::vector<metrics::Label> labels;
|
|
labels.push_back({"connection", cfg.name()});
|
|
|
|
std::array<double, 30> buckets;
|
|
auto i = 1;
|
|
for (auto &b : buckets)
|
|
b = (i *= 2);
|
|
|
|
_tick_to_order = std::make_shared<metrics::HistogramMetric<30>>(
|
|
"spark_tick_to_order_ns", buckets, labels);
|
|
_fill_to_order = std::make_shared<metrics::HistogramMetric<30>>(
|
|
"spark_fill_to_order_ns", buckets, labels);
|
|
_stats.publish(_tick_to_order);
|
|
_stats.publish(_fill_to_order);
|
|
}
|
|
|
|
SocketConnection::~SocketConnection() {
|
|
if (_fd > 0)
|
|
close(_fd);
|
|
}
|
|
|
|
bool SocketConnection::connect() {
|
|
bts_assert(_fd == -1);
|
|
_fd = socket(AF_INET, SOCK_STREAM, 0);
|
|
if (_fd < 0) {
|
|
auto er = errno;
|
|
char msg[256];
|
|
LOG(Jetfix, Error,
|
|
"Failed to allocate socket when connecting to {}:{} errno:{} msg: {}",
|
|
_cfg.socket_connect_host_ip(), _cfg.socket_connect_port(), er,
|
|
std::string(strerror_r(er, msg, sizeof(msg))));
|
|
_fd = -1;
|
|
return false;
|
|
}
|
|
struct sockaddr_in serv_addr;
|
|
memset(&serv_addr, '0', sizeof(serv_addr));
|
|
|
|
serv_addr.sin_family = AF_INET;
|
|
serv_addr.sin_port = htons(_cfg.socket_connect_port());
|
|
auto result = inet_pton(AF_INET, _cfg.socket_connect_host_ip().c_str(),
|
|
&serv_addr.sin_addr);
|
|
bts_assert(result > 0);
|
|
|
|
result = ::connect(_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
|
|
|
|
if (result < 0) {
|
|
LOG(Jetfix, Info, "Failed to connect to {}:{}",
|
|
_cfg.socket_connect_host_ip(), _cfg.socket_connect_port());
|
|
return false;
|
|
} else {
|
|
LOG(Jetfix, Info, "Connected to {}:{}", _cfg.socket_connect_host_ip(),
|
|
_cfg.socket_connect_port());
|
|
int flag = 1;
|
|
result = ::setsockopt(_fd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag,
|
|
sizeof(int));
|
|
bts_assert(result == 0);
|
|
return true;
|
|
}
|
|
}
|
|
|
|
void SocketConnection::shutdown() {
|
|
bts_assert(_fd >= 0);
|
|
::shutdown(_fd, SHUT_RDWR);
|
|
}
|
|
|
|
int SocketConnection::read(char *to_buffer, size_t read_count) {
|
|
|
|
bts_assert(_fd >= 0);
|
|
auto result = ::read(_fd, to_buffer, read_count);
|
|
if (_log && result > 0) {
|
|
_in_log.write(to_buffer, result);
|
|
_in_log.flush();
|
|
}
|
|
if (result < 0) {
|
|
auto er = errno;
|
|
char msg[256];
|
|
LOG(Jetfix, Warn, "Could not read from jetfix connection. {}",
|
|
std::string(strerror_r(er, msg, sizeof(msg))));
|
|
}
|
|
return result;
|
|
}
|
|
|
|
int SocketConnection::write(WriteVec const &vec, size_t vec_size,
|
|
PerfTrace const &trace) {
|
|
bts_assert(_fd >= 0);
|
|
|
|
struct iovec iv[MaxWriteVecSz];
|
|
for (size_t i = 0; i < vec_size; i++) {
|
|
iv[i].iov_base = (void *)vec[i].first;
|
|
iv[i].iov_len = vec[i].second;
|
|
}
|
|
|
|
auto trace_cpy = trace;
|
|
trace_cpy.fix_socket_w = utils::TimeStamp::now();
|
|
tracepoint(bts_trace, trace_order, (char *)"SocketConnection.pre_write", 0,
|
|
trace.md_sn, 0);
|
|
auto result = ::writev(_fd, iv, (int)vec_size);
|
|
|
|
if (trace.md.to_uint64()) {
|
|
auto s2s = (trace_cpy.fix_socket_w - trace_cpy.md);
|
|
LOG(Latency, Info,
|
|
"perf_event:tick_to_order s2s_us:{} fix_socket_r:{} md:{} "
|
|
"md_src:{} spreader_on_book:{} spreader_modify_order:{} "
|
|
"spreader_on_child_order_update:{} "
|
|
"order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{} order_key:{}",
|
|
s2s.microseconds(), trace_cpy.fix_socket_r.to_uint64(),
|
|
trace_cpy.md.to_uint64(), trace_cpy.md_src.to_uint64(),
|
|
trace_cpy.spreader_on_book.to_uint64(),
|
|
trace_cpy.spreader_modify_order.to_uint64(),
|
|
trace_cpy.spreader_on_child_order_update.to_uint64(),
|
|
trace_cpy.order_update.to_uint64(),
|
|
trace_cpy.exe_gw.to_uint64(), trace_cpy.fix_socket_w.to_uint64(),
|
|
trace_cpy.md_sn,
|
|
trace_cpy.order_key);
|
|
_tick_to_order->observe(s2s.nanoseconds());
|
|
}
|
|
|
|
if (trace.fix_socket_r.to_uint64()) {
|
|
auto s2s = (trace_cpy.fix_socket_w - trace_cpy.fix_socket_r);
|
|
LOG(Latency, Info,
|
|
"perf_event:fill_to_order s2s_us:{} fix_socket_r:{} md:{} md_src:{} "
|
|
"spreader_on_book:{} spreader_modify_order:{} "
|
|
"spreader_on_child_order_update:{} "
|
|
"order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{} order_key:{}",
|
|
s2s.microseconds(), trace_cpy.fix_socket_r.to_uint64(),
|
|
trace_cpy.md.to_uint64(), trace_cpy.md_src.to_uint64(),
|
|
trace_cpy.spreader_on_book.to_uint64(),
|
|
trace_cpy.spreader_modify_order.to_uint64(),
|
|
trace_cpy.spreader_on_child_order_update.to_uint64(),
|
|
trace_cpy.order_update.to_uint64(),
|
|
trace_cpy.exe_gw.to_uint64(), trace_cpy.fix_socket_w.to_uint64(),
|
|
trace_cpy.md_sn,
|
|
trace.order_key);
|
|
_fill_to_order->observe(s2s.nanoseconds());
|
|
}
|
|
|
|
tracepoint(bts_trace, trace_order, (char *)"SocketConnection.post_write", 0,
|
|
trace.md_sn, 0);
|
|
|
|
if (_log && result > 0) {
|
|
for (size_t i = 0; i < vec_size; i++)
|
|
_out_log.write(vec[i].first, vec[i].second);
|
|
_out_log.flush();
|
|
}
|
|
if (result < 0) {
|
|
auto er = errno;
|
|
char msg[256];
|
|
LOG(Jetfix, Warn, "Could not write to jetfix connection. ",
|
|
std::string(strerror_r(er, msg, sizeof(msg))));
|
|
}
|
|
return result;
|
|
}
|
|
|
|
} // namespace jetfix
|
|
} // namespace bts
|