Better protection against multi-threading.

neunhoef · neunhoef · commit b37d8fff4875 · 2016-05-06T23:06:10.000+02:00
diff --git a/arangod/Cluster/AgencyCallback.cpp b/arangod/Cluster/AgencyCallback.cpp
@@ -92,8 +92,8 @@ bool AgencyCallback::executeEmpty() {
     result = _cb(VPackSlice::noneSlice());
   }
 
+  CONDITION_LOCKER(locker, _cv);
   if (_useCv) {
-    CONDITION_LOCKER(locker, _cv);
     _cv.signal();
   }
   return result;
@@ -107,8 +107,8 @@ bool AgencyCallback::execute(std::shared_ptr<VPackBuilder> newData) {
     result = _cb(newData->slice());
   }
 
+  CONDITION_LOCKER(locker, _cv);
   if (_useCv) {
-    CONDITION_LOCKER(locker, _cv);
     _cv.signal();
   }
   return result;
diff --git a/arangod/Cluster/ClusterInfo.cpp b/arangod/Cluster/ClusterInfo.cpp
@@ -901,14 +901,19 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
   std::vector<ServerID> DBServers = getCurrentDBServers();
 
   int dbServerResult = -1;
+
   std::function<bool(VPackSlice const& result)> dbServerChanged =
     [&](VPackSlice const& result) {
     size_t numDbServers;
     {
       MUTEX_LOCKER(guard, dbServersMutex);
       numDbServers = DBServers.size();
     }
-    if (result.isObject() && result.length() == numDbServers) {
+    if (result.isObject() && result.length() >= numDbServers) {
+      // We use >= here since the number of DBservers could have increased
+      // during the creation of the database and we might not yet have
+      // the latest list. Thus there could be more reports than we know
+      // servers.
       VPackObjectIterator dbs(result);
       
       std::string tmpMsg = "";
@@ -934,16 +939,24 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
         }
       }
       if (tmpHaveError) {
+        MUTEX_LOCKER(guard, dbServersMutex);
         errorMsg = "Error in creation of database:" + tmpMsg;
         dbServerResult = TRI_ERROR_CLUSTER_COULD_NOT_CREATE_DATABASE;
         return true;
       }
       loadCurrent();  // update our cache
-      dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, errorMsg);
+      {
+        MUTEX_LOCKER(guard, dbServersMutex);
+        dbServerResult = setErrormsg(TRI_ERROR_NO_ERROR, errorMsg);
+      }
     }
     return true;
   };
   
+  // ATTENTION: The following callback calls the above closure in a
+  // different thread. Nevertheless, the closure accesses some of our 
+  // local variables. Therefore we have to protect all accesses to them
+  // by the above mutex.
   auto agencyCallback = std::make_shared<AgencyCallback>(
     ac, "Current/Databases/" + name, dbServerChanged, true, false);
   _agencyCallbackRegistry->registerCallback(agencyCallback);
@@ -974,8 +987,11 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
   int count = 0;  // this counts, when we have to reload the DBServers
   while (TRI_microtime() <= endTime) {
     agencyCallback->executeByCallbackOrTimeout(getReloadServerListTimeout() / interval);
-    if (dbServerResult >= 0) {
-      break;
+    {
+      MUTEX_LOCKER(guard, dbServersMutex);
+      if (dbServerResult >= 0) {
+        break;
+      }
     }
 
     if (++count >= static_cast<int>(getReloadServerListTimeout() / interval)) {
@@ -991,8 +1007,11 @@ int ClusterInfo::createDatabaseCoordinator(std::string const& name,
       count = 0;
     }
   }
-  if (dbServerResult >= 0) {
-    return dbServerResult;
+  {
+    MUTEX_LOCKER(guard, dbServersMutex);
+    if (dbServerResult >= 0) {
+      return dbServerResult;
+    }
   }
   return setErrormsg(TRI_ERROR_CLUSTER_TIMEOUT, errorMsg);
 }

Original file line number	Diff line number	Diff line change
`@@ -92,8 +92,8 @@ bool AgencyCallback::executeEmpty() {`
`92`	`92`	`result = _cb(VPackSlice::noneSlice());`
`93`	`93`	`}`
`94`	`94`
	`95`	`+ CONDITION_LOCKER(locker, _cv);`
`95`	`96`	`if (_useCv) {`
`96`		`- CONDITION_LOCKER(locker, _cv);`
`97`	`97`	`_cv.signal();`
`98`	`98`	`}`
`99`	`99`	`return result;`
`@@ -107,8 +107,8 @@ bool AgencyCallback::execute(std::shared_ptr<VPackBuilder> newData) {`
`107`	`107`	`result = _cb(newData->slice());`
`108`	`108`	`}`
`109`	`109`
	`110`	`+ CONDITION_LOCKER(locker, _cv);`
`110`	`111`	`if (_useCv) {`
`111`		`- CONDITION_LOCKER(locker, _cv);`
`112`	`112`	`_cv.signal();`
`113`	`113`	`}`
`114`	`114`	`return result;`