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

arangod/RocksDBEngine/RocksDBPrimaryIndex.cpp

@@ -115,7 +115,6 @@ RocksDBAllIndexIterator::RocksDBAllIndexIterator(
     LogicalCollection* collection, transaction::Methods* trx,
     ManagedDocumentResult* mmdr, RocksDBPrimaryIndex const* index, bool reverse)
     : IndexIterator(collection, trx, mmdr, index),
-      _index(index),
       _cmp(index->_cmp),
       _reverse(reverse),
       _bounds(RocksDBKeyBounds::PrimaryIndex(index->objectId())) {

arangod/RocksDBEngine/RocksDBPrimaryIndex.h

@@ -89,7 +89,6 @@ class RocksDBAllIndexIterator final : public IndexIterator {
  private:
   bool outOfRange() const;
 
-  RocksDBPrimaryIndex const* _index;
   RocksDBComparator const* _cmp;
   bool const _reverse;
   std::unique_ptr<rocksdb::Iterator> _iterator;

arangod/RocksDBEngine/RocksDBReplicationContext.cpp

@@ -285,12 +285,12 @@ void RocksDBReplicationContext::deleted() { _isDeleted = true; }
 
 bool RocksDBReplicationContext::isUsed() const { return _isUsed; }
 
-void RocksDBReplicationContext::use() {
+void RocksDBReplicationContext::use(double ttl) {
   TRI_ASSERT(!_isDeleted);
   TRI_ASSERT(!_isUsed);
 
   _isUsed = true;
-  _expires = TRI_microtime() + RocksDBReplicationContextTTL;
+  _expires = TRI_microtime() + ttl;
 }
 
 void RocksDBReplicationContext::release() {

arangod/RocksDBEngine/RocksDBReplicationContext.h

@@ -45,9 +45,6 @@ class RocksDBReplicationResult : public Result {
   uint64_t _maxTick;
 };
 
-/// ttl in seconds
-const double RocksDBReplicationContextTTL = 30 * 60.0;
-
 class RocksDBReplicationContext {
  private:
   typedef std::function<void(DocumentIdentifierToken const& token)>
@@ -79,7 +76,7 @@ class RocksDBReplicationContext {
   bool isDeleted() const;
   void deleted();
   bool isUsed() const;
-  void use();
+  void use(double ttl);
   /// remove use flag
   void release();
 

arangod/RocksDBEngine/RocksDBReplicationManager.cpp

@@ -34,6 +34,7 @@ using namespace arangodb;
 using namespace arangodb::rocksutils;
 
 size_t const RocksDBReplicationManager::MaxCollectCount = 32;
+double const RocksDBReplicationManager::DefaultTTL = 30 * 60.0;
 
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief create a context repository
@@ -103,10 +104,9 @@ RocksDBReplicationContext* RocksDBReplicationManager::createContext() {
     if (_contexts.empty()) {
       disableFileDeletions();
     }
-
+    
     _contexts.emplace(id, context.get());
   }
-
   return context.release();
 }
 
@@ -160,7 +160,7 @@ bool RocksDBReplicationManager::remove(RocksDBReplicationId id) {
 ////////////////////////////////////////////////////////////////////////////////
 
 RocksDBReplicationContext* RocksDBReplicationManager::find(
-    RocksDBReplicationId id, bool& busy) {
+    RocksDBReplicationId id, bool& busy, double ttl) {
   RocksDBReplicationContext* context = nullptr;
   busy = false;
 
@@ -186,7 +186,7 @@ RocksDBReplicationContext* RocksDBReplicationManager::find(
       return nullptr;
     }
 
-    context->use();
+    context->use(ttl);
   }
 
   return context;

arangod/RocksDBEngine/RocksDBReplicationManager.h

@@ -70,7 +70,8 @@ class RocksDBReplicationManager {
   /// not
   //////////////////////////////////////////////////////////////////////////////
 
-  RocksDBReplicationContext* find(RocksDBReplicationId, bool& isBusy);
+  RocksDBReplicationContext* find(RocksDBReplicationId, bool& isBusy,
+                                  double ttl = DefaultTTL);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief return a context for later use
@@ -95,6 +96,9 @@ class RocksDBReplicationManager {
   //////////////////////////////////////////////////////////////////////////////
 
   bool garbageCollect(bool);
+  
+  /// default time-to-live for contexts
+  static double const DefaultTTL;
 
  private:
   void disableFileDeletions();

arangod/RocksDBEngine/RocksDBRestReplicationHandler.cpp

@@ -328,9 +328,105 @@ void RocksDBRestReplicationHandler::handleCommandLoggerState() {
 ////////////////////////////////////////////////////////////////////////////////
 
 void RocksDBRestReplicationHandler::handleCommandBatch() {
-  generateError(rest::ResponseCode::NOT_IMPLEMENTED,
-                TRI_ERROR_NOT_YET_IMPLEMENTED,
-                "replication API is not fully implemented for RocksDB yet");
+  // extract the request type
+  auto const type = _request->requestType();
+  auto const& suffixes = _request->suffixes();
+  size_t const len = suffixes.size();
+  
+  TRI_ASSERT(len >= 1);
+  
+  if (type == rest::RequestType::POST) {
+    // create a new blocker
+    std::shared_ptr<VPackBuilder> input = _request->toVelocyPackBuilderPtr();
+    
+    if (input == nullptr || !input->slice().isObject()) {
+      generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
+                    "invalid JSON");
+      return;
+    }
+    
+    // extract ttl
+    //double expires =
+    //VelocyPackHelper::getNumericValue<double>(input->slice(), "ttl", 0);
+    
+    //TRI_voc_tick_t id;
+    //StorageEngine* engine = EngineSelectorFeature::ENGINE;
+    //int res = engine->insertCompactionBlocker(_vocbase, expires, id);
+    
+    RocksDBReplicationContext *ctx = _manager->createContext();
+    if (ctx == nullptr) {
+      THROW_ARANGO_EXCEPTION(TRI_ERROR_FAILED);
+    }
+    
+    VPackBuilder b;
+    b.add(VPackValue(VPackValueType::Object));
+    b.add("id", VPackValue(std::to_string(ctx->id())));
+    b.close();
+    generateResult(rest::ResponseCode::OK, b.slice());
+    return;
+  }
+  
+  if (type == rest::RequestType::PUT && len >= 2) {
+    // extend an existing blocker
+    TRI_voc_tick_t id =
+    static_cast<TRI_voc_tick_t>(StringUtils::uint64(suffixes[1]));
+    
+    auto input = _request->toVelocyPackBuilderPtr();
+    
+    if (input == nullptr || !input->slice().isObject()) {
+      generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
+                    "invalid JSON");
+      return;
+    }
+    
+    // extract ttl
+    double expires =
+    VelocyPackHelper::getNumericValue<double>(input->slice(), "ttl", 0);
+   
+    int res = TRI_ERROR_NO_ERROR;
+    bool busy;
+    RocksDBReplicationContext *ctx = _manager->find(id, busy, expires);
+    if (busy) {
+      res = TRI_ERROR_CURSOR_BUSY;
+    } else if (ctx == nullptr) {
+      res = TRI_ERROR_CURSOR_NOT_FOUND;
+    } else {
+      _manager->release(ctx);
+    }
+    
+    // now extend the blocker
+    //StorageEngine* engine = EngineSelectorFeature::ENGINE;
+    // res = engine->extendCompactionBlocker(_vocbase, id, expires);
+    
+    if (res == TRI_ERROR_NO_ERROR) {
+      resetResponse(rest::ResponseCode::NO_CONTENT);
+    } else {
+      generateError(GeneralResponse::responseCode(res), res);
+    }
+    return;
+  }
+  
+  if (type == rest::RequestType::DELETE_REQ && len >= 2) {
+    // delete an existing blocker
+    TRI_voc_tick_t id =
+    static_cast<TRI_voc_tick_t>(StringUtils::uint64(suffixes[1]));
+    
+    bool found = _manager->remove(id);
+    //StorageEngine* engine = EngineSelectorFeature::ENGINE;
+    //int res = engine->removeCompactionBlocker(_vocbase, id);
+    
+    if (found) {
+      resetResponse(rest::ResponseCode::NO_CONTENT);
+    } else {
+      int res = TRI_ERROR_CURSOR_NOT_FOUND;
+      generateError(GeneralResponse::responseCode(res), res);
+    }
+    return;
+  }
+  
+  // we get here if anything above is invalid
+  generateError(rest::ResponseCode::METHOD_NOT_ALLOWED,
+                TRI_ERROR_HTTP_METHOD_NOT_ALLOWED);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -483,19 +579,27 @@ void RocksDBRestReplicationHandler::handleCommandDetermineOpenTransactions() {
 ////////////////////////////////////////////////////////////////////////////////
 
 void RocksDBRestReplicationHandler::handleCommandInventory() {
+  RocksDBReplicationContext *ctx = nullptr;
+  bool found, busy;
+  std::string batchId = _request->value("batchId", found);
+  if (found) {
+    ctx = _manager->find(StringUtils::uint64(batchId), busy);
+  }
+  if (!found || busy || ctx == nullptr) {
+    generateError(rest::ResponseCode::NOT_FOUND,
+                  TRI_ERROR_CURSOR_NOT_FOUND,
+                  "batchId not specified");
+  }
 
   TRI_voc_tick_t tick = TRI_CurrentTickServer();
   
   // include system collections?
   bool includeSystem = true;
-  bool found;
   std::string const& value = _request->value("includeSystem", found);
-  
   if (found) {
     includeSystem = StringUtils::boolean(value);
   }
   
-  RocksDBReplicationContext* ctx = _manager->createContext();
   std::pair<RocksDBReplicationResult, std::shared_ptr<VPackBuilder>> result =
     ctx->getInventory(this->_vocbase, includeSystem);
   if (!result.first.ok()) {
@@ -509,9 +613,6 @@ void RocksDBRestReplicationHandler::handleCommandInventory() {
   
   VPackBuilder builder;
   builder.openObject();
-  
-  // add context id
-  builder.add("contextId", VPackValue(ctx->id()));
 
   // add collections data
   builder.add("collections", collections);
@@ -640,9 +741,8 @@ void RocksDBRestReplicationHandler::handleCommandRemoveKeys() {
 ////////////////////////////////////////////////////////////////////////////////
 
 void RocksDBRestReplicationHandler::handleCommandDump() {
+  
   bool found = false;
-  uint64_t contextId = 0;
-
   // handle collection
   std::string const& collection = _request->value("collection");
   if (collection.empty()) {
@@ -650,6 +750,19 @@ void RocksDBRestReplicationHandler::handleCommandDump() {
                   "invalid collection parameter");
     return;
   }
+  
+  
+  bool busy; // find cursor
+  RocksDBReplicationContext *ctx = nullptr;
+  std::string batchId = _request->value("batchId", found);
+  if (found) {
+    ctx = _manager->find(StringUtils::uint64(batchId), busy);
+  }
+  if (!found || busy || ctx == nullptr) {
+    generateError(rest::ResponseCode::NOT_FOUND,
+                  TRI_ERROR_CURSOR_NOT_FOUND,
+                  "batchId not specified");
+  }
 
   arangodb::LogicalCollection* c = _vocbase->lookupCollection(collection);
   if (c == nullptr) {
@@ -658,23 +771,10 @@ void RocksDBRestReplicationHandler::handleCommandDump() {
     return;
   }
 
-  // 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");
-  }
-
-  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 '" << ctx->id() << "'";
 
   // fail
   generateError(rest::ResponseCode::NOT_IMPLEMENTED,