|
@@ -25,6 +25,12 @@ void MemoryStore::start() {
|
|
|
|
|
|
|
|
// Start expiration thread
|
|
// Start expiration thread
|
|
|
expirationThread_ = std::thread(&MemoryStore::expirationLoop, this);
|
|
expirationThread_ = std::thread(&MemoryStore::expirationLoop, this);
|
|
|
|
|
+
|
|
|
|
|
+ // Start eviction thread
|
|
|
|
|
+ evictionThread_ = std::thread(&MemoryStore::evictionLoop, this);
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::info("MemoryStore started (max memory: {} MB, eviction threshold: {}%)",
|
|
|
|
|
+ config_.maxMemoryBytes / (1024 * 1024), config_.evictionThresholdPercent);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void MemoryStore::stop() {
|
|
void MemoryStore::stop() {
|
|
@@ -35,6 +41,9 @@ void MemoryStore::stop() {
|
|
|
if (expirationThread_.joinable()) {
|
|
if (expirationThread_.joinable()) {
|
|
|
expirationThread_.join();
|
|
expirationThread_.join();
|
|
|
}
|
|
}
|
|
|
|
|
+ if (evictionThread_.joinable()) {
|
|
|
|
|
+ evictionThread_.join();
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void MemoryStore::setEventCallback(EventCallback callback) {
|
|
void MemoryStore::setEventCallback(EventCallback callback) {
|
|
@@ -103,6 +112,26 @@ bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
return true;
|
|
return true;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+bool MemoryStore::alterCollection(const std::string& name, const CollectionOptions& options) {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it == collections_.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> collLock(it->second->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Update collection options
|
|
|
|
|
+ it->second->options = options;
|
|
|
|
|
+ it->second->updatedAt = currentTimeMs();
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::debug("Altered collection '{}': pinned={}, encrypted={}",
|
|
|
|
|
+ name, options.pinned, options.encrypted);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
std::vector<std::string> MemoryStore::listCollections() const {
|
|
std::vector<std::string> MemoryStore::listCollections() const {
|
|
|
std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
|
|
@@ -205,6 +234,10 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
std::optional<Document> MemoryStore::get(const std::string& collection, const std::string& id) const {
|
|
std::optional<Document> MemoryStore::get(const std::string& collection, const std::string& id) const {
|
|
|
const CollectionData* coll = getCollection(collection);
|
|
const CollectionData* coll = getCollection(collection);
|
|
|
if (!coll) {
|
|
if (!coll) {
|
|
|
|
|
+ // Check if document was evicted (collection may not exist in memory)
|
|
|
|
|
+ if (const_cast<MemoryStore*>(this)->isDocumentEvicted(collection, id)) {
|
|
|
|
|
+ return const_cast<MemoryStore*>(this)->loadEvictedDocument(collection, id);
|
|
|
|
|
+ }
|
|
|
return std::nullopt;
|
|
return std::nullopt;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -212,6 +245,11 @@ std::optional<Document> MemoryStore::get(const std::string& collection, const st
|
|
|
|
|
|
|
|
auto it = coll->documents.find(id);
|
|
auto it = coll->documents.find(id);
|
|
|
if (it == coll->documents.end()) {
|
|
if (it == coll->documents.end()) {
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+ // Check if document was evicted
|
|
|
|
|
+ if (const_cast<MemoryStore*>(this)->isDocumentEvicted(collection, id)) {
|
|
|
|
|
+ return const_cast<MemoryStore*>(this)->loadEvictedDocument(collection, id);
|
|
|
|
|
+ }
|
|
|
return std::nullopt;
|
|
return std::nullopt;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -220,6 +258,9 @@ std::optional<Document> MemoryStore::get(const std::string& collection, const st
|
|
|
return std::nullopt;
|
|
return std::nullopt;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ // Update last access time (const_cast needed since get() is const for API compatibility)
|
|
|
|
|
+ const_cast<Document&>(it->second).lastAccessedAt = currentTimeMs();
|
|
|
|
|
+
|
|
|
return it->second;
|
|
return it->second;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -1555,4 +1596,258 @@ const MemoryStore::CollectionData* MemoryStore::getCollection(const std::string&
|
|
|
return nullptr;
|
|
return nullptr;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// ===== LRU Eviction Implementation =====
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::setDocumentLoadCallback(DocumentLoadCallback callback) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ documentLoadCallback_ = std::move(callback);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::evictionLoop() {
|
|
|
|
|
+ while (running_.load()) {
|
|
|
|
|
+ std::this_thread::sleep_for(
|
|
|
|
|
+ std::chrono::milliseconds(config_.evictionCheckIntervalMs)
|
|
|
|
|
+ );
|
|
|
|
|
+
|
|
|
|
|
+ if (!running_.load()) break;
|
|
|
|
|
+
|
|
|
|
|
+ // Check memory usage
|
|
|
|
|
+ auto stats = getStats();
|
|
|
|
|
+ uint64_t threshold = config_.maxMemoryBytes * config_.evictionThresholdPercent / 100;
|
|
|
|
|
+
|
|
|
|
|
+ if (stats.estimatedMemoryBytes > threshold) {
|
|
|
|
|
+ uint64_t target = config_.maxMemoryBytes * config_.evictionTargetPercent / 100;
|
|
|
|
|
+ uint64_t bytesToFree = stats.estimatedMemoryBytes - target;
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::info("Memory at {} MB ({}%), starting eviction to free {} MB",
|
|
|
|
|
+ stats.estimatedMemoryBytes / (1024 * 1024),
|
|
|
|
|
+ stats.estimatedMemoryBytes * 100 / config_.maxMemoryBytes,
|
|
|
|
|
+ bytesToFree / (1024 * 1024));
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t evicted = evictDocuments();
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::info("Eviction complete: {} documents evicted", evicted);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::evictDocuments() {
|
|
|
|
|
+ auto stats = getStats();
|
|
|
|
|
+ uint64_t target = config_.maxMemoryBytes * config_.evictionTargetPercent / 100;
|
|
|
|
|
+
|
|
|
|
|
+ if (stats.estimatedMemoryBytes <= target) {
|
|
|
|
|
+ return 0; // No eviction needed
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t bytesToFree = stats.estimatedMemoryBytes - target;
|
|
|
|
|
+ auto candidates = selectDocumentsForEviction(bytesToFree);
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t evicted = 0;
|
|
|
|
|
+ for (const auto& [collection, id] : candidates) {
|
|
|
|
|
+ // Get current WAL sequence (we'll use version as proxy since we don't have direct WAL access)
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+ uint64_t walSequence = 0;
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it != coll->documents.end()) {
|
|
|
|
|
+ // Use version as a proxy for WAL sequence
|
|
|
|
|
+ // The actual WAL sequence will be determined by the persistence layer
|
|
|
|
|
+ walSequence = it->second.version;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (evictDocument(collection, id, walSequence)) {
|
|
|
|
|
+ evicted++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return evicted;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::pair<std::string, std::string>> MemoryStore::selectDocumentsForEviction(uint64_t bytesToFree) {
|
|
|
|
|
+ // Collect all documents with their access times from non-pinned collections
|
|
|
|
|
+ // tuple: (lastAccessedAt, estimatedSize, collection, id)
|
|
|
|
|
+ std::vector<std::tuple<uint64_t, uint64_t, std::string, std::string>> docs;
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& [collName, coll] : collections_) {
|
|
|
|
|
+ // Skip pinned collections
|
|
|
|
|
+ if (coll->options.pinned) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ // Use lastAccessedAt, falling back to updatedAt or createdAt
|
|
|
|
|
+ uint64_t accessTime = doc.lastAccessedAt > 0 ? doc.lastAccessedAt :
|
|
|
|
|
+ (doc.updatedAt > 0 ? doc.updatedAt : doc.createdAt);
|
|
|
|
|
+ uint64_t size = estimateDocumentSize(doc);
|
|
|
|
|
+ docs.emplace_back(accessTime, size, collName, id);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Sort by access time (oldest first = least recently used)
|
|
|
|
|
+ std::sort(docs.begin(), docs.end());
|
|
|
|
|
+
|
|
|
|
|
+ // Select documents until we have enough to free
|
|
|
|
|
+ std::vector<std::pair<std::string, std::string>> candidates;
|
|
|
|
|
+ uint64_t totalSize = 0;
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& [accessTime, size, coll, id] : docs) {
|
|
|
|
|
+ if (totalSize >= bytesToFree) break;
|
|
|
|
|
+ candidates.emplace_back(coll, id);
|
|
|
|
|
+ totalSize += size;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return candidates;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::evictDocument(const std::string& collection, const std::string& id, uint64_t walSequence) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false; // Document not found
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Create stub for recovery
|
|
|
|
|
+ EvictedDocumentStub stub;
|
|
|
|
|
+ stub.id = id;
|
|
|
|
|
+ stub.collection = collection;
|
|
|
|
|
+ stub.version = it->second.version;
|
|
|
|
|
+ stub.walSequence = walSequence;
|
|
|
|
|
+ stub.estimatedSize = estimateDocumentSize(it->second);
|
|
|
|
|
+ stub.evictedAt = currentTimeMs();
|
|
|
|
|
+
|
|
|
|
|
+ // Remove from expiration index
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove document from memory
|
|
|
|
|
+ coll->documents.erase(it);
|
|
|
|
|
+
|
|
|
|
|
+ collLock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Store stub in evicted map
|
|
|
|
|
+ {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> evictLock(evictionMutex_);
|
|
|
|
|
+ evictedDocs_[collection][id] = stub;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments--;
|
|
|
|
|
+ stats_.evictedCount++;
|
|
|
|
|
+ stats_.evictionCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ SPDLOG_DEBUG("Evicted document {}/{} (size: {} bytes)", collection, id, stub.estimatedSize);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<Document> MemoryStore::loadEvictedDocument(const std::string& collection, const std::string& id) {
|
|
|
|
|
+ EvictedDocumentStub stub;
|
|
|
|
|
+
|
|
|
|
|
+ // Get stub
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(evictionMutex_);
|
|
|
|
|
+ auto collIt = evictedDocs_.find(collection);
|
|
|
|
|
+ if (collIt == evictedDocs_.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+ auto stubIt = collIt->second.find(id);
|
|
|
|
|
+ if (stubIt == collIt->second.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+ stub = stubIt->second;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Load from persistence via callback
|
|
|
|
|
+ std::optional<Document> doc;
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ if (documentLoadCallback_) {
|
|
|
|
|
+ doc = documentLoadCallback_(collection, id, stub.walSequence);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!doc) {
|
|
|
|
|
+ spdlog::warn("Failed to recover evicted document {}/{}", collection, id);
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Re-insert into memory
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+ {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ doc->collection = collection;
|
|
|
|
|
+ doc->lastAccessedAt = currentTimeMs(); // Mark as recently accessed
|
|
|
|
|
+
|
|
|
|
|
+ // Add to expiration index if TTL is set
|
|
|
|
|
+ if (doc->expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, id, doc->expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents[id] = *doc;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove from evicted map
|
|
|
|
|
+ {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> evictLock(evictionMutex_);
|
|
|
|
|
+ auto collIt = evictedDocs_.find(collection);
|
|
|
|
|
+ if (collIt != evictedDocs_.end()) {
|
|
|
|
|
+ collIt->second.erase(id);
|
|
|
|
|
+ if (collIt->second.empty()) {
|
|
|
|
|
+ evictedDocs_.erase(collIt);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ stats_.evictedCount--;
|
|
|
|
|
+ stats_.recoveryCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::debug("Recovered evicted document {}/{}", collection, id);
|
|
|
|
|
+
|
|
|
|
|
+ return doc;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::isDocumentEvicted(const std::string& collection, const std::string& id) const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(evictionMutex_);
|
|
|
|
|
+ auto collIt = evictedDocs_.find(collection);
|
|
|
|
|
+ if (collIt == evictedDocs_.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ return collIt->second.find(id) != collIt->second.end();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<MemoryStore::EvictedDocumentStub> MemoryStore::getEvictedStub(
|
|
|
|
|
+ const std::string& collection, const std::string& id) const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(evictionMutex_);
|
|
|
|
|
+ auto collIt = evictedDocs_.find(collection);
|
|
|
|
|
+ if (collIt == evictedDocs_.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+ auto stubIt = collIt->second.find(id);
|
|
|
|
|
+ if (stubIt == collIt->second.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+ return stubIt->second;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
} // namespace smartbotic::database
|
|
} // namespace smartbotic::database
|