Merge branch 'spdvpk' of ssh://github.com/ArangoDB/ArangoDB into spdvpk

arangod/Aql/IndexBlock.cpp

@@ -42,6 +42,7 @@ using Json = arangodb::basics::Json;
 IndexBlock::IndexBlock(ExecutionEngine* engine, IndexNode const* en)
     : ExecutionBlock(engine, en),
       _collection(en->collection()),
+      _result(std::make_shared<OperationResult>(TRI_ERROR_NO_ERROR)),
       _posInDocs(0),
       _currentIndex(0),
       _indexes(en->getIndexes()),
@@ -377,17 +378,17 @@ bool IndexBlock::readIndex(size_t atMost) {
   bool isReverse = (static_cast<IndexNode const*>(getPlanNode()))->_reverse;
   bool isLastIndex = (_currentIndex == lastIndexNr && !isReverse) ||
                      (_currentIndex == 0 && isReverse);
-  auto result = std::make_shared<OperationResult>(TRI_ERROR_NO_ERROR);
+  auto _result = std::make_shared<OperationResult>(TRI_ERROR_NO_ERROR);
   while (_cursor != nullptr) {
     if (!_cursor->hasMore()) {
       startNextCursor();
       continue;
     }
-    _cursor->getMore(result, atMost, true);
+    _cursor->getMore(_result, atMost, true);
-    if (result->failed()) {
+    if (_result->failed()) {
-      THROW_ARANGO_EXCEPTION(result->code);
+      THROW_ARANGO_EXCEPTION(_result->code);
     }
-    VPackSlice slice = result->slice();
+    VPackSlice slice = _result->slice();
     TRI_ASSERT(slice.isArray());
     size_t length = static_cast<size_t>(slice.length());
 

arangod/Aql/IndexBlock.h

@@ -130,6 +130,12 @@ class IndexBlock : public ExecutionBlock {
 
   Collection const* _collection;
 
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief document result
+  //////////////////////////////////////////////////////////////////////////////
+
+  std::shared_ptr<OperationResult> _result;
+  
   //////////////////////////////////////////////////////////////////////////////
   /// @brief document buffer
   //////////////////////////////////////////////////////////////////////////////

arangod/Cluster/ApplicationCluster.cpp

@@ -355,16 +355,15 @@ bool ApplicationCluster::open() {
     return true;
   }
   
-  ServerState::RoleEnum role = ServerState::instance()->getRole();
-
-  // tell the agency that we are ready
-  {
   AgencyComm comm;
   AgencyCommResult result;
   
+  bool success; 
+  do {
     AgencyCommLocker locker("Current", "WRITE");
     
-    if (locker.successful()) {
+    success = locker.successful();
+    if (success) {
       VPackBuilder builder;
       try {
         VPackObjectBuilder b(&builder);
@@ -382,68 +381,21 @@ bool ApplicationCluster::open() {
       LOG(FATAL) << "unable to register server in agency: http code: " << result.httpCode() << ", body: " << result.body(); FATAL_ERROR_EXIT();
     }
     
+    if (success) {
+      break;
+    }
+    sleep(1);
+  } while (true);
+
+  ServerState::RoleEnum role = ServerState::instance()->getRole();
+
   if (role == ServerState::ROLE_COORDINATOR) {
-      VPackBuilder builder;
-      try {
-        builder.add(VPackValue("none"));
-      } catch (...) {
-        locker.unlock();
-        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
-      }
-
     ServerState::instance()->setState(ServerState::STATE_SERVING);
-
-      // register coordinator
-      AgencyCommResult result =
-          comm.setValue("Current/Coordinators/" + _myId, builder.slice(), 0.0);
-
-      if (!result.successful()) {
-        locker.unlock();
-        LOG(FATAL) << "unable to register coordinator in agency"; FATAL_ERROR_EXIT();
-      }
   } else if (role == ServerState::ROLE_PRIMARY) {
-      VPackBuilder builder;
-      try {
-        builder.add(VPackValue("none"));
-      } catch (...) {
-        locker.unlock();
-        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
-      }
-
     ServerState::instance()->setState(ServerState::STATE_SERVINGASYNC);
-
-      // register server
-      AgencyCommResult result =
-          comm.setValue("Current/DBServers/" + _myId, builder.slice(), 0.0);
-
-      if (!result.successful()) {
-        locker.unlock();
-        LOG(FATAL) << "unable to register db server in agency"; FATAL_ERROR_EXIT();
-      }
   } else if (role == ServerState::ROLE_SECONDARY) {
-      std::string keyName = std::string("\"") + _myId + std::string("\"");
-      VPackBuilder builder;
-      try {
-        builder.add(VPackValue(keyName));
-      } catch (...) {
-        locker.unlock();
-        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
-      }
-
     ServerState::instance()->setState(ServerState::STATE_SYNCING);
-
-      // register server
-      AgencyCommResult result = comm.setValue(
-          "Current/DBServers/" + ServerState::instance()->getPrimaryId(),
-          builder.slice(), 0.0);
-
-      if (!result.successful()) {
-        locker.unlock();
-        LOG(FATAL) << "unable to register secondary db server in agency"; FATAL_ERROR_EXIT();
   }
-    }
-  }
-
   return true;
 }
 

arangod/Cluster/ServerState.cpp

@@ -37,6 +37,12 @@ using namespace arangodb::basics;
 /// running
 ////////////////////////////////////////////////////////////////////////////////
 
+static bool isClusterRole(ServerState::RoleEnum role) {
+  return (role == ServerState::ROLE_PRIMARY ||
+          role == ServerState::ROLE_SECONDARY ||
+          role == ServerState::ROLE_COORDINATOR);
+}
+
 static ServerState Instance;
 
 ServerState::ServerState()
@@ -172,10 +178,25 @@ void ServerState::setAuthentication(std::string const& username,
 
 std::string ServerState::getAuthentication() { return _authentication; }
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief find and set our role
+////////////////////////////////////////////////////////////////////////////////
+void ServerState::findAndSetRoleBlocking() {
+  while (true) {
+    auto role = determineRole(_localInfo, _id);
+    std::string roleString = roleToString(role);
+    LOG(DEBUG) << "Found my role: " << roleString;
+
+    if (storeRole(role)) {
+      break;
+    }
+    sleep(1);
+  }
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief flush the server state (used for testing)
 ////////////////////////////////////////////////////////////////////////////////
-
 void ServerState::flush() {
   {
     WRITE_LOCKER(writeLocker, _lock);
@@ -187,7 +208,7 @@ void ServerState::flush() {
     _address = ClusterInfo::instance()->getTargetServerEndpoint(_id);
   }
   
-  storeRole(determineRole(_localInfo, _id));
+  findAndSetRoleBlocking();
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -228,9 +249,7 @@ bool ServerState::isDBServer(ServerState::RoleEnum role) {
 bool ServerState::isRunningInCluster() {
   auto role = loadRole();
   
-  return (role == ServerState::ROLE_PRIMARY ||
+  return isClusterRole(role);
-          role == ServerState::ROLE_SECONDARY ||
-          role == ServerState::ROLE_COORDINATOR);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -272,15 +291,8 @@ ServerState::RoleEnum ServerState::getRole() {
     LOG(DEBUG) << "Have stored " << builder.slice().toJson() << " under Current/NewServers/" << _localInfo << " in agency.";
   }
 
-  // role not yet set
+  findAndSetRoleBlocking();
-  role = determineRole(info, id);
+  return loadRole();
-  std::string roleString = roleToString(role);
-
-  LOG(DEBUG) << "Found my role: " << roleString;
-
-  storeRole(role);
-
-  return role;
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -351,8 +363,8 @@ bool ServerState::registerWithRole(ServerState::RoleEnum role) {
   }
   
   _id = id;
-  storeRole(role);
 
+  findAndSetRoleBlocking();
   LOG(DEBUG) << "We successfully announced ourselves as " << roleToString(role) << " and our id is " << id;
   
   return true;
@@ -779,7 +791,9 @@ bool ServerState::redetermineRole() {
   RoleEnum oldRole = loadRole();
   if (role != oldRole) {
     LOG(INFO) << "Changed role to: " << roleString;
-    storeRole(role);
+    if (!storeRole(role)) {
+      return false;
+    }
     return true;
   }
   if (_idOfPrimary != saveIdOfPrimary) {
@@ -807,20 +821,11 @@ ServerState::RoleEnum ServerState::determineRole(std::string const& info,
     setId(id);
   }
   
-  ServerState::RoleEnum role = checkServersList(id);
+  ServerState::RoleEnum role = checkCoordinatorsList(id);
-  ServerState::RoleEnum role2 = checkCoordinatorsList(id);
-
   if (role == ServerState::ROLE_UNDEFINED) {
-    // role is still unknown. check if we are a coordinator
+    role = checkServersList(id);
-    role = role2;
-  } else {
-    // we are a primary or a secondary.
-    // now we double-check that we are not a coordinator as well
-    if (role2 != ServerState::ROLE_UNDEFINED) {
-      role = ServerState::ROLE_UNDEFINED;
   }
-  }
+  // mop: role might still be undefined
-
   return role;
 }
 
@@ -1056,3 +1061,81 @@ ServerState::RoleEnum ServerState::checkServersList(std::string const& id) {
 
   return role;
 }
+
+//////////////////////////////////////////////////////////////////////////////
+/// @brief store the server role
+//////////////////////////////////////////////////////////////////////////////
+bool ServerState::storeRole(RoleEnum role)
+{
+  if (isClusterRole(role)) {
+    AgencyComm comm;
+    AgencyCommResult result;
+    std::unique_ptr<AgencyCommLocker> locker;
+
+    locker.reset(new AgencyCommLocker("Current", "WRITE"));
+    if (!locker->successful()) {
+      return false;
+    }
+
+    if (role == ServerState::ROLE_COORDINATOR) {
+      VPackBuilder builder;
+      try {
+        builder.add(VPackValue("none"));
+      } catch (...) {
+        locker->unlock();
+        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
+      }
+
+      // register coordinator
+      AgencyCommResult result =
+        comm.setValue("Current/Coordinators/" + _id, builder.slice(), 0.0);
+
+      if (!result.successful()) {
+        locker->unlock();
+        LOG(FATAL) << "unable to register coordinator in agency"; FATAL_ERROR_EXIT();
+      }
+    } else if (role == ServerState::ROLE_PRIMARY) {
+      VPackBuilder builder;
+      try {
+        builder.add(VPackValue("none"));
+      } catch (...) {
+        locker->unlock();
+        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
+      }
+
+      // register server
+      AgencyCommResult result =
+        comm.setValue("Current/DBServers/" + _id, builder.slice(), 0.0);
+
+      if (!result.successful()) {
+        locker->unlock();
+        LOG(FATAL) << "unable to register db server in agency"; FATAL_ERROR_EXIT();
+      }
+    } else if (role == ServerState::ROLE_SECONDARY) {
+      std::string keyName = _id;
+      VPackBuilder builder;
+      try {
+        builder.add(VPackValue(keyName));
+      } catch (...) {
+        locker->unlock();
+        LOG(FATAL) << "out of memory"; FATAL_ERROR_EXIT();
+      }
+      
+      // register server
+      AgencyCommResult result = comm.casValue(
+          "Current/DBServers/" + ServerState::instance()->getPrimaryId(),
+          builder.slice(),
+          true,
+          0.0,
+          0.0);
+
+      if (!result.successful()) {
+        locker->unlock();
+        // mop: fail gracefully (allow retry)
+        return false;
+      }
+    }
+  }
+  _role.store(role, std::memory_order_release);
+  return true;
+}

arangod/Cluster/ServerState.h

@@ -376,12 +376,14 @@ class ServerState {
   }
   
   //////////////////////////////////////////////////////////////////////////////
-  /// @brief atomically stores the server role
+  /// @brief determine role and save role blocking
   //////////////////////////////////////////////////////////////////////////////
+  void findAndSetRoleBlocking();
 
-  void storeRole(RoleEnum role) {
+  //////////////////////////////////////////////////////////////////////////////
-    _role.store(role, std::memory_order_release);
+  /// @brief store the server role
-  }
+  //////////////////////////////////////////////////////////////////////////////
+  bool storeRole(RoleEnum role);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief determine the server role

js/actions/api-cluster.js

@@ -34,6 +34,7 @@
 var actions = require("@arangodb/actions");
 var cluster = require("@arangodb/cluster");
 var internal = require("internal");
+var _ = require("lodash");
 
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -921,42 +922,34 @@ actions.defineHttp({
 function changeAllShardReponsibilities (oldServer, newServer) {
   // This is only called when we have the write lock and we "only" have to
   // make sure that either all or none of the shards are moved.
-  var l = ArangoAgency.get("Plan/Collections", true, false);
+  var collections = ArangoAgency.get("Plan/Collections", true, false);
-  var ll = Object.keys(l);
+  var done = {};
+  try {
+  Object.keys(collections).forEach(function(collectionKey) {
+    var collection = collections[collectionKey];
+    var old = _.cloneDeep(collection);
 
-  var i = 0;
+    Object.keys(collection.shards).forEach(function(shardKey) {
-  var c;
+      var servers = collection.shards[shardKey];
-  var oldShards = [];
+      collection.shards[shardKey] = servers.map(function(server) {
-  var shards;
+        if (server === oldServer) {
-  var names;
+          return newServer;
-  var j;
+        } else {
+          return server;
+        }
+      });
+    });
+    ArangoAgency.set(collectionKey, collection, 0);
+    done[collectionKey] = old;
+  });
+  } catch (e) {
+    // mop: rollback
     try {
-    while (i < ll.length) {
+      Object.keys(done).forEach(function(collectionKey) {
-      c = l[ll[i]];   // A collection entry
+        ArangoAgency.set(collectionKey, done[collectionKey], 0);
-      shards = c.shards;
+      });
-      names = Object.keys(shards);
+    } catch (e2) {
-      // Poor man's deep copy:
+      console.error("Got error during rolback", e2);
-      oldShards.push(JSON.parse(JSON.stringify(shards)));
-      for (j = 0; j < names.length; j++) {
-        if (shards[names[j]] === oldServer) {
-          shards[names[j]] = newServer;
-        }
-      }
-      ArangoAgency.set(ll[i], c, 0);
-      i += 1;
-    }
-  }
-  catch (e) {
-    i -= 1;
-    while (i >= 0) {
-      c = l[ll[i]];
-      c.shards = oldShards[i];
-      try {
-        ArangoAgency.set(ll[i], c, 0);
-      }
-      catch (e2) {
-      }
-      i -= 1;
     }
     throw e;
   }

js/client/modules/@arangodb/testing.js

@@ -2357,7 +2357,8 @@ testFuncs.arangob = function(options) {
         break;
       }
 
-      let args = benchTodo;
+      let args = _.clone(benchTodo);
+      delete args.transaction;
 
       if (options.hasOwnProperty('benchargs')) {
         args = _.extend(args, options.benchargs);

js/server/modules/@arangodb/cluster.js

@@ -541,7 +541,7 @@ function handleDatabaseChanges (plan) {
 /// @brief create collections if they exist in the plan but not locally
 ////////////////////////////////////////////////////////////////////////////////
 
-function createLocalCollections (plannedCollections, planVersion) {
+function createLocalCollections (plannedCollections, planVersion, takeOverResponsibility) {
   var ourselves = global.ArangoServerState.id();
   
   var createCollectionAgency = function (database, shard, collInfo, error) {
@@ -551,11 +551,16 @@ function createLocalCollections (plannedCollections, planVersion) {
                     indexes: collInfo.indexes,
                     servers: [ ourselves ],
                     planVersion: planVersion };
+
     global.ArangoAgency.set("Current/Collections/" + database + "/" + 
                             collInfo.planId + "/" + shard,
                             payload);
   };
   
+  // mop: just a function alias but this way one at least knows what it is supposed to do :S
+  var takeOver = createCollectionAgency;
+
+
   var db = require("internal").db;
   db._useDatabase("_system");
   var localDatabases = getLocalDatabases();
@@ -576,11 +581,9 @@ function createLocalCollections (plannedCollections, planVersion) {
           var localCollections = getLocalCollections();
 
           var collections = plannedCollections[database];
-          var collection;
 
           // diff the collections
-          for (collection in collections) {
+          Object.keys(collections).forEach(function(collection) {
-            if (collections.hasOwnProperty(collection)) {
             var collInfo = collections[collection];
             var shards = collInfo.shards;
             var shard;
@@ -592,6 +595,7 @@ function createLocalCollections (plannedCollections, planVersion) {
 
             for (shard in shards) {
               if (shards.hasOwnProperty(shard)) {
+                var didWrite = false;
                 if (shards[shard][0] === ourselves) {
                   // found a shard we are responsible for
 
@@ -628,6 +632,7 @@ function createLocalCollections (plannedCollections, planVersion) {
                     writeLocked({ part: "Current" },
                                 createCollectionAgency,
                                 [ database, shard, collInfo, error ]);
+                    didWrite = true;
                   }
                   else {
                     if (localCollections[shard].status !== collInfo.status) {
@@ -650,6 +655,7 @@ function createLocalCollections (plannedCollections, planVersion) {
                       writeLocked({ part: "Current" },
                                   createCollectionAgency,
                                   [ database, shard, collInfo, error ]);
+                      didWrite = true;
                     }
 
                     // collection exists, now compare collection properties
@@ -679,10 +685,16 @@ function createLocalCollections (plannedCollections, planVersion) {
                       writeLocked({ part: "Current" },
                                   createCollectionAgency,
                                   [ database, shard, collInfo, error ]);
+                      didWrite = true;
                     }
                   }
 
                   if (error.error) {
+                    if (takeOverResponsibility && !didWrite) {
+                      writeLocked({ part: "Current" },
+                                  takeOver,
+                                  [ database, shard, collInfo, error ]);
+                    }
                     continue;  // No point to look for properties and
                                // indices, if the creation has not worked
                   }
@@ -722,6 +734,7 @@ function createLocalCollections (plannedCollections, planVersion) {
                         writeLocked({ part: "Current" },
                                     createCollectionAgency,
                                     [ database, shard, collInfo, error ]);
+                        didWrite = true;
                       }
                     }
 
@@ -761,16 +774,23 @@ function createLocalCollections (plannedCollections, planVersion) {
                       writeLocked({ part: "Current" },
                                   createCollectionAgency,
                                   [ database, shard, collInfo, error ]);
+                      didWrite = true;
                     }
                   }
 
+                  if (takeOverResponsibility && !didWrite) {
+                    console.info("HMMMM WRITE");
+                    writeLocked({ part: "Current" },
+                        takeOver,
+                        [ database, shard, collInfo, error ]);
+                  }
                 }
               }
             }
             collInfo.id = save[0];
             collInfo.name = save[1];
-            }
+            
-          }
+          });
         }
         catch (err) {
           // always return to previous database
@@ -1062,20 +1082,20 @@ function synchronizeLocalFollowerCollections (plannedCollections) {
 /// @brief handle collection changes
 ////////////////////////////////////////////////////////////////////////////////
 
-function handleCollectionChanges (plan) {
+function handleCollectionChanges (plan, takeOverResponsibility) {
   var plannedCollections = getByPrefix3d(plan, "Plan/Collections/");
 
   var ok = true;
 
   try {
-    createLocalCollections(plannedCollections, plan["Plan/Version"]);
+    createLocalCollections(plannedCollections, plan["Plan/Version"], takeOverResponsibility);
     dropLocalCollections(plannedCollections);
     cleanupCurrentCollections(plannedCollections);
     synchronizeLocalFollowerCollections(plannedCollections);
   }
   catch (err) {
     console.error("Caught error in handleCollectionChanges: " + 
-                  JSON.stringify(err));
+                  JSON.stringify(err), JSON.stringify(err.stack));
     ok = false; 
   }
   return ok;
@@ -1183,8 +1203,11 @@ function handleChanges (plan, current) {
     }
     else { // role === "SECONDARY"
       if (plan.hasOwnProperty("Plan/DBServers/"+myId)) {
-        // Ooops! We are now a primary!
         changed = ArangoServerState.redetermineRole();
+        if (!changed) {
+          // mop: oops...changing role has failed. retry next time.
+          return false;
+        }
       }
       else {
         var found = null;
@@ -1219,7 +1242,7 @@ function handleChanges (plan, current) {
   if (role === "PRIMARY" || role === "COORDINATOR") {
     // Note: This is only ever called for DBservers (primary and secondary),
     // we keep the coordinator case here just in case...
-    success = handleCollectionChanges(plan, current);
+    success = handleCollectionChanges(plan, changed);
   }
   else {
     success = setupReplication();

lib/Rest/Endpoint.cpp

@@ -219,7 +219,11 @@ Endpoint* Endpoint::factory(const Endpoint::EndpointType type,
       return nullptr;
     }
 
+#ifndef _WIN32
     return new EndpointSrv(specification.substr(6));
+#else
+    return nullptr;
+#endif
   }
 
   else if (!StringUtils::isPrefix(domainType, "tcp://")) {