Merge branch 'engine-api' of https://github.com/arangodb/arangodb into engine-api

arangod/RocksDBEngine/RocksDBEngine.cpp

@@ -44,6 +44,7 @@
 #include "RocksDBEngine/RocksDBIndexFactory.h"
 #include "RocksDBEngine/RocksDBKey.h"
 #include "RocksDBEngine/RocksDBRestHandlers.h"
+#include "RocksDBEngine/RocksDBReplicationManager.h"
 #include "RocksDBEngine/RocksDBTransactionCollection.h"
 #include "RocksDBEngine/RocksDBTransactionContextData.h"
 #include "RocksDBEngine/RocksDBTransactionState.h"
@@ -205,6 +206,8 @@ void RocksDBEngine::start() {
     TRI_ASSERT(false);
   }
 
+  _replicationManager = new RocksDBReplicationManager{};
+
   if (!systemDatabaseExists()) {
     addSystemDatabase();
   }
@@ -228,6 +231,9 @@ void RocksDBEngine::unprepare() {
     delete _db;
     _db = nullptr;
   }
+
+  delete _replicationManager;
+  _replicationManager = nullptr;
 }
 
 transaction::ContextData* RocksDBEngine::createTransactionContextData() {
@@ -1026,4 +1032,9 @@ RocksDBCounterManager* RocksDBEngine::counterManager() {
   return _counterManager.get();
 }
 
+RocksDBReplicationManager  * RocksDBEngine::replicationManager() {
+  TRI_ASSERT(_replicationManager);
+  return _replicationManager;
+}
+
 }  // namespace

arangod/RocksDBEngine/RocksDBEngine.h

