105 lines
3.2 KiB
C++
105 lines
3.2 KiB
C++
#include "SecurityCache.h"
|
|
#include "SecurityDB.h"
|
|
#include "lib/utils/Dispatcher.h"
|
|
|
|
#include <unordered_map>
|
|
#include <tbb/concurrent_queue.h>
|
|
#include <mutex>
|
|
|
|
namespace bts {
|
|
namespace md {
|
|
|
|
struct SecurityCacheImpl {
|
|
std::mutex cache_mutex;
|
|
std::unordered_map<SecurityKey, SecurityPtr> cache;
|
|
tbb::concurrent_queue<SecurityPtr> update_queue;
|
|
void enqueue_update_db(const SecurityPtr&);
|
|
void flush();
|
|
};
|
|
|
|
void SecurityCacheImpl::enqueue_update_db(const SecurityPtr& 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()) {
|
|
if (db::insert_or_replace(work) == Result::error) {
|
|
// 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() {}
|
|
|
|
SecurityPtr SecurityCache::try_get_security(SecurityKey key) const {
|
|
std::lock_guard<std::mutex> lock(_me->cache_mutex);
|
|
auto iter = _me->cache.find(key);
|
|
if (iter != _me->cache.end()) {
|
|
return iter->second;
|
|
}
|
|
return nullptr;
|
|
}
|
|
|
|
SecurityPtr SecurityCache::get_security(SecurityKey key) {
|
|
std::unique_lock<std::mutex> lock(_me->cache_mutex);
|
|
auto iter = _me->cache.find(key);
|
|
if (iter != _me->cache.end()) {
|
|
return iter->second;
|
|
}
|
|
else {
|
|
// unlock for db read
|
|
lock.unlock();
|
|
auto security = db::find_security(key);
|
|
if (security) {
|
|
LOG(md,Info, "Loaded security from database: {} {}", security->symbol(), security->id());
|
|
return add_db_security(security);
|
|
}
|
|
return security;
|
|
}
|
|
}
|
|
|
|
std::vector<api::Security const*> SecurityCache::get_securities()
|
|
{
|
|
std::vector<api::Security const*> result;
|
|
result.reserve(100000);
|
|
|
|
std::unique_lock<std::mutex> lock(_me->cache_mutex);
|
|
for (auto const&p : _me->cache)
|
|
result.push_back(p.second.get());
|
|
return result;
|
|
}
|
|
|
|
SecurityPtr SecurityCache::add_db_security(const SecurityPtr& security) {
|
|
bts_assert(security);
|
|
std::lock_guard<std::mutex> lock(_me->cache_mutex);
|
|
SecurityKey key(security->id());
|
|
auto result = _me->cache.insert(std::make_pair(SecurityKey(security->id()), security));
|
|
// If result was inserted result.first->second is equal to security
|
|
// If the value was already present it must be a recently inserted 'live' security and we should use it
|
|
return result.first->second;
|
|
}
|
|
bool SecurityCache::add_live_security(const SecurityPtr& security) {
|
|
bts_assert(security);
|
|
std::lock_guard<std::mutex> lock(_me->cache_mutex);
|
|
SecurityKey key(security->id());
|
|
// Insert but don't overwrite existing
|
|
auto result = _me->cache.insert(std::make_pair(SecurityKey(security->id()), security));
|
|
_me->enqueue_update_db(security);
|
|
return result.second;
|
|
}
|
|
|
|
void SecurityCache::flush_to_disk() { _me->flush(); }
|
|
|
|
}}//ns
|