soualid
diff --git a/‎arangod/RocksDBEngine/RocksDBCollection.cpp
Lines changed: 7 additions & 0 deletions b/‎arangod/RocksDBEngine/RocksDBCollection.cpp
Lines changed: 7 additions & 0 deletions
diff --git a/‎arangod/RocksDBEngine/RocksDBSettingsManager.cpp
Lines changed: 110 additions & 31 deletions b/‎arangod/RocksDBEngine/RocksDBSettingsManager.cpp
Lines changed: 110 additions & 31 deletions
diff --git a/‎arangod/RocksDBEngine/RocksDBSettingsManager.h
Lines changed: 18 additions & 17 deletions b/‎arangod/RocksDBEngine/RocksDBSettingsManager.h
Lines changed: 18 additions & 17 deletions
@@ -1847,6 +1847,7 @@ RocksDBCollection::serializeIndexEstimates(
     output.clear();
     RocksDBIndex* cindex = static_cast<RocksDBIndex*>(index.get());
     TRI_ASSERT(cindex != nullptr);
+    
     if (cindex->needToPersistEstimate()) {
       LOG_TOPIC(TRACE, Logger::ENGINES)
         << "beginning estimate serialization for index '"
@@ -1922,6 +1923,12 @@ void RocksDBCollection::recalculateIndexEstimates(
 arangodb::Result RocksDBCollection::serializeKeyGenerator(
     rocksdb::Transaction* rtrx) const {
+  if (!_logicalCollection.keyGenerator()->hasDynamicState()) {
+    // the key generator will not produce any dynamic data,
+    // so it does not need to be serialized (nor recovered)
+    return Result();
+  }
+
   VPackBuilder builder;
 
   builder.openObject();
 
@@ -67,6 +67,7 @@ arangodb::Result writeCounterValue(
 
   // Skip values which we did not change
   auto const& it = syncedSeqNums.find(pair.first);
+
   if (it != syncedSeqNums.end() && it->second == pair.second._sequenceNum) {
     return Result();
   }
@@ -164,6 +165,7 @@ writeIndexEstimatorsAndKeyGenerator(
     // or start fresh.
     return std::make_pair(Result(), returnSeq);
   }
+
   TRI_DEFER(vocbase->release());
 
   auto collection = vocbase->lookupCollection(dbColPair.second);
@@ -227,7 +229,35 @@ void RocksDBSettingsManager::CMValue::serialize(VPackBuilder& b) const {
 /// Constructor needs to be called synchrunously,
 /// will load counts from the db and scan the WAL
 RocksDBSettingsManager::RocksDBSettingsManager(rocksdb::TransactionDB* db)
-    : _lastSync(0), _syncing(false), _db(db), _initialReleasedTick(0) {}
+    : _lastSync(0), 
+      _syncing(false), 
+      _db(db), 
+      _initialReleasedTick(0),
+      _maxUpdateSeqNo(1),
+      _lastSyncedSeqNo(0) {}
+
+/// bump up the value of the last rocksdb::SequenceNumber we have seen
+/// and that is pending a sync update
+void RocksDBSettingsManager::setMaxUpdateSequenceNumber(rocksdb::SequenceNumber seqNo) {
+  if (seqNo == 0) {
+    // we don't care about this
+    return;
+  }
+
+  auto current = _maxUpdateSeqNo.load(std::memory_order_acquire);
+
+  if (current >= seqNo) {
+    // current sequence number is already higher than the one we got
+    return;
+  }
+
+  bool res = _maxUpdateSeqNo.compare_exchange_strong(current, seqNo, std::memory_order_release);
+
+  if (!res) {
+    // someone else has updated the max sequence number
+    TRI_ASSERT(current > seqNo);
+  }
+}
 
 /// retrieve initial values from the database
 void RocksDBSettingsManager::retrieveInitialValues() {
@@ -244,11 +274,13 @@ RocksDBSettingsManager::CounterAdjustment RocksDBSettingsManager::loadCounter(
   TRI_ASSERT(objectId != 0);  // TODO fix this
 
   READ_LOCKER(guard, _rwLock);
+
   auto const& it = _counters.find(objectId);
   if (it != _counters.end()) {
     return CounterAdjustment(it->second._sequenceNum, it->second._count, 0,
                              it->second._revisionId);
   }
+
   return CounterAdjustment();  // do not create
 }
 
@@ -258,6 +290,7 @@ RocksDBSettingsManager::CounterAdjustment RocksDBSettingsManager::loadCounter(
 void RocksDBSettingsManager::updateCounter(uint64_t objectId,
                                            CounterAdjustment const& update) {
   bool needsSync = false;
+  auto seqNo = update.sequenceNumber();
   {
     WRITE_LOCKER(guard, _rwLock);
 
@@ -266,19 +299,22 @@ void RocksDBSettingsManager::updateCounter(uint64_t objectId,
       it->second._count += update.added();
       it->second._count -= update.removed();
       // just use the latest trx info
-      if (update.sequenceNumber() > it->second._sequenceNum) {
-        it->second._sequenceNum = update.sequenceNumber();
+      if (seqNo > it->second._sequenceNum) {
+        it->second._sequenceNum = seqNo;
         it->second._revisionId = update.revisionId();
       }
     } else {
       // insert new counter
       _counters.emplace(std::make_pair(
           objectId,
-          CMValue(update.sequenceNumber(), update.added() - update.removed(),
+          CMValue(seqNo, update.added() - update.removed(),
                   update.revisionId())));
       needsSync = true;  // only count values from WAL if they are in the DB
     }
   }
+
+  setMaxUpdateSequenceNumber(seqNo);
+
   if (needsSync) {
     sync(true);
   }
@@ -287,24 +323,36 @@ void RocksDBSettingsManager::updateCounter(uint64_t objectId,
 arangodb::Result RocksDBSettingsManager::setAbsoluteCounter(uint64_t objectId,
                                                             uint64_t value) {
   arangodb::Result res;
-  WRITE_LOCKER(guard, _rwLock);
-  auto it = _counters.find(objectId);
-  if (it != _counters.end()) {
-    rocksdb::TransactionDB* db = rocksutils::globalRocksDB();
-    it->second._sequenceNum = db->GetLatestSequenceNumber();
-    it->second._count = value;
-  } else {
-    // nothing to do as the counter has never been written it can not be set to
-    // a value that would require correction. but we use the return value to
-    // signal that no sync is rquired
-    res.reset(TRI_ERROR_INTERNAL, "counter value not found - no sync required");
+  rocksdb::SequenceNumber seqNo = 0;
+
+  {
+    WRITE_LOCKER(guard, _rwLock);
+    
+    auto it = _counters.find(objectId);
+
+    if (it != _counters.end()) {
+      rocksdb::TransactionDB* db = rocksutils::globalRocksDB();
+      seqNo = db->GetLatestSequenceNumber();
+      it->second._sequenceNum = seqNo;
+      it->second._count = value;
+    } else {
+      // nothing to do as the counter has never been written it can not be set to
+      // a value that would require correction. but we use the return value to
+      // signal that no sync is rquired
+      res.reset(TRI_ERROR_INTERNAL, "counter value not found - no sync required");
+    }
   }
+  
+  setMaxUpdateSequenceNumber(seqNo);
+
   return res;
 }
 
 void RocksDBSettingsManager::removeCounter(uint64_t objectId) {
   WRITE_LOCKER(guard, _rwLock);
-  auto const& it = _counters.find(objectId);
+
+  auto it = _counters.find(objectId);
+
   if (it != _counters.end()) {
     RocksDBKey key;
     key.constructCounterValue(it->first);
@@ -361,24 +409,41 @@ Result RocksDBSettingsManager::sync(bool force) {
   if (!lockForSync(force)) {
     return Result();
   }
-  TRI_DEFER(_syncing = false);
+  
+  // only one thread can enter here at a time
+
+  // make sure we give up our lock when we exit this function
+  auto guard = scopeGuard([this]() { _syncing = false; });
+  
+  auto maxUpdateSeqNo =  _maxUpdateSeqNo.load(std::memory_order_acquire);
+
+  if (!force && maxUpdateSeqNo <= _lastSyncedSeqNo) {
+    // if noone has updated any counters etc. since we were here last,
+    // there is no need to do anything!
+    return Result();
+  }
+
+  // ok, when we are here, we will write out something back to the database
 
   std::unordered_map<uint64_t, CMValue> copy;
   {  // block all updates
     WRITE_LOCKER(guard, _rwLock);
     copy = _counters;
   }
 
-  rocksdb::WriteOptions writeOptions;
   // fetch the seq number prior to any writes; this guarantees that we save
   // any subsequent updates in the WAL to replay if we crash in the middle
   auto seqNumber = _db->GetLatestSequenceNumber();
+ 
+  rocksdb::WriteOptions writeOptions;
   std::unique_ptr<rocksdb::Transaction> rtrx(
       _db->BeginTransaction(writeOptions));
 
-  VPackBuilder b;
+  // recycle our builder
+  _builder.clear();
+
   for (std::pair<uint64_t, CMValue> const& pair : copy) {
-    Result res = writeCounterValue(_syncedSeqNums, rtrx.get(), b, pair);
+    Result res = writeCounterValue(_syncedSeqNums, rtrx.get(), _builder, pair);
     if (res.fail()) {
       return res;
     }
@@ -391,14 +456,18 @@ Result RocksDBSettingsManager::sync(bool force) {
     seqNumber = std::min(seqNumber, writeResult.second);
   }
 
-  Result res = writeSettings(rtrx.get(), b, seqNumber);
+  Result res = writeSettings(rtrx.get(), _builder, seqNumber);
+
   if (res.fail()) {
     return res;
   }
 
   // we have to commit all counters in one batch
   auto s = rtrx->Commit();
+  
   if (s.ok()) {
+    _lastSyncedSeqNo = maxUpdateSeqNo;
+
     {
       WRITE_LOCKER(guard, _rwLock);
       _lastSync = seqNumber;
@@ -467,6 +536,7 @@ void RocksDBSettingsManager::loadSettings() {
 
 void RocksDBSettingsManager::loadIndexEstimates() {
   WRITE_LOCKER(guard, _rwLock);
+
   RocksDBKeyBounds bounds = RocksDBKeyBounds::IndexEstimateValues();
 
   auto cf = RocksDBColumnFamily::definitions();
@@ -481,16 +551,16 @@ void RocksDBSettingsManager::loadIndexEstimates() {
     uint64_t lastSeqNumber =
         rocksutils::uint64FromPersistent(iter->value().data());
 
-    StringRef estimateSerialisation(iter->value().data() + sizeof(uint64_t),
+    StringRef estimateSerialization(iter->value().data() + sizeof(uint64_t),
                                     iter->value().size() - sizeof(uint64_t));
     // If this hits we have two estimates for the same index
     TRI_ASSERT(_estimators.find(objectId) == _estimators.end());
     try {
       if (RocksDBCuckooIndexEstimator<uint64_t>::isFormatSupported(
-              estimateSerialisation)) {
+              estimateSerialization)) {
         auto it = _estimators.emplace(
             objectId, std::make_unique<RocksDBCuckooIndexEstimator<uint64_t>>(
-                          lastSeqNumber, estimateSerialisation));
+                          lastSeqNumber, estimateSerialization));
         if (it.second) {
           auto estimator = it.first->second.get();
           LOG_TOPIC(TRACE, Logger::ENGINES)
@@ -543,26 +613,35 @@ RocksDBSettingsManager::stealIndexEstimator(uint64_t objectId) {
   std::unique_ptr<RocksDBCuckooIndexEstimator<uint64_t>> res;
 
   WRITE_LOCKER(guard, _rwLock);
+
   auto it = _estimators.find(objectId);
+  
   if (it != _estimators.end()) {
     // We swap out the stored estimate in order to move it to the caller
     res.swap(it->second);
     // Drop the now empty estimator
     _estimators.erase(objectId);
   }
+  
   return res;
 }
 
 uint64_t RocksDBSettingsManager::stealKeyGenerator(uint64_t objectId) {
-  WRITE_LOCKER(guard, _rwLock);
   uint64_t res = 0;
-  auto it = _generators.find(objectId);
-  if (it != _generators.end()) {
-    // We swap out the stored estimate in order to move it to the caller
-    res = it->second;
-    // Drop the now empty estimator
-    _generators.erase(objectId);
+
+  {
+    WRITE_LOCKER(guard, _rwLock);
+    
+    auto it = _generators.find(objectId);
+
+    if (it != _generators.end()) {
+      // We swap out the stored generator state in order to move it to the caller
+      res = it->second;
+      // we are now not responsible for the generator anymore
+      _generators.erase(objectId);
+    }
   }
+  
   return res;
 }
 
 
@@ -143,58 +143,59 @@ class RocksDBSettingsManager {
   rocksdb::SequenceNumber earliestSeqNeeded() const;
 
  private:
+  /// bump up the value of the last rocksdb::SequenceNumber we have seen
+  /// and that is pending a sync update
+  void setMaxUpdateSequenceNumber(rocksdb::SequenceNumber seqNo);
+  
   void loadCounterValues();
   void loadSettings();
   void loadIndexEstimates();
   void loadKeyGenerators();
 
   bool lockForSync(bool force);
 
-  //////////////////////////////////////////////////////////////////////////////
+  /// @brief a reusable builder, used inside sync() to serialize objects
+  arangodb::velocypack::Builder _builder;
+
   /// @brief counter values
-  //////////////////////////////////////////////////////////////////////////////
   std::unordered_map<uint64_t, CMValue> _counters;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief Key generator container
-  //////////////////////////////////////////////////////////////////////////////
   std::unordered_map<uint64_t, uint64_t> _generators;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief Index Estimator contianer.
   ///        Note the elements in this container will be moved into the
   ///        index classes and are only temporarily stored here during recovery.
-  //////////////////////////////////////////////////////////////////////////////
   std::unordered_map<uint64_t,
                      std::unique_ptr<RocksDBCuckooIndexEstimator<uint64_t>>>
       _estimators;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief synced sequence numbers
-  //////////////////////////////////////////////////////////////////////////////
   std::unordered_map<uint64_t, rocksdb::SequenceNumber> _syncedSeqNums;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief last sync sequence number
-  //////////////////////////////////////////////////////////////////////////////
   rocksdb::SequenceNumber _lastSync;
-
-  //////////////////////////////////////////////////////////////////////////////
+  
   /// @brief currently syncing
-  //////////////////////////////////////////////////////////////////////////////
   std::atomic<bool> _syncing;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief rocksdb instance
-  //////////////////////////////////////////////////////////////////////////////
   rocksdb::TransactionDB* _db;
 
-  //////////////////////////////////////////////////////////////////////////////
   /// @brief protect _syncing and _counters
-  //////////////////////////////////////////////////////////////////////////////
   mutable basics::ReadWriteLock _rwLock;
 
   TRI_voc_tick_t _initialReleasedTick;
+  
+  /// @brief the maximum sequence number that we have encountered
+  /// when updating a counter value
+  std::atomic<rocksdb::SequenceNumber> _maxUpdateSeqNo;
+  
+  /// @brief the last maximum sequence number we stored when we last synced
+  /// all counters back to persistent storage
+  /// if this is identical to _maxUpdateSeqNo, we do not need to write
+  /// back any counter values to disk and can save I/O
+  rocksdb::SequenceNumber _lastSyncedSeqNo;
 };
 }  // namespace arangodb