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

#ifndef WIN32
#include "AsyncReader.h"

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

#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,
         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),
      high_water_mark (high_water_mark),
      connection_name(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;
   // 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;
};

AsyncReader::AsyncReader (IConnection &connection,
                          std::string const&connection_name,
                          ICoreGatewayEventSink &sink,
                          size_t buffer_sz, size_t high_water_mark):
   me (new PImpl(connection,connection_name,sink,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 ());
      PerfTrace trace;
      trace.fix_socket_r = utils::TimeStamp::now();
      
      // 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) {
            msg.set_trace(trace);
            tracepoint(bts_trace, trace_order, (char *)"AsyncReader.on_fix", 0,
                       msg.get_seq_num(), 0);
            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 ();
}

}}
#endif