spark/src/lib/md/SecurityCache.cpp

#include "SecurityCache.h"
#include "SecurityDB.h"
#include "lib/domain_model/Security.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/logging/logging.h"
#include "lib/md/security_type_index.h"
#include "lib/md/venue_index.h"
#include "lib/utils/LockBox.h"

#include <btree/map.h>
#include <tbb/concurrent_queue.h>
#include <unordered_map>

namespace bts {
namespace md {

#define ASSERT_FN(test, fn, ...)                                               \
   bts_assert_msg(test, "SecurityCache::" fn ". " __VA_ARGS__)
#define LOG_FN(level, fn, ...)                                                 \
   LOG(md, level, "SecurityCache::" fn ". " __VA_ARGS__)
#define DIAG_FN(fn, ...) LOG_FN(Diag, fn, __VA_ARGS__)
#define INFO_FN(fn, ...) LOG_FN(Info, fn, __VA_ARGS__)
#define WARN_FN(fn, ...) LOG_FN(Warn, fn, __VA_ARGS__)
#define ERROR_FN(fn, ...) LOG_FN(Error, fn, __VA_ARGS__)

using std::string;
using std::unordered_map;

struct SecurityCacheIndex {
   btree::map<SecurityKey, SecurityPtr> cache;
   btree::map<ExchangeSecurityKey, SecurityPtr> ex_cache;
   VenueIndex<SecurityTypeIndex<unordered_map<string, uint32_t>>>
       venue_type_ids;

   const bts::api::Security *get_security(
       api::Venue venue, const std::string &desc) {
      auto id = get_id_(venue, desc);
      auto sec_it = cache.find(SecurityKey{id});
      if (sec_it == cache.end()) {
         return nullptr;
      }
      return sec_it->second.get();
   }

   const bts::api::Security *get_security(
       api::Venue venue, api::SecurityType type, const std::string &desc) {
      auto id = get_id_(venue, type, desc);
      auto sec_it = cache.find(SecurityKey{id});
      if (sec_it == cache.end()) {
         return nullptr;
      }
      return sec_it->second.get();
   }

   const Security *try_get_security(SecurityKey key) const {
      auto iter = cache.find(key);
      if (iter != cache.end()) {
         return iter->second.get();
      }
      return (const api::Security *)nullptr;
   }

   const Security *try_get_security(ExchangeSecurityKey key) const {
      auto iter = ex_cache.find(key);
      if (iter != ex_cache.end()) {
         return iter->second.get();
      }
      return (const api::Security *)nullptr;
   }

   std::vector<const Security *> get_securities(
       api::Venue venue, api::SecurityType type, const std::string &name) {
      std::vector<api::Security const *> result;
      const bool all_types = type == api::SECURITYTYPE_UNSET;
      const bool all_venues = venue == api::VENUE_UNSET;
      const bool all_names = name.empty();
      for (auto const &p : cache) {
         const auto &s = *p.second;
         if ((all_types ||
              type == securities::security_type(s.Contract_case())) &&
             (all_venues || venue == s.exchange()) &&
             (all_names || name == securities::descriptor(s))) {
            result.push_back(p.second.get());
         }
      }
      return result;
   }

   std::vector<const Security *> get_securities(
       std::function<bool(const Security &)> include) {
      std::vector<api::Security const *> result;
      for (auto const &p : cache) {
         auto &s = *p.second;
         if (include(s)) {
            result.push_back(&s);
         }
      }
      return result;
   }

   bool add_live_security(SecurityPtr security) {
      bts_assert(security);
      if (security->has_glbx()) {
         ExchangeSecurityKey key(security->id());
         ex_cache.insert(std::make_pair(key, security));
      }
      SecurityKey key(security->id());
      // Insert but don't overwrite existing
      auto result =
          cache.insert(std::make_pair(SecurityKey(security->id()), security));
      venue_type_ids.get_or_add(security->exchange())
          .get_or_add(bts::securities::security_type(*security))
          .try_emplace(bts::securities::descriptor(*security), security->id());
      return result.second;
   }

   const Security *add_db_security(SecurityPtr security) {
      bts_assert(security);
      if (security->has_glbx()) {
         ExchangeSecurityKey key(security->id());
         ex_cache.insert(std::make_pair(key, security));
      }
      SecurityKey key(security->id());
      auto it = cache.insert(std::make_pair(key, security));
      const Security *active_sec = it.first->second.get();
      // If the security was just added, active_sec is equal to security.
      // If it was already present, it must be a recently inserted 'live'
      // security and we should use it (instead of the db security).
      venue_type_ids.get_or_add(security->exchange())
          .get_or_add(bts::securities::security_type(*active_sec))
          .try_emplace(
              bts::securities::descriptor(*active_sec), active_sec->id());
      return active_sec;
   }

