spark/src/lib/md/SecurityIDRecorder.cpp

#ifndef _WIN32
#include "SecurityIDRecorder.h"
#include "lib/domain_model/Security.h"
#include "lib/logging/logging.h"
#include "lib/persistence/IStore.h"
#include "lib/persistence/SqliteStore.h"
#include "lib/utils/checked_arithmetic.h"

#include <boost/filesystem.hpp>
#include <boost/pool/object_pool.hpp>
#include <memory>
#include <sqlite/sqlite3.h>
#include <stdexcept>
#include <string>
#include <tbb/concurrent_queue.h>

using namespace std;
using namespace bts::utils;
using std::size_t;

namespace bts {
namespace md {

using bts::api::Security;
using bts::api::SecurityType;
using bts::api::Venue;
using bts::persistence::IStore;
using bts::persistence::SqliteStore;

uint64_t exchange_id(const api::Security &security);

static std::unique_ptr<IStore> get_default_store() {
   return std::make_unique<SqliteStore>("key_ids");
}

static persistence::StoreRow get_row(
    Venue venue,
    SecurityType type,
    const std::string &name,
    const std::string &prefix = "") {
   return persistence::StoreRow{
       fmt::format(
           "{}{}.{}", std::string{prefix}, Venue_Name(venue).substr(6),
           SecurityType_Name(type).substr(13)),
       name, ""};
}

static persistence::StoreRow get_row(
    const api::Security &s, const std::string &prefix = "") {
   return get_row(
       s.exchange(), bts::securities::security_type(s),
       securities::descriptor(s), prefix);
}

struct SecurityIdRequest {
   Venue venue;
   SecurityType type;
   std::string name;
   std::function<void(Venue, SecurityType, const std::string &, std::uint32_t)>
       callback;
};

struct SecurityIDRecorderState {
   SecurityIDRecorderState(
       std::unique_ptr<IStore> store, KeyLookupOption lookup_key)
       : store(store == nullptr ? get_default_store() : std::move(store)),
         pending_handle(), thread_id(), timer(), lookup_key(lookup_key),
         unresolved_legs() {}

   std::unique_ptr<IStore> store{nullptr};
   tbb::concurrent_queue<std::shared_ptr<SecurityIdRequest>> id_request_queue;
   tbb::concurrent_queue<MDProducerHandle> pending_handle;
   std::thread::id thread_id;
   unique_ptr<DispatcherTimer> timer;

   KeyLookupOption lookup_key;
   std::map<ExchangeKey, std::vector<FixedSpreadLeg *>> unresolved_legs;

   // Oh no, I got methods in my state

   void run(utils::Dispatcher &dispatcher) {
      timer = std::make_unique<utils::DispatcherTimer>(
          dispatcher, 200, [=, this]() { do_work(); });
      timer->start();
   }
   void do_work() {
      thread_id = std::this_thread::get_id();
      work_spreads();
      work_handles();
      work_db();
   }
   void work_spreads();
   void work_handles();
   void work_db();

   void assign_key(
       const api::Security &security,
       const SecurityIDRecorder::Callback &key_assigned);
   void assign_key(
       api::Venue venue,
       api::SecurityType type,
       const std::string &name,
       std::function<void(
           api::Venue, api::SecurityType, const std::string &, std::uint32_t)>
           key_assigned);

