spark/src/lib/md/MDProducer.cpp

#ifndef NO_CPP0X
#include "MDProducer.h"
#include "MemorySegmentUtils.h"
#include "SecurityIDRecorder.h"
#include "Shm/MDStructure.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/logging/logging.h"
#include "lib/md/Diag/ShmDumper.h"
#include "lib/md/ShmSegmentName.h"
#include "lib/os/os.h"
#include "lib/utils/SharedList.h"
#include "lib/utils/WaitFreeQueue.h"

using namespace std;

namespace bts {
namespace md {

// Owns shared memory for a single symbol+type
class MDProducerImpl {
 public:
   MDProducerImpl(
       const string &symbol,
       SecurityType type,
       Venue exchange,
       MDProducerFlags flags)
       : _symbol(symbol), _type(type), _exchange(exchange),
         _hot(
             ShmSegmentName::hot_name(symbol, type, exchange),
             config::get<PublisherOptions>().hot_initial(),
             config::get<PublisherOptions>().hot_growth(),
             flags & MDProducerFlags::Shared),
         _cold(
             ShmSegmentName::cold_name(symbol, type, exchange),
             config::get<PublisherOptions>().cold_initial(),
             config::get<PublisherOptions>().cold_growth(),
             flags & MDProducerFlags::Shared),
         _trades(
             _cold.header().trade_buffer,
             BTS_ARRAYSIZE(_cold.header().trade_buffer),
             true),
         _rfqs(
             _cold.header().rfq_buffer,
             BTS_ARRAYSIZE(_cold.header().rfq_buffer),
             true) {
      // NOTE: Block sizes have been tuned based on sim.
      // We should monitor and adjust based on production use for optimal
      // performance
      _cold.header().has_depth = enable_depth(symbol, type);

      // static_assert(sizeof(HotEntry) <= 32,"HotEntry should be less than 32
      // bytes");
   }

   MDProducerHandle add_security(
       const FixedSecurity &security, SecurityMapper &smapper) {
      LOG(md, Diag, "MDProducerImpl::add_security desc: {} \nf_sec: {}",
          std::string(security.security_desc()),
          ShmDumper::dump_security(security));
      void *h_entry = nullptr;
      if (has_depth()) {
         h_entry = new (_hot.add(sizeof(HotDepthEntry))) HotDepthEntry();
      } else {
         h_entry = new (_hot.add(sizeof(HotEntry))) HotEntry();
      }

      // Calculating the size for the variable sized entries is tricky
      ColdEntry *c_entry =
          new (_cold.add(ColdEntry::fixed_size() + security.size))
              ColdEntry(security.size);
      os::memcpy(&c_entry->security, &security, security.size);
      smapper.apply(c_entry->security);

      MDProducerHandle handle(
          _cold.header(), _hot.header(), *c_entry, h_entry,
          MDProducerChannel(*this));

      if (security.key != invalid_security_key) {
         // The only safe way to set the security key is through the set_key()
         // function If the security had a key it was copied to cold by the
         // memcpy but not the hot So we clear it from the cold and let
         // set_key() set it in hot and cold and make the security public.
         c_entry->security.key = invalid_security_key;
         handle.set_key(security.key);
      }

      return handle;
   }

   void heartbeat() { _cold.header().heartbeat = utils::TimeStamp::now(); }

   void add_trade(const co2::TradeMessage &trade) {
      bts_assert(trade.key() != invalid_security_key);
      _trades.push(trade);
      _cold.header().trade_count++;
   }

   void add_rfq(const co2::QuoteRequestMessage &rfq) {
      ColdHeader::RFQBuffer buffer = {{0}};
      if (rfq.SerializePartialToArray(&buffer[0], buffer.size())) {
         _rfqs.push(buffer);
         _cold.header().rfq_count++;
      } else {
         LOG(md, Warn, "Unable to store RFQ due to size limitation {}",
             rfq.DebugString());
      }
   }

   const string &symbol() const { return _symbol; }
   SecurityType type() const { return _type; }
   Venue exchange() const { return _exchange; }
   bool has_depth() { return _cold.header().has_depth; }
   void gap() { _cold.header().gap_count++; }