 private:
   std::uint32_t get_id_(api::Venue venue, const std::string &desc) {
      auto id = get_id_(venue, api::SECURITYTYPE_SPOT, desc);
      if (id != 0) {
         return id;
      }
      id = get_id_(venue, api::SECURITYTYPE_FUTURE, desc);
      if (id != 0) {
         return id;
      }
      id = get_id_(venue, api::SECURITYTYPE_OPTION, desc);
      if (id != 0) {
         return id;
      }
      id = get_id_(venue, api::SECURITYTYPE_SPREAD, desc);
      if (id != 0) {
         return id;
      }
      id = get_id_(venue, api::SECURITYTYPE_SYNTHETIC, desc);
      if (id != 0) {
         return id;
      }
      return 0;
   }

   std::uint32_t get_id_(
       api::Venue venue, api::SecurityType type, const std::string &desc) {
      auto &type_store = venue_type_ids.get_or_add(venue);
      auto &ids = type_store.get_or_add(type);
      auto it = ids.find(desc);
      if (it != ids.end()) {
         return it->second;
      }
      return 0;
   }
};

struct SecurityCacheImpl {
   utils::LockBox<SecurityCacheIndex> cache;
   tbb::concurrent_queue<SecurityPtr> update_queue;
   void enqueue_update_db(const SecurityPtr &);
   void flush();
};

void SecurityCacheImpl::enqueue_update_db(const SecurityPtr &security) {
   DIAG_FN(
       "enqueue_update_db", "security: {}", securities::descriptor(*security));
   update_queue.push(security);
}

void SecurityCacheImpl::flush() {
#ifndef _WIN32
   if (!update_queue.empty()) {
      std::vector<SecurityPtr> work;
      SecurityPtr temp;
      while (update_queue.try_pop(temp)) {
         work.push_back(temp);
      }
      if (!work.empty()) {
         INFO_FN("flush", "Saving securities to db. count: {}", work.size());
         if (db::insert_or_replace(work) == Result::error) {
            WARN_FN(
                "flush",
                "Failed to save securities to db. Will retry on next flush. "
                "count: {}",
                work.size());
            // If we fail to insert then retry later
            for (auto &s : work)
               update_queue.push(std::move(s));
         }
      }
   }
#endif
}

//////////////////

SecurityCache::SecurityCache() : _me(std::make_shared<SecurityCacheImpl>()) {}
SecurityCache::~SecurityCache() {}

const Security *SecurityCache::try_get_security(SecurityKey key) const {
   return _me->cache.lock(
       [key](SecurityCacheIndex &t) { return t.try_get_security(key); });
}

const Security *SecurityCache::try_get_security(ExchangeSecurityKey key) const {
   return _me->cache.lock(
       [key](SecurityCacheIndex &t) { return t.try_get_security(key); });
}

const Security *SecurityCache::get_security(SecurityKey key) {
   const Security *result = try_get_security(key);
   if (result == nullptr) {
      // outside lock
      auto security = db::find_security(key);
      if (security) {
         LOG(md, Info, "Loaded security from database: {} {}",
             security->symbol(), security->id());
         return _me->cache.lock([&security](SecurityCacheIndex &cache) {
            return cache.add_db_security(security);
         });
      }
      return nullptr;
   }
   return result;
}

const Security *SecurityCache::get_security(
    api::Venue venue, const std::string &name) {
   return _me->cache.lock([venue, name](SecurityCacheIndex &cache) {
      return cache.get_security(venue, name);
   });
}

const Security *SecurityCache::get_security(
    api::Venue venue, api::SecurityType type, const std::string &name) {
   return _me->cache.lock([venue, type, name](SecurityCacheIndex &cache) {
      return cache.get_security(venue, type, name);
   });
}

std::vector<const Security *> SecurityCache::get_securities(
    api::Venue venue, api::SecurityType type, const std::string &name) {
   return _me->cache.lock([venue, type, &name](SecurityCacheIndex &cache) {
      return cache.get_securities(venue, type, name);
   });
}

std::vector<const Security *> SecurityCache::get_securities(
    std::function<bool(const Security &)> include) {
   std::vector<api::Security const *> result;
   _me->cache.lock([&result, include](SecurityCacheIndex &cache) {
      for (auto const &p : cache.cache) {
         auto &s = *p.second;
         if (include(s)) {
            result.push_back(&s);
         }
      }
   });
   return result;
}

std::vector<api::Security const *> SecurityCache::get_securities() {
   std::vector<api::Security const *> result;
   result.reserve(100000);

   _me->cache.lock([&result](SecurityCacheIndex &cache) {
      for (const auto &p : cache.cache) {
         result.push_back(p.second.get());
      }
   });
   return result;
}

bool SecurityCache::add_live_security(const SecurityPtr &security) {
   ASSERT_FN(security, "add_live_security", "security is null");
   const bool is_new_sec = _me->cache.lock([&](SecurityCacheIndex &cache) {
      const bool is_new_sec = cache.add_live_security(security);
      _me->enqueue_update_db(security);
      return is_new_sec;
   });
   if (is_new_sec) {
      INFO_FN(
          "add_live_security", "new security: {}",
          bts::securities::descriptor(*security));
   }
   return is_new_sec;
}

void SecurityCache::flush_to_disk() { _me->flush(); }

} // namespace md
} // namespace bts