spark/src/lib/jetfix/connection/AsyncReader.cpp

#ifndef _WIN32
#include "AsyncReader.h"

#include "IConnection.h"
#include "lib/jetfix/gateway/ICoreGatewayEventSink.h"
#include "lib/os/os.h"
#include "lib/tracing/bts_trace.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/thread_manager.h"
#include <thread>

#include <fstream>

namespace bts {
namespace jetfix {

// frame FIX messages contained in a raw buffer, storing their
// location in a vector InMsgBuffer. Returns how many bytes were
// successfully framed.
size_t extract(
    char *start,
    char *end,
    std::shared_ptr<std::string> const &buffer,
    std::vector<InMsgBuffer> *msgs) {
   auto first_unprocessed_byte = 0;
   auto current = start;

   for (;;) {
      // consume prefix
      auto this_msg_begin = current;
      constexpr const char *prefix = "8=FIX.4.2\0019=";
      for (int i = 0; i < 12; i++)
         if (current == end)
            return first_unprocessed_byte;
         else if (*current != prefix[i])
            return first_unprocessed_byte;
         else
            current++;

      // consume body_length;
      auto body_length = 0;
      for (;;) {
         if (current == end)
            return first_unprocessed_byte;
         else if (*current == 1) {
            current++;
            break;
         } else {
            body_length *= 10;
            body_length += *current - '0';
            current++;
         }
      }
      bts_assert(body_length);

      // consume body
      current += body_length + sizeof("10=000");
      if (current > end)
         return first_unprocessed_byte;
      else {
         msgs->emplace_back(this_msg_begin, current - this_msg_begin, buffer);
         first_unprocessed_byte = current - start;
      }
   }
}

struct AsyncReader::PImpl {
   PImpl(
       IConnection &connection,
       std::string const &connection_name,
       ICoreGatewayEventSink &sink,
       trace::PerfLog &l,
       size_t buffer_sz,
       size_t high_water_mark)
       : connection(connection),
         working_buffer(std::make_shared<std::string>(buffer_sz, 0)),
         first_pending_msg(0), write_cursor(0), reader_thread(), sink(sink),
         perf_log(l), high_water_mark(high_water_mark),
         connection_name(connection_name),
         channel(new std::string(connection_name)) {}

   // abstraction for source of raw bytes
   IConnection &connection;
   // a buffer used to store ready message, wheter partial or complete
   std::shared_ptr<std::string> working_buffer;
   // offset in working buffer of the first incomplete (and non
   // framed) msg
   size_t first_pending_msg;
   // the current position in the working buffer
   size_t write_cursor;
   // the thread that will be pulling bytes out of the connection
   std::thread reader_thread;
   // an observer that will process the messages we frame
   ICoreGatewayEventSink &sink;
   // For performance analysis
   trace::PerfLog &perf_log;
   // a position in the current buffer that if reached signals the
   // need to increase the buffer size
   uint32_t high_water_mark;
   const std::string connection_name;

   // we leak this because we do not have a good way to pass a pointer
   // to the perf_log object and guarantee that pointer is valid by
   // the time the string is written to disk
   const std::string *channel;
};

AsyncReader::AsyncReader(
    IConnection &connection,
    std::string const &connection_name,
    ICoreGatewayEventSink &sink,
    trace::PerfLog &log,
    size_t buffer_sz,
    size_t high_water_mark)
    : me(new PImpl(
          connection, connection_name, sink, log, buffer_sz, high_water_mark)) {
}

void AsyncReader::start() {
   auto thread_name = fmt::format("jf_ar_{}", me->connection_name);
   me->reader_thread =
       threading::make_thread(thread_name, [this]() { dispatch(); });
}

char *AsyncReader::cursor() {
   return &((*me->working_buffer)[me->write_cursor]);
}

size_t AsyncReader::free_bytes() {
   return me->working_buffer->size() - me->write_cursor;
}

void AsyncReader::dispatch() {
   std::vector<InMsgBuffer> framed_msgs;
   framed_msgs.reserve(100);

   for (;;) {
      // read as many bytes as we have buffer space for
      auto read_count = me->connection.read(cursor(), free_bytes());
      me->perf_log.log(trace::CorID{me->channel->c_str(), 0}, "fix_socket_r");
      auto socket_read_time = utils::TimeStamp::now();
      trace::CorID trace;
      trace.channel = me->channel->c_str();

      // signal a disconnect if the connection broke
      if (read_count <= 0) {
         me->sink.reader_disconnected();
         return;
      } else {
         // move the write cursor
         me->write_cursor += read_count;

         // see what we can parse with the just read bytes
         framed_msgs.clear();

         // extract as many messages as we can from the last unprocess
         // byte to the current write cursor, return offset of first
         // byte in the buffer that we could not process (possibly the
         // first byte of an incomplete msg)
         auto first_unprocessed_offset = extract(
             &((*me->working_buffer)[me->first_pending_msg]),
             &((*me->working_buffer)[me->write_cursor]), me->working_buffer,
             &framed_msgs);
         me->first_pending_msg += first_unprocessed_offset;

         for (auto &msg : framed_msgs) {
            trace.sn = msg.get_seq_num();
            socket_read_time.shift_ns(1);
            me->perf_log.log(trace, socket_read_time, "fix_msg_r");
            msg.set_trace(trace);
            me->sink.in_msg_raw(msg);
         }

         // do we need a new buffer?
         if (me->write_cursor >=
             me->working_buffer->size() - me->high_water_mark) {
            auto old_buffer = me->working_buffer;
            auto pending_size = me->write_cursor - me->first_pending_msg;

            me->working_buffer = std::make_shared<std::string>(
                std::max(2 * pending_size, me->working_buffer->size()), 0);
            memcpy(
                &((*me->working_buffer)[0]),
                &((*old_buffer)[me->first_pending_msg]), pending_size);
            me->first_pending_msg = 0;
            me->write_cursor = pending_size;
         }
      }
   }
}

AsyncReader::~AsyncReader() {
   if (me->reader_thread.joinable())
      me->reader_thread.join();
}

} // namespace jetfix
} // namespace bts
#endif