   void check_access() { bts_assert(thread_id == std::this_thread::get_id()); }
   void add_unresolved_spread(MDProducerHandle &handle);
   security_key search(const persistence::StoreRow &row);
   security_key insert(const persistence::StoreRow &row);
};

SecurityIDRecorder::SecurityIDRecorder(
    std::unique_ptr<persistence::IStore> store,
    KeyLookupOption lookup_key,
    utils::Dispatcher *dispatcher)
    : _self(std::make_unique<SecurityIDRecorderState>(
          store == nullptr
              ? std::make_unique<persistence::SqliteStore>("key_ids")
              : std::move(store),
          lookup_key)) {
   init();
   if (dispatcher != nullptr) {
      _self->run(*dispatcher);
   }
}

SecurityIDRecorder::~SecurityIDRecorder() {}

void SecurityIDRecorder::init() {}

void SecurityIDRecorder::ensure_key(MDProducerHandle handle) {
   LOG(md, Info, "SecurityIDRecorder::ensure_key desc: {} key: {} num_id: {}",
       std::string(handle.security().security_desc()), handle.security().key,
       handle.security().numeric_id);
   _self->pending_handle.push(handle);
}

void SecurityIDRecorder::assign_key(
    const api::Security &security, const Callback &callback) {
   assign_key(
       security.exchange(), bts::securities::security_type(security),
       bts::securities::descriptor(security),
       [callback, msec = security](
           api::Venue, api::SecurityType, const std::string &,
           uint32_t id) mutable {
          msec.set_id(id);
          callback(msec);
       });
}

void SecurityIDRecorderState::assign_key(
    api::Venue venue,
    api::SecurityType type,
    const std::string &name,
    std::function<
        void(api::Venue, api::SecurityType, const std::string &, std::uint32_t)>
        callback) {
   LOG(md, Info, "SecurityIDRecorder::assign_key {} {} {}", venue, type, name);
   auto id_req = std::make_shared<SecurityIdRequest>();
   id_req->venue = venue;
   id_req->type = type;
   id_req->name = name;
   id_req->callback = callback;
   id_request_queue.push(id_req);
}

void SecurityIDRecorder::assign_key(
    api::Venue venue,
    api::SecurityType type,
    const std::string &name,
    std::function<
        void(api::Venue, api::SecurityType, const std::string &, std::uint32_t)>
        callback) {
   _self->assign_key(venue, type, name, std::move(callback));
}

void SecurityIDRecorder::do_work() { _self->do_work(); }

void SecurityIDRecorderState::add_unresolved_spread(MDProducerHandle &handle) {
   bts_assert_d(lookup_key);
   FixedSpreadSecurityXL *pspread =
       static_cast<FixedSpreadSecurityXL *>(handle.mutable_security());
   if (pspread != nullptr) {
      for (size_t x = 0; x < pspread->leg_count(); ++x) {
         auto &leg = pspread->legs[x];
         bts_assert_d(leg.exchange_key.value() != 0);
         if (leg.key == invalid_security_key) {
            // Because leg is a reference to shared memory we know it will not
            // move
            for (const auto *existing : unresolved_legs[leg.exchange_key]) {
               if (existing == &leg)
                  goto next;
            }
            unresolved_legs[leg.exchange_key].push_back(&leg);
         }
      next:;
      }
   }
}

void SecurityIDRecorderState::work_spreads() {
   check_access();
   if (!lookup_key)
      return;
   KeyLookup &lookup = *lookup_key;
   int count = 0;
   for (auto iter = unresolved_legs.begin();
        count < 500 && iter != unresolved_legs.end();) {
      const auto &pair = *iter;
      ++count;
      SecurityKey key = lookup(pair.first);
      if (key.value() != invalid_security_key) {
         for (FixedSpreadLeg *leg : pair.second) {
            leg->key = key.value();
         }
         iter = unresolved_legs.erase(iter);
      } else {
         ++iter;
      }
   }
}

bool has_all_leg_keys(const MDProducerHandle &handle) {
   if (handle.security().type == SecurityType::SECURITYTYPE_SPREAD) {
      const FixedSpreadSecurityXL *pspread =
          static_cast<const FixedSpreadSecurityXL *>(&handle.security());
      for (size_t x = 0; x < pspread->leg_count(); ++x) {
         if (pspread->legs[x].key == invalid_security_key) {
            return false;
         }
      }
   }
   // SyntheticFixedSecurity is guaranteed to have all of its legs
   return true;
}

void SecurityIDRecorderState::work_handles() {
   check_access();
   Security security;
   MDProducerHandle handle;
   vector<MDProducerHandle> retry;
   while (pending_handle.try_pop(handle)) {
      security.Clear();
      // if a spread does not have all of its legs we need to wait
      if (!has_all_leg_keys(handle)) {
         // add
         add_unresolved_spread(handle);
         retry.push_back(handle);
      } else {
         convert_security(handle.security(), &security);
         assign_key(
             security.exchange(), bts::securities::security_type(security),
             bts::securities::descriptor(security),
             [handle](
                 api::Venue, api::SecurityType, const std::string &,
                 uint32_t id) mutable { handle.set_key(id); });
      }
   }

   for (const auto &h : retry) {
      pending_handle.push(h);
   }
}

void SecurityIDRecorderState::work_db() {
   check_access();
   // Drain the work queue
   // Query the database for each security. If it has an ID set it
   // If it does not have an ID add it to a list
   // Then persist that list
   // Once a security has a key it is public so record it in the security
   // database
   std::shared_ptr<SecurityIdRequest> temp;
   vector<std::pair<uint32_t, std::shared_ptr<SecurityIdRequest>>> new_entries;
   // SPARK-1446 Need to try to add existing securities to the db in case we
   // deleted them by hand
   vector<std::shared_ptr<SecurityIdRequest>> existing_securities;
   vector<persistence::StoreRow> new_rows;
   size_t count = 0;
   while (id_request_queue.try_pop(temp)) {
      security_key key = invalid_security_key;
      auto row = get_row(temp->venue, temp->type, temp->name);
      try {
         key = search(row);
      } catch (const std::overflow_error &ex) {
         LOG(persist, Error,
             "SecurityIDRecorderState::work_db. Row search threw ex: {}",
             std::string{ex.what()});
      }
      if (key != invalid_security_key) {
         temp->callback(temp->venue, temp->type, temp->name, key);
         existing_securities.push_back(std::move(temp));
      } else {
         // always push in parallel
         new_entries.push_back(std::make_pair(key, std::move(temp)));
         new_rows.push_back(std::move(row));
         ++count;
         // Limit the amount of time we spend writing to sqlite during startup
         // to avoid blocking the heartbeat
         if (count > 200) {
            break;
         }
      }
   }

   if (!new_entries.empty()) {
      store->insert_many_resolve_ids(new_rows);
      for (size_t x = 0; x < new_entries.size(); ++x) {
         auto &entry = new_entries[x];
         auto &row = new_rows[x];
         entry.first = row.id;
         bts_assert_d(entry.first != invalid_security_key);
      }
   }

   if (!new_entries.empty()) {
      for (size_t x = 0; x < new_entries.size(); ++x) {
         auto &[id, req] = new_entries[x];
         req->callback(req->venue, req->type, req->name, id);
      }
   }
}

void SecurityIDRecorder::bulk_insert(
    const std::vector<const api::Security *> &data, const char *prefix) {
   std::vector<persistence::StoreRow> rows;
   rows.reserve(data.size());
   for (const auto *s : data) {
      rows.push_back(get_row(*s, prefix));
   }
   _self->store->insert_many_resolve_ids(rows);
}

security_key SecurityIDRecorderState::search(const persistence::StoreRow &row) {
   auto [exists, rowid, val] = store->retrieve_one(row.ns, row.key);
   if (exists) {
      if (!val.empty()) {
         bts_assert(stol(val) == rowid);
      }
      return system::math::checked_ltou(rowid);
   }
   return invalid_security_key;
}

security_key SecurityIDRecorderState::insert(const persistence::StoreRow &row) {
   return store->insert(row.ns, row.key, row.value);
}

} // namespace md
} // namespace bts

#endif