@@ -37,10 +37,11 @@
 #include <velocypack/Slice.h>
 
 namespace arangodb {
-class RocksDBComparator;
-class RocksDBCounterManager;
 class PhysicalCollection;
 class PhysicalView;
+class RocksDBComparator;
+class RocksDBCounterManager;
+class RocksDBReplicationManager;
 class TransactionCollection;
 class TransactionState;
 
@@ -259,11 +260,13 @@ class RocksDBEngine final : public StorageEngine {
   static std::string const EngineName;
   static std::string const FeatureName;
   RocksDBCounterManager* counterManager();
+  RocksDBReplicationManager* replicationManager();
 
  private:
   rocksdb::TransactionDB*
       _db;  // single rocksdb database used in this storage engine
   rocksdb::Options _options;  // default read options
+  RocksDBReplicationManager* _replicationManager;
   std::unique_ptr<RocksDBComparator>
       _cmp;           // arangodb comparator - requried because of vpack in keys
   std::string _path;  // path used by rocksdb (inside _basePath)

arangod/RocksDBEngine/RocksDBReplicationContext.h

@@ -46,7 +46,7 @@ class RocksDBReplicationResult : public Result {
 };
 
 /// ttl in seconds
-double RocksDBReplicationContextTTL = 30 * 60.0;
+const double RocksDBReplicationContextTTL = 30 * 60.0;
 
 class RocksDBReplicationContext {
  private:

arangod/RocksDBEngine/RocksDBReplicationManager.cpp

@@ -24,6 +24,7 @@
 #include "RocksDBReplicationManager.h"
 #include "Basics/MutexLocker.h"
 #include "Logger/Logger.h"
+#include "RocksDBEngine/RocksDBEngine.h"
 #include "RocksDBEngine/RocksDBCommon.h"
 #include "RocksDBEngine/RocksDBReplicationContext.h"
 

arangod/RocksDBEngine/RocksDBReplicationManager.h

@@ -64,11 +64,12 @@ class RocksDBReplicationManager {
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief find an existing context by id
-  /// if found, the context will be returned with the usage flag set to true.
+  /// if found, the context will be returned with the isUsed() flag set to true.
   /// it must be returned later using release() or destroy()
+  /// the second parameter shows if the context you are looking for is busy or not
   //////////////////////////////////////////////////////////////////////////////
 
-  RocksDBReplicationContext* find(RocksDBReplicationId, bool&);
+  RocksDBReplicationContext* find(RocksDBReplicationId, bool& isBusy);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief return a context for later use

arangod/RocksDBEngine/RocksDBRestReplicationHandler.cpp

@@ -35,9 +35,13 @@
 #include "GeneralServer/GeneralServer.h"
 #include "Indexes/Index.h"
 #include "Logger/Logger.h"
-#include "MMFiles/MMFilesCollectionKeys.h"
-#include "MMFiles/MMFilesLogfileManager.h"
-#include "MMFiles/mmfiles-replication-dump.h"
+#include "RocksDBEngine/RocksDBCommon.h"
+#include "RocksDBEngine/RocksDBEngine.h"
+#include "RocksDBEngine/RocksDBReplicationContext.h"
+#include "RocksDBEngine/RocksDBReplicationManager.h"
+//#include "MMFiles/MMFilesCollectionKeys.h"
+//#include "MMFiles/MMFilesLogfileManager.h"
+//#include "MMFiles/mmfiles-replication-dump.h"
 #include "Replication/InitialSyncer.h"
 #include "Rest/HttpRequest.h"
 #include "Rest/Version.h"
@@ -67,10 +71,12 @@
 using namespace arangodb;
 using namespace arangodb::basics;
 using namespace arangodb::rest;
+using namespace arangodb::rocksutils;
 
 RocksDBRestReplicationHandler::RocksDBRestReplicationHandler(
     GeneralRequest* request, GeneralResponse* response)
-    : RestVocbaseBaseHandler(request, response) {}
+    : RestVocbaseBaseHandler(request, response),
+      _manager(globalRocksEngine()->replicationManager()) {}
 
 RocksDBRestReplicationHandler::~RocksDBRestReplicationHandler() {}
 
@@ -479,19 +485,21 @@ void RocksDBRestReplicationHandler::handleCommandDetermineOpenTransactions() {
 ////////////////////////////////////////////////////////////////////////////////
 
 void RocksDBRestReplicationHandler::handleCommandInventory() {
-  uint64_t contextId = TRI_CurrentTickServer();
+  RocksDBReplicationContext* context = _manager->createContext();
+
   // add context id to store
   // TODO - add context id to store
 
   VPackBuilder builder;
   builder.openObject();
-  builder.add("contextId", VPackValue(std::to_string(contextId)));
+  builder.add("contextId", VPackValue(std::to_string(context->id())));
   builder.close();
+  _manager->release(context);
 
   generateError(rest::ResponseCode::NOT_IMPLEMENTED,
                 TRI_ERROR_NOT_YET_IMPLEMENTED,
                 "replication API is not fully implemented for RocksDB yet");
-  //generateResult(rest::ResponseCode::OK, builder.slice());
+  // generateResult(rest::ResponseCode::OK, builder.slice());
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -589,22 +597,23 @@ void RocksDBRestReplicationHandler::handleCommandDump() {
     return;
   }
 
-  //get contextId
+  // get contextId
   std::string const& contextIdString = _request->value("contextId", found);
   if (found) {
     contextId = StringUtils::uint64(contextIdString);
   } else {
-    generateError(rest::ResponseCode::NOT_FOUND,
-                TRI_ERROR_HTTP_BAD_PARAMETER,
-                "replication dump request misses contextId");
+    generateError(rest::ResponseCode::NOT_FOUND, TRI_ERROR_HTTP_BAD_PARAMETER,
+                  "replication dump request misses contextId");
   }
+
+  bool isBusy = false;
+  _manager->find(contextId, isBusy);
   // lock context id || die
 
   // print request
   LOG_TOPIC(TRACE, arangodb::Logger::FIXME)
       << "requested collection dump for collection '" << collection
-      << "' using contextId '" << contextId
-      << "'";
+      << "' using contextId '" << contextId << "'";
 
   // fail
   generateError(rest::ResponseCode::NOT_IMPLEMENTED,

arangod/RocksDBEngine/RocksDBRestReplicationHandler.h

@@ -27,6 +27,7 @@
 
 #include "Basics/Common.h"
 
+#include "RocksDBEngine/RocksDBReplicationManager.h"
 #include "RestHandler/RestVocbaseBaseHandler.h"
 #include "VocBase/replication-common.h"
 
@@ -246,6 +247,8 @@ class RocksDBRestReplicationHandler : public RestVocbaseBaseHandler {
   //////////////////////////////////////////////////////////////////////////////
 
   void handleCommandGetIdForReadLockCollection();
+
+  RocksDBReplicationManager* _manager;
 };
 }