 private:
   BTS_NO_COPY(MDProducerImpl);
   string _symbol;
   SecurityType _type;
   Venue _exchange;
   utils::SharedList<HotHeader, HotEntry> _hot;
   utils::SharedList<ColdHeader, ColdEntry> _cold;
   utils::WaitFreeQueue<co2::TradeMessage> _trades;
   utils::WaitFreeQueue<ColdHeader::RFQBuffer> _rfqs;
};

//
// MDProducer
//

MDProducer::MDProducer(MDProducerFlags flags)
    : _mutex(new std::mutex()), _producers(), _id_recorder(), _flags(flags),
      _smapper(flags & MDProducerFlags::SecurityMapper) {}
#ifndef _WIN32
MDProducer::MDProducer(SecurityIDRecorder recorder, MDProducerFlags flags)
    : _mutex(new std::mutex()), _producers(),
      _id_recorder(std::make_unique<SecurityIDRecorder>(recorder)),
      _flags(flags), _smapper(flags & MDProducerFlags::SecurityMapper) {}
#endif
MDProducer::MDProducer(MDProducer &&other)
    : _mutex(std::move(other._mutex)), _producers(std::move(other._producers)),
      _flags(other._flags), _smapper(_flags & MDProducerFlags::SecurityMapper) {
}

MDProducer::~MDProducer() {}

MDProducerImpl &MDProducer::get_producer(
    const char *symbol, SecurityType type, Venue exchange) {
   // First search existing producers for one that matches
   for (auto &impl : _producers) {
      if ((impl->symbol() == symbol && impl->type() == type &&
           impl->exchange() == exchange) ||
          // single channel for all synthetics
          (impl->type() == api::SECURITYTYPE_SYNTHETIC &&
           type == api::SECURITYTYPE_SYNTHETIC)) {
         return *impl;
      }
   }

   // Not found add a new one
   unique_ptr<MDProducerImpl> newp;
   if (type == api::SECURITYTYPE_SYNTHETIC) {
      // synthetics are special. We don't want to allocate a separate section
      // for each one so we put them all in one
      newp.reset(new MDProducerImpl(
          SYNTHETIC_SYMBOL, api::SECURITYTYPE_SYNTHETIC, Venue::VENUE_BTS,
          _flags));
   } else {
      newp.reset(new MDProducerImpl(symbol, type, exchange, _flags));
   }
   _producers.push_back(std::move(newp));
   return *_producers.back();
}

MDProducerHandle MDProducer::add_security(const FixedSecurity &security) {
   bts_assert_d(strlen(security.symbol()) > 0);
   bts_assert(security.size >= sizeof(FixedSecurity));
   LOG(md, Diag, "MDProducer::add_security desc:{} key: {} num_id: {}",
       std::string(security.security_desc()), security.key,
       security.numeric_id);
   std::lock_guard<std::mutex> lock(*_mutex);
   auto &impl =
       get_producer(security.symbol(), security.type, security.exchange);
   auto result = impl.add_security(security, _smapper);
   if (security.key != invalid_security_key) {
      result.set_key(security.key);
   } else if (_id_recorder) {
      _id_recorder->ensure_key(result);
   }
   return result;
}

MDProducerChannel MDProducer::get_channel(
    const char *symbol, SecurityType type, Venue exchange) {
   std::lock_guard<std::mutex> lock(*_mutex);
   return MDProducerChannel(get_producer(symbol, type, exchange));
}

void MDProducer::heartbeat() {
   std::lock_guard<std::mutex> lock(*_mutex);
   for (auto &impl : _producers) {
      impl->heartbeat();
   }
}

//
// MDProducerChannel
//

MDProducerChannel::MDProducerChannel(MDProducerImpl &impl) : _me(&impl) {}
void MDProducerChannel::heartbeat() { _me->heartbeat(); }
void MDProducerChannel::add_trade(const co2::TradeMessage &trade) {
   _me->add_trade(trade);
}
void MDProducerChannel::add_rfq(const co2::QuoteRequestMessage &rfq) {
   _me->add_rfq(rfq);
}
void MDProducerChannel::gap() { _me->gap(); }

} // namespace md
} // namespace bts

#endif