|
|
@@ -96,9 +96,14 @@ bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
}
|
|
|
|
|
|
uint64_t docCount = 0;
|
|
|
+ uint64_t totalSize = 0;
|
|
|
{
|
|
|
std::shared_lock<std::shared_mutex> collLock(it->second->mutex);
|
|
|
docCount = it->second->documents.size();
|
|
|
+ // Calculate total memory being freed
|
|
|
+ for (const auto& [id, doc] : it->second->documents) {
|
|
|
+ totalSize += estimateDocumentSize(doc);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
collections_.erase(it);
|
|
|
@@ -109,6 +114,9 @@ bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
stats_.totalDocuments -= docCount;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically
|
|
|
+ estimatedMemoryBytes_.fetch_sub(totalSize, std::memory_order_relaxed);
|
|
|
+
|
|
|
return true;
|
|
|
}
|
|
|
|
|
|
@@ -176,6 +184,8 @@ bool MemoryStore::collectionExists(const std::string& name) const {
|
|
|
// ===== Document CRUD =====
|
|
|
|
|
|
std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
+ auto startTime = std::chrono::steady_clock::now();
|
|
|
+
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
@@ -215,6 +225,9 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
coll->documents[docId] = doc;
|
|
|
coll->updatedAt = doc.updatedAt;
|
|
|
|
|
|
+ // Track memory increment
|
|
|
+ uint64_t docSize = estimateDocumentSize(doc);
|
|
|
+
|
|
|
lock.unlock();
|
|
|
|
|
|
// Update stats
|
|
|
@@ -224,6 +237,15 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
stats_.insertCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically
|
|
|
+ estimatedMemoryBytes_.fetch_add(docSize, std::memory_order_relaxed);
|
|
|
+
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ insertTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(insertMaxMicros_, elapsed);
|
|
|
+
|
|
|
// Emit callbacks
|
|
|
emitPersist(collection, docId, doc, EventType::INSERT);
|
|
|
emitEvent(EventType::INSERT, collection, docId, doc.data);
|
|
|
@@ -232,11 +254,20 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
}
|
|
|
|
|
|
std::optional<Document> MemoryStore::get(const std::string& collection, const std::string& id) const {
|
|
|
+ auto startTime = std::chrono::steady_clock::now();
|
|
|
+
|
|
|
const CollectionData* coll = getCollection(collection);
|
|
|
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);
|
|
|
+ auto result = const_cast<MemoryStore*>(this)->loadEvictedDocument(collection, id);
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ const_cast<MemoryStore*>(this)->getCount_.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ const_cast<MemoryStore*>(this)->getTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(const_cast<MemoryStore*>(this)->getMaxMicros_, elapsed);
|
|
|
+ return result;
|
|
|
}
|
|
|
return std::nullopt;
|
|
|
}
|
|
|
@@ -248,7 +279,14 @@ std::optional<Document> MemoryStore::get(const std::string& collection, const st
|
|
|
lock.unlock();
|
|
|
// Check if document was evicted
|
|
|
if (const_cast<MemoryStore*>(this)->isDocumentEvicted(collection, id)) {
|
|
|
- return const_cast<MemoryStore*>(this)->loadEvictedDocument(collection, id);
|
|
|
+ auto result = const_cast<MemoryStore*>(this)->loadEvictedDocument(collection, id);
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ const_cast<MemoryStore*>(this)->getCount_.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ const_cast<MemoryStore*>(this)->getTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(const_cast<MemoryStore*>(this)->getMaxMicros_, elapsed);
|
|
|
+ return result;
|
|
|
}
|
|
|
return std::nullopt;
|
|
|
}
|
|
|
@@ -261,10 +299,19 @@ std::optional<Document> MemoryStore::get(const std::string& collection, const st
|
|
|
// Update last access time (const_cast needed since get() is const for API compatibility)
|
|
|
const_cast<Document&>(it->second).lastAccessedAt = currentTimeMs();
|
|
|
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ const_cast<MemoryStore*>(this)->getCount_.fetch_add(1, std::memory_order_relaxed);
|
|
|
+ const_cast<MemoryStore*>(this)->getTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(const_cast<MemoryStore*>(this)->getMaxMicros_, elapsed);
|
|
|
+
|
|
|
return it->second;
|
|
|
}
|
|
|
|
|
|
bool MemoryStore::update(const std::string& collection, const std::string& id, const Document& doc) {
|
|
|
+ auto startTime = std::chrono::steady_clock::now();
|
|
|
+
|
|
|
CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
|
std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
@@ -274,6 +321,9 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
+ // Track memory change (old size)
|
|
|
+ uint64_t oldSize = estimateDocumentSize(it->second);
|
|
|
+
|
|
|
// Remove from old expiration index
|
|
|
if (it->second.expiresAt > 0) {
|
|
|
removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
@@ -298,6 +348,9 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
it->second = updated;
|
|
|
coll->updatedAt = updated.updatedAt;
|
|
|
|
|
|
+ // Track memory change (new size)
|
|
|
+ uint64_t newSize = estimateDocumentSize(updated);
|
|
|
+
|
|
|
lock.unlock();
|
|
|
|
|
|
// Update stats
|
|
|
@@ -306,6 +359,19 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
stats_.updateCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically (subtract old, add new)
|
|
|
+ if (newSize > oldSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize - oldSize, std::memory_order_relaxed);
|
|
|
+ } else if (oldSize > newSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_sub(oldSize - newSize, std::memory_order_relaxed);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ updateTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(updateMaxMicros_, elapsed);
|
|
|
+
|
|
|
// Emit callbacks
|
|
|
emitPersist(collection, id, updated, EventType::UPDATE);
|
|
|
emitEvent(EventType::UPDATE, collection, id, updated.data);
|
|
|
@@ -329,6 +395,7 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
|
|
|
std::string docId = doc.id;
|
|
|
bool isInsert = false;
|
|
|
+ uint64_t oldSize = 0;
|
|
|
|
|
|
auto it = coll->documents.find(docId);
|
|
|
if (it == coll->documents.end()) {
|
|
|
@@ -350,7 +417,9 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
|
|
|
coll->documents[docId] = doc;
|
|
|
} else {
|
|
|
- // Update
|
|
|
+ // Update - track old size for memory adjustment
|
|
|
+ oldSize = estimateDocumentSize(it->second);
|
|
|
+
|
|
|
if (it->second.expiresAt > 0) {
|
|
|
removeFromExpirationIndex(*coll, docId, it->second.expiresAt);
|
|
|
}
|
|
|
@@ -373,6 +442,9 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
|
|
|
coll->updatedAt = doc.updatedAt;
|
|
|
|
|
|
+ // Track new document size
|
|
|
+ uint64_t newSize = estimateDocumentSize(doc);
|
|
|
+
|
|
|
lock.unlock();
|
|
|
|
|
|
// Update stats
|
|
|
@@ -386,6 +458,18 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically
|
|
|
+ if (isInsert) {
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize, std::memory_order_relaxed);
|
|
|
+ } else {
|
|
|
+ // Update: subtract old, add new
|
|
|
+ if (newSize > oldSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize - oldSize, std::memory_order_relaxed);
|
|
|
+ } else if (oldSize > newSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_sub(oldSize - newSize, std::memory_order_relaxed);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
// Emit callbacks
|
|
|
emitPersist(collection, docId, doc, isInsert ? EventType::INSERT : EventType::UPDATE);
|
|
|
emitEvent(isInsert ? EventType::INSERT : EventType::UPDATE, collection, docId, doc.data);
|
|
|
@@ -403,6 +487,9 @@ bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
+ // Track memory before removal
|
|
|
+ uint64_t docSize = estimateDocumentSize(it->second);
|
|
|
+
|
|
|
// Remove from expiration index
|
|
|
if (it->second.expiresAt > 0) {
|
|
|
removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
@@ -421,6 +508,9 @@ bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
|
stats_.deleteCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically
|
|
|
+ estimatedMemoryBytes_.fetch_sub(docSize, std::memory_order_relaxed);
|
|
|
+
|
|
|
// Emit callbacks
|
|
|
emitPersist(collection, id, std::nullopt, EventType::DELETE);
|
|
|
emitEvent(EventType::DELETE, collection, id);
|
|
|
@@ -459,6 +549,9 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
+ // Track memory change (old size)
|
|
|
+ uint64_t oldSize = estimateDocumentSize(it->second);
|
|
|
+
|
|
|
// Remove from old expiration index
|
|
|
if (it->second.expiresAt > 0) {
|
|
|
removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
@@ -483,6 +576,9 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
it->second = updated;
|
|
|
coll->updatedAt = updated.updatedAt;
|
|
|
|
|
|
+ // Track memory change (new size)
|
|
|
+ uint64_t newSize = estimateDocumentSize(updated);
|
|
|
+
|
|
|
lock.unlock();
|
|
|
|
|
|
// Update stats
|
|
|
@@ -491,6 +587,13 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
stats_.updateCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically (subtract old, add new)
|
|
|
+ if (newSize > oldSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize - oldSize, std::memory_order_relaxed);
|
|
|
+ } else if (oldSize > newSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_sub(oldSize - newSize, std::memory_order_relaxed);
|
|
|
+ }
|
|
|
+
|
|
|
// Emit callbacks
|
|
|
emitPersist(collection, id, updated, EventType::UPDATE);
|
|
|
emitEvent(EventType::UPDATE, collection, id, updated.data);
|
|
|
@@ -501,6 +604,8 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
// ===== Query Operations =====
|
|
|
|
|
|
QueryResult MemoryStore::find(const std::string& collection, const Query& query) const {
|
|
|
+ auto startTime = std::chrono::steady_clock::now();
|
|
|
+
|
|
|
const CollectionData* coll = getCollection(collection);
|
|
|
if (!coll) {
|
|
|
return {};
|
|
|
@@ -596,6 +701,12 @@ QueryResult MemoryStore::find(const std::string& collection, const Query& query)
|
|
|
|
|
|
result.hasMore = end < matches.size();
|
|
|
|
|
|
+ // Record timing
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
+ std::chrono::steady_clock::now() - startTime).count();
|
|
|
+ const_cast<MemoryStore*>(this)->queryTotalMicros_.fetch_add(elapsed, std::memory_order_relaxed);
|
|
|
+ updateMax(const_cast<MemoryStore*>(this)->queryMaxMicros_, elapsed);
|
|
|
+
|
|
|
return result;
|
|
|
}
|
|
|
|
|
|
@@ -859,6 +970,9 @@ uint64_t MemoryStore::expireDocuments() {
|
|
|
for (const auto& id : toExpire) {
|
|
|
auto docIt = coll->documents.find(id);
|
|
|
if (docIt != coll->documents.end()) {
|
|
|
+ // Track memory before removal
|
|
|
+ uint64_t docSize = estimateDocumentSize(docIt->second);
|
|
|
+
|
|
|
coll->documents.erase(docIt);
|
|
|
expired++;
|
|
|
|
|
|
@@ -871,6 +985,9 @@ uint64_t MemoryStore::expireDocuments() {
|
|
|
stats_.expiredCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically
|
|
|
+ estimatedMemoryBytes_.fetch_sub(docSize, std::memory_order_relaxed);
|
|
|
+
|
|
|
emitPersist(collName, id, std::nullopt, EventType::EXPIRE);
|
|
|
emitEvent(EventType::EXPIRE, collName, id);
|
|
|
|
|
|
@@ -915,29 +1032,19 @@ MemoryStore::Stats MemoryStore::getStats() const {
|
|
|
|
|
|
Stats stats = stats_;
|
|
|
|
|
|
- // Calculate estimated memory (documents + version history)
|
|
|
- std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
- stats.estimatedMemoryBytes = 0;
|
|
|
- uint64_t versionHistoryBytes = 0;
|
|
|
-
|
|
|
- for (const auto& [name, coll] : collections_) {
|
|
|
- std::shared_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
-
|
|
|
- // Count document memory
|
|
|
- for (const auto& [id, doc] : coll->documents) {
|
|
|
- stats.estimatedMemoryBytes += estimateDocumentSize(doc);
|
|
|
- }
|
|
|
+ // Populate timing stats from atomic counters (lock-free read)
|
|
|
+ stats.getCount = getCount_.load(std::memory_order_relaxed);
|
|
|
+ stats.getTotalMicros = getTotalMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.getMaxMicros = getMaxMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.insertTotalMicros = insertTotalMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.insertMaxMicros = insertMaxMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.updateTotalMicros = updateTotalMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.updateMaxMicros = updateMaxMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.queryTotalMicros = queryTotalMicros_.load(std::memory_order_relaxed);
|
|
|
+ stats.queryMaxMicros = queryMaxMicros_.load(std::memory_order_relaxed);
|
|
|
|
|
|
- // Count version history memory (fast estimate without JSON serialization)
|
|
|
- for (const auto& [docId, history] : coll->versionHistory) {
|
|
|
- for (const auto& ver : history) {
|
|
|
- // Fast estimate: use JSON tree traversal instead of dump()
|
|
|
- versionHistoryBytes += estimateJsonSize(ver.data) + 256;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- stats.estimatedMemoryBytes += versionHistoryBytes;
|
|
|
+ // Use atomic counter for document memory (O(1) instead of O(n) iteration)
|
|
|
+ stats.estimatedMemoryBytes = estimatedMemoryBytes_.load(std::memory_order_relaxed);
|
|
|
|
|
|
return stats;
|
|
|
}
|
|
|
@@ -1237,6 +1344,9 @@ void MemoryStore::loadDocument(const std::string& collection, Document doc) {
|
|
|
|
|
|
doc.collection = collection;
|
|
|
|
|
|
+ // Calculate size before moving
|
|
|
+ uint64_t docSize = estimateDocumentSize(doc);
|
|
|
+
|
|
|
// Add to expiration index
|
|
|
if (doc.expiresAt > 0) {
|
|
|
addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
@@ -1248,6 +1358,9 @@ void MemoryStore::loadDocument(const std::string& collection, Document doc) {
|
|
|
std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
stats_.totalDocuments++;
|
|
|
}
|
|
|
+
|
|
|
+ // Update memory tracking
|
|
|
+ estimatedMemoryBytes_.fetch_add(docSize, std::memory_order_relaxed);
|
|
|
}
|
|
|
|
|
|
void MemoryStore::loadDocumentWithHistory(const std::string& collection, Document doc) {
|
|
|
@@ -1257,9 +1370,15 @@ void MemoryStore::loadDocumentWithHistory(const std::string& collection, Documen
|
|
|
|
|
|
doc.collection = collection;
|
|
|
|
|
|
+ // Calculate size before moving
|
|
|
+ uint64_t newSize = estimateDocumentSize(doc);
|
|
|
+
|
|
|
// If document already exists, save current version to history before overwriting
|
|
|
auto it = coll->documents.find(doc.id);
|
|
|
if (it != coll->documents.end()) {
|
|
|
+ // Track old size for memory adjustment
|
|
|
+ uint64_t oldSize = estimateDocumentSize(it->second);
|
|
|
+
|
|
|
saveToHistory(*coll, it->second);
|
|
|
|
|
|
// Remove old expiration index entry
|
|
|
@@ -1272,6 +1391,13 @@ void MemoryStore::loadDocumentWithHistory(const std::string& collection, Documen
|
|
|
addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
}
|
|
|
it->second = std::move(doc);
|
|
|
+
|
|
|
+ // Update memory tracking (subtract old, add new)
|
|
|
+ if (newSize > oldSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize - oldSize, std::memory_order_relaxed);
|
|
|
+ } else if (oldSize > newSize) {
|
|
|
+ estimatedMemoryBytes_.fetch_sub(oldSize - newSize, std::memory_order_relaxed);
|
|
|
+ }
|
|
|
} else {
|
|
|
// New document — same as loadDocument
|
|
|
if (doc.expiresAt > 0) {
|
|
|
@@ -1280,8 +1406,13 @@ void MemoryStore::loadDocumentWithHistory(const std::string& collection, Documen
|
|
|
std::string docId = doc.id;
|
|
|
coll->documents[docId] = std::move(doc);
|
|
|
|
|
|
- std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
- stats_.totalDocuments++;
|
|
|
+ {
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
+ stats_.totalDocuments++;
|
|
|
+ }
|
|
|
+
|
|
|
+ // Update memory tracking
|
|
|
+ estimatedMemoryBytes_.fetch_add(newSize, std::memory_order_relaxed);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -1803,6 +1934,9 @@ bool MemoryStore::evictDocument(const std::string& collection, const std::string
|
|
|
stats_.evictionCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically (subtract evicted document size)
|
|
|
+ estimatedMemoryBytes_.fetch_sub(stub.estimatedSize, std::memory_order_relaxed);
|
|
|
+
|
|
|
SPDLOG_DEBUG("Evicted document {}/{} (size: {} bytes)", collection, id, stub.estimatedSize);
|
|
|
|
|
|
return true;
|
|
|
@@ -1875,6 +2009,9 @@ std::optional<Document> MemoryStore::loadEvictedDocument(const std::string& coll
|
|
|
stats_.recoveryCount++;
|
|
|
}
|
|
|
|
|
|
+ // Update memory tracking atomically (add recovered document size)
|
|
|
+ estimatedMemoryBytes_.fetch_add(estimateDocumentSize(*doc), std::memory_order_relaxed);
|
|
|
+
|
|
|
spdlog::debug("Recovered evicted document {}/{}", collection, id);
|
|
|
|
|
|
return doc;
|