changes

arangodb · jsteemann · Dec 21, 2018 · Dec 3, 2018 · Dec 3, 2018 · Dec 4, 2018
commit db46a40334f50b706f1510f21cd37230b96c7a4a
diff --git a/arangod/Cluster/ClusterInfo.cpp b/arangod/Cluster/ClusterInfo.cpp
@@ -2433,14 +2433,11 @@ int ClusterInfo::ensureIndexCoordinator(
 
   // check index id
   uint64_t iid = 0;
-
   VPackSlice const idSlice = slice.get(StaticStrings::IndexId);
-  if (idSlice.isString()) {
-    // use predefined index id
+  if (idSlice.isString()) { // use predefined index id
     iid = arangodb::basics::StringUtils::uint64(idSlice.copyString());
   }
-  if (iid == 0) {
-    // no id set, create a new one!
+  if (iid == 0) { // no id set, create a new one!
     iid = uniqid();
   }
   std::string const idString = arangodb::basics::StringUtils::itoa(iid);

diff --git a/arangod/ClusterEngine/ClusterCollection.cpp b/arangod/ClusterEngine/ClusterCollection.cpp
@@ -368,14 +368,12 @@ std::shared_ptr<Index> ClusterCollection::createIndex(
   WRITE_LOCKER(guard, _exclusiveLock);
   std::shared_ptr<Index> idx;
 
-  {
-    WRITE_LOCKER(guard, _indexesLock);
-    idx = findIndex(info, _indexes);
-    if (idx) {
-      created = false;
-      // We already have this index.
-      return idx;
-    }
+  WRITE_LOCKER(guard2, _indexesLock);
+  idx = lookupIndex(info);
+  if (idx) {
+    created = false;
+    // We already have this index.
+    return idx;
   }
 
   StorageEngine* engine = EngineSelectorFeature::ENGINE;

diff --git a/arangod/RocksDBEngine/RocksDBBuilderIndex.cpp b/arangod/RocksDBEngine/RocksDBBuilderIndex.cpp
@@ -71,30 +71,19 @@ Result RocksDBBuilderIndex::insertInternal(transaction::Methods* trx, RocksDBMet
                                            arangodb::velocypack::Slice const& slice,
                                            OperationMode mode) {
   Result r = _wrapped->insertInternal(trx, mthd, documentId, slice, mode);
-  if (!r.ok() && !_hasError.load(std::memory_order_acquire)) {
-    _errorResult = r;
-    _hasError.store(true, std::memory_order_release);
+  if (r.is(TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED) ||
+      r.is(TRI_ERROR_LOCK_TIMEOUT) ||
+      r.is(TRI_ERROR_DEADLOCK) ||
+      r.is(TRI_ERROR_ARANGO_CONFLICT)) {
+    bool expected = false;
+    if (!r.ok() && !_hasError.compare_exchange_strong(expected, true, std::memory_order_release)) {
+      std::lock_guard<std::mutex> guard(_errorMutex);
+      _errorResult = r;
+    }
   }
   return Result();
 }
 
-//Result RocksDBBuilderIndex::updateInternal(transaction::Methods* trx, RocksDBMethods* mthd,
-//                                           LocalDocumentId const& oldDocumentId,
-//                                           arangodb::velocypack::Slice const& oldDoc,
-//                                           LocalDocumentId const& newDocumentId,
-//                                           arangodb::velocypack::Slice const& newDoc,
-//                                           OperationMode mode) {
-//  // It is illegal to call this method on the primary index
-//  // RocksDBPrimaryIndex must override this method accordingly
-//  TRI_ASSERT(type() != TRI_IDX_TYPE_PRIMARY_INDEX);
-//
-//  Result res = removeInternal(trx, mthd, oldDocumentId, oldDoc, mode);
-//  if (!res.ok()) {
-//    return res;
-//  }
-//  return insertInternal(trx, mthd, newDocumentId, newDoc, mode);
-//}
-
 /// remove index elements and put it in the specified write batch.
 Result RocksDBBuilderIndex::removeInternal(transaction::Methods* trx, RocksDBMethods* mthd,
                                            LocalDocumentId const& documentId,
@@ -112,9 +101,15 @@ Result RocksDBBuilderIndex::removeInternal(transaction::Methods* trx, RocksDBMet
   }
 
   Result r = _wrapped->removeInternal(trx, mthd, documentId, slice, mode);
-  if (!r.ok() && !_hasError.load(std::memory_order_acquire)) {
-    _errorResult = r;
-    _hasError.store(true, std::memory_order_release);
+  if (r.is(TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED) ||
+      r.is(TRI_ERROR_LOCK_TIMEOUT) ||
+      r.is(TRI_ERROR_DEADLOCK) ||
+      r.is(TRI_ERROR_ARANGO_CONFLICT)) {
+    bool expected = false;
+    if (!r.ok() && !_hasError.compare_exchange_strong(expected, true, std::memory_order_release)) {
+      std::lock_guard<std::mutex> guard(_errorMutex);
+      _errorResult = r;
+    }
   }
   return Result();
 }
@@ -128,6 +123,12 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
   //  3. Avoid conflicts on unique index keys by using rocksdb::Transaction snapshot conflict checking
   //  4. Supress unique constraint violations / conflicts or client drivers
 
+  auto lockedDocsGuard = scopeGuard([&] { // clear all the processed documents
+    std::lock_guard<std::mutex> guard(_lockedDocsMutex);
+    _lockedDocs.clear();
+    _lockedDocsCond.notify_all();
+  });
+
   // fillindex can be non transactional, we just need to clean up
   RocksDBEngine* engine = rocksutils::globalRocksEngine();
   RocksDBCollection* rcoll = static_cast<RocksDBCollection*>(_collection.getPhysical());
@@ -177,7 +178,7 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
   } else {
     rtrx->DisableIndexing(); // we never check for existing index keys
   }
-  RocksDBSubTrxMethods batched(state, rtrx.get());
+  RocksDBSideTrxMethods batched(state, rtrx.get());
 
   RocksDBIndex* internal = _wrapped.get();
   TRI_ASSERT(internal != nullptr);
@@ -186,6 +187,7 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
   while (it->Valid() && it->key().compare(upper) < 0) {
 
     if (_hasError.load(std::memory_order_acquire)) {
+      std::lock_guard<std::mutex> guard(_errorMutex);
       res = _errorResult; // a Writer got an error
       break;
     }
@@ -197,7 +199,7 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
         continue;
       }
       std::lock_guard<std::mutex> guard2(_lockedDocsMutex);
-      _lockedDocs.insert(docId.id());// must be done under _removedDocsMutex
+      _lockedDocs.insert(docId.id());
     }
 
     res = internal->insertInternal(&trx, &batched, docId,
@@ -215,7 +217,7 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
         break;
       }
       { // clear all the processed documents
-        std::lock_guard<std::mutex> guard2(_lockedDocsMutex);
+        std::lock_guard<std::mutex> guard(_lockedDocsMutex);
         _lockedDocs.clear();
         _lockedDocsCond.notify_all();
       }
@@ -240,11 +242,6 @@ arangodb::Result RocksDBBuilderIndex::fillIndexBackground(std::function<void()>
     res = trx.commit(); // required to commit selectivity estimates
   }
 
-  // clear all the processed documents
-  std::lock_guard<std::mutex> guard2(_lockedDocsMutex);
-  _lockedDocs.clear();
-  _lockedDocsCond.notify_all();
-
   return res;
 }
 
@@ -331,15 +328,12 @@ static arangodb::Result fillIndexFast(transaction::Methods& trx,
 
 /// non-transactional: fill index with existing documents
 /// from this collection
-arangodb::Result RocksDBBuilderIndex::fillIndex(std::function<void()> const& unlock) {
-//  TRI_ASSERT(trx.state()->collection(_collection.id(), AccessMode::Type::WRITE));
+arangodb::Result RocksDBBuilderIndex::fillIndexFast() {
 
-#if 0
   RocksDBCollection* coll = static_cast<RocksDBCollection*>(_collection.getPhysical());
-  unlock(); // we do not need the outer lock
   SingleCollectionTransaction trx(transaction::StandaloneContext::Create(_collection.vocbase()),
                                   _collection, AccessMode::Type::EXCLUSIVE);
-  res = trx.begin();
+  Result res = trx.begin();
   if (!res.ok()) {
     THROW_ARANGO_EXCEPTION(res);
   }
@@ -355,6 +349,4 @@ arangodb::Result RocksDBBuilderIndex::fillIndex(std::function<void()> const& unl
     rocksdb::WriteBatch batch(32 * 1024 * 1024);
     return ::fillIndexFast<rocksdb::WriteBatch, RocksDBBatchedMethods>(trx, this, coll, batch);
   }
-#endif
-    return fillIndexBackground(unlock);
 }
diff --git a/arangod/RocksDBEngine/RocksDBBuilderIndex.h b/arangod/RocksDBEngine/RocksDBBuilderIndex.h
@@ -30,6 +30,9 @@
 
 namespace arangodb {
 
+/// Dummy index class that contains the logic to build indexes
+/// without an exclusive lock. It wraps the actual index implementation
+/// and adds some required synchronization logic on top
 class RocksDBBuilderIndex final : public arangodb::RocksDBIndex {
 
  public:
@@ -89,13 +92,6 @@ class RocksDBBuilderIndex final : public arangodb::RocksDBIndex {
                         LocalDocumentId const& documentId,
                         arangodb::velocypack::Slice const&,
                         OperationMode mode) override;
-//
-//  Result updateInternal(transaction::Methods* trx, RocksDBMethods*,
-//                        LocalDocumentId const& oldDocumentId,
-//                        arangodb::velocypack::Slice const& oldDoc,
-//                        LocalDocumentId const& newDocumentId,
-//                        velocypack::Slice const& newDoc,
-//                        OperationMode mode) override;
 
   /// remove index elements and put it in the specified write batch.
   Result removeInternal(transaction::Methods* trx, RocksDBMethods*,
@@ -116,17 +112,18 @@ class RocksDBBuilderIndex final : public arangodb::RocksDBIndex {
     _wrapped->recalculateEstimates();
   }
 
-  /// @brief fill the index
-  /// @param unlock will be called when the index lock can be released
-  Result fillIndex(std::function<void()> const& unlock);
-
-private:
+  /// @brief fill index, will exclusively lock the collection
+  Result fillIndexFast();
 
+  /// @brief fill the index, assume already locked exclusively
+  /// @param unlock called when collection lock can be released
   Result fillIndexBackground(std::function<void()> const& unlock);
 
  private:
   std::shared_ptr<arangodb::RocksDBIndex> _wrapped;
+
   std::atomic<bool> _hasError;
+  std::mutex _errorMutex;
   Result _errorResult;
 
   std::mutex _removedDocsMutex;

diff --git a/arangod/RocksDBEngine/RocksDBCollection.cpp b/arangod/RocksDBEngine/RocksDBCollection.cpp
@@ -312,8 +312,8 @@ void RocksDBCollection::prepareIndexes(
 }
 
 std::shared_ptr<Index> RocksDBCollection::createIndex(
-    arangodb::velocypack::Slice const& info, bool restore,
-    bool& created) {
+    arangodb::velocypack::Slice const& info,
+    bool restore, bool& created) {
   TRI_ASSERT(info.isObject());
   Result res;
 
@@ -345,7 +345,7 @@ std::shared_ptr<Index> RocksDBCollection::createIndex(
   }
 
   RocksDBEngine* engine = static_cast<RocksDBEngine*>(EngineSelectorFeature::ENGINE);
-
+  
   // Step 2. We are sure that we do not have an index of this type.
   // We also hold the lock. Create it
   const bool generateKey = !restore;
@@ -367,11 +367,10 @@ std::shared_ptr<Index> RocksDBCollection::createIndex(
   }
 
   auto buildIdx = std::make_shared<RocksDBBuilderIndex>(std::static_pointer_cast<RocksDBIndex>(idx));
-
+  
   // Step 3. add index to collection entry (for removal after a crash)
-  if (!engine->inRecovery()) {
-    // read collection info from database
-    RocksDBKey key;
+  if (!engine->inRecovery()) { // manually modify collection entry, other methods need lock
+    RocksDBKey key; // read collection info from database
     key.constructCollection(_logicalCollection.vocbase().id(), _logicalCollection.id());
     rocksdb::PinnableSlice value;
     rocksdb::Status s = engine->db()->Get(rocksdb::ReadOptions(),
@@ -399,13 +398,21 @@ std::shared_ptr<Index> RocksDBCollection::createIndex(
                                                 RocksDBLogValue::Empty());
     }
   }
+
+  bool inBackground = basics::VelocyPackHelper::getBooleanValue(
+                        info, StaticStrings::IndexInBackground, false);
 
   // Step 4. fill index
   if (res.ok()) {
-    _indexes.emplace_back(buildIdx); // add index to indexes list
-    res = buildIdx->fillIndex([&] {
+    if (inBackground) { // allow concurrent inserts into index
+      _indexes.emplace_back(buildIdx);
+      res = buildIdx->fillIndexBackground([&] {
+        unlockGuard.fire();
+      });
+    } else {
       unlockGuard.fire();
-    });
+      res = buildIdx->fillIndexFast(); // will lock again internally
+    }
   }
 
   // Step 5. cleanup

diff --git a/arangod/RocksDBEngine/RocksDBIndex.cpp b/arangod/RocksDBEngine/RocksDBIndex.cpp
@@ -53,8 +53,10 @@ using namespace arangodb::rocksutils;
 
 uint64_t const arangodb::RocksDBIndex::ESTIMATOR_SIZE = 4096;
 
-inline static uint64_t ensureObjectId(uint64_t oid) {
-  return (oid != 0) ? oid : TRI_NewTickServer();
+namespace {
+  inline uint64_t ensureObjectId(uint64_t oid) {
+    return (oid != 0) ? oid : TRI_NewTickServer();
+  }
 }
 
 RocksDBIndex::RocksDBIndex(
@@ -68,7 +70,7 @@ RocksDBIndex::RocksDBIndex(
     bool useCache
 )
     : Index(id, collection, attributes, unique, sparse),
-      _objectId(ensureObjectId(objectId)),
+      _objectId(::ensureObjectId(objectId)),
       _cf(cf),
       _cache(nullptr),
       _cachePresent(false),
@@ -94,7 +96,7 @@ RocksDBIndex::RocksDBIndex(
     bool useCache
 )
     : Index(id, collection, info),
-      _objectId(ensureObjectId(basics::VelocyPackHelper::stringUInt64(info.get("objectId")))),
+      _objectId(::ensureObjectId(basics::VelocyPackHelper::stringUInt64(info.get("objectId")))),
       _cf(cf),
       _cache(nullptr),
       _cachePresent(false),