From 467088b8af209a43513cfa675d4412c3cd3aeeab Mon Sep 17 00:00:00 2001
From: Max Neunhoeffer <max@arangodb.com>
Date: Wed, 22 Mar 2017 15:24:01 +0100
Subject: [PATCH] Fix locking of shards in TraverserEngines.

---
 arangod/Aql/ExecutionEngine.cpp                | 16 +++++++++++++---
 arangod/Cluster/TraverserEngine.cpp            | 10 ++++++----
 arangod/RestHandler/RestVocbaseBaseHandler.cpp | 12 ++++++++++--
 3 files changed, 29 insertions(+), 9 deletions(-)

diff --git a/arangod/Aql/ExecutionEngine.cpp b/arangod/Aql/ExecutionEngine.cpp
index 92d46db241..1db1c1b60a 100644
--- a/arangod/Aql/ExecutionEngine.cpp
+++ b/arangod/Aql/ExecutionEngine.cpp
@@ -1007,7 +1007,14 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
       if (!shardSet.empty()) {
         arangodb::CoordTransactionID coordTransactionID = TRI_NewTickServer();
         std::unordered_map<std::string, std::string> headers;
-
+        std::string shardList;
+        for (auto const& shard : shardSet) {
+          if (!shardList.empty()) {
+            shardList += ",";
+          }
+          shardList += shard;
+        }
+        headers["X-Arango-Nolock"] = shardList;  // Prevent locking
         auto res = cc->syncRequest("", coordTransactionID, "server:" + list.first,
                                    RequestType::POST, url, engineInfo.toJson(),
                                    headers, 30.0);
@@ -1018,7 +1025,7 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
             message += std::string(" : ") + res->errorMessage;
           }
           THROW_ARANGO_EXCEPTION_MESSAGE(
-              TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED, message);
+              TRI_ERROR_CLUSTER_BACKEND_UNAVAILABLE, message);
         } else {
           // Only if the result was successful we will get here
           arangodb::basics::StringBuffer& body = res->result->getBody();
@@ -1028,7 +1035,7 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
           VPackSlice resultSlice = builder->slice();
           if (!resultSlice.isNumber()) {
             THROW_ARANGO_EXCEPTION_MESSAGE(
-                TRI_ERROR_INTERNAL, "got unexpected response from engine lock request");
+                TRI_ERROR_INTERNAL, "got unexpected response from engine build request");
           }
           auto engineId = resultSlice.getNumericValue<traverser::TraverserEngineID>();
           TRI_ASSERT(engineId != 0);
@@ -1209,6 +1216,9 @@ ExecutionEngine* ExecutionEngine::instantiateFromPlan(
           engine->_lockedShards = new std::unordered_set<std::string>();
           engine->_previouslyLockedShards = nullptr;
         }
+        // Note that it is crucial that this is a map and not an unordered_map,
+        // because we need to guarantee the order of locking by using
+        // alphabetical order on the shard names!
         std::map<std::string, std::pair<std::string, bool>> forLocking;
         for (auto& q : inst.get()->queryIds) {
           std::string theId = q.first;
diff --git a/arangod/Cluster/TraverserEngine.cpp b/arangod/Cluster/TraverserEngine.cpp
index dedecb709e..778be45b39 100644
--- a/arangod/Cluster/TraverserEngine.cpp
+++ b/arangod/Cluster/TraverserEngine.cpp
@@ -94,9 +94,12 @@ BaseTraverserEngine::BaseTraverserEngine(TRI_vocbase_t* vocbase,
   auto params = std::make_shared<VPackBuilder>();
   auto opts = std::make_shared<VPackBuilder>();
 
-  _trx = new aql::AqlTransaction(
-      arangodb::transaction::StandaloneContext::Create(vocbase),
-      _collections.collections(), false);
+  _trx = new arangodb::AqlTransaction(
+      arangodb::StandaloneTransactionContext::Create(vocbase),
+      _collections.collections(), true);
+  // true here as last argument is crucial: it leads to the fact that the
+  // created transaction is considered a "MAIN" part and will not switch
+  // off collection locking completely!
   _query = new aql::Query(true, vocbase, "", 0, params, opts, aql::PART_DEPENDENT);
   _query->injectTransaction(_trx);
 
@@ -112,7 +115,6 @@ BaseTraverserEngine::BaseTraverserEngine(TRI_vocbase_t* vocbase,
     }
   }
 
-
   _trx->begin(); // We begin the transaction before we lock.
                  // We also setup indexes before we lock.
 }
diff --git a/arangod/RestHandler/RestVocbaseBaseHandler.cpp b/arangod/RestHandler/RestVocbaseBaseHandler.cpp
index 17b1533da2..b36598b167 100644
--- a/arangod/RestHandler/RestVocbaseBaseHandler.cpp
+++ b/arangod/RestHandler/RestVocbaseBaseHandler.cpp
@@ -627,8 +627,16 @@ void RestVocbaseBaseHandler::prepareExecute() {
   std::string const& shardId = _request->header("x-arango-nolock", found);
 
   if (found) {
-    _nolockHeaderSet =
-        new std::unordered_set<std::string>{std::string(shardId)};
+    _nolockHeaderSet = new std::unordered_set<std::string>();
+    // Split value at commas, if there are any, otherwise take full value:
+    size_t pos = shardId.find(',');
+    size_t oldpos = 0;
+    while (pos != std::string::npos) {
+      _nolockHeaderSet->emplace(shardId.substr(oldpos, pos - oldpos));
+      oldpos = pos + 1;
+      pos = shardId.find(',', oldpos);
+    }
+    _nolockHeaderSet->emplace(shardId.substr(oldpos));
     CollectionLockState::_noLockHeaders = _nolockHeaderSet;
   }
 }