fixed some stuff

arangod/Pregel/Algorithm.h

@@ -84,7 +84,7 @@ struct Algorithm : IAlgorithm {
   virtual MessageFormat<M>* messageFormat() const = 0;
   virtual MessageCombiner<M>* messageCombiner() const = 0;
   virtual VertexComputation<V, E, M>* createComputation(uint64_t gss) const = 0;
-  virtual VertexCompensation<V, E, M>* createCompensation() {
+  virtual VertexCompensation<V, E, M>* createCompensation(uint64_t gss) const {
     return nullptr;
   }
 protected:

arangod/Pregel/Algos/PageRank.cpp

@@ -110,6 +110,18 @@ PageRankAlgorithm::createComputation(uint64_t gss) const {
   return new PageRankComputation(_threshold);
 }
 
+struct PageRankCompensation : public VertexCompensation<float, float, float> {
+  PageRankCompensation() {}
+  void compensate(bool inLostPartition) override {
+  
+  }
+};
+
+VertexCompensation<float, float, float>*
+PageRankAlgorithm::createCompensation(uint64_t gss) const {
+  return new PageRankCompensation();
+}
+
 Aggregator* PageRankAlgorithm::aggregator(std::string const& name) const {
   if (name == "convergence") {
     return new FloatMaxAggregator(0);

arangod/Pregel/Algos/PageRank.h

@@ -42,6 +42,7 @@ struct PageRankAlgorithm : public SimpleAlgorithm<float, float, float> {
   MessageCombiner<float>* messageCombiner() const override;
   VertexComputation<float, float, float>* createComputation(uint64_t gss)
       const override;
+  VertexCompensation<float, float, float>* createCompensation(uint64_t gss) const override;
   Aggregator* aggregator(std::string const& name) const override;
 };
 }

arangod/Pregel/Utils.cpp

@@ -43,6 +43,7 @@ std::string const Utils::finishedGSSPath = "finishedGSS";
 std::string const Utils::messagesPath = "messages";
 std::string const Utils::finalizeExecutionPath = "finalizeExecution";
 std::string const Utils::startRecoveryPath = "startRecovery";
+std::string const Utils::continueRecoveryPath = "continueRecovery";
 std::string const Utils::finishedRecoveryPath = "finishedRecovery";
 
 std::string const Utils::executionNumberKey = "exn";

arangod/Pregel/Utils.h

@@ -49,6 +49,7 @@ class Utils {
   static std::string const messagesPath;
   static std::string const finalizeExecutionPath;
   static std::string const startRecoveryPath;
+  static std::string const continueRecoveryPath;
   static std::string const finishedRecoveryPath;
 
   static std::string const executionNumberKey;

arangod/Pregel/VertexComputation.h

@@ -95,7 +95,7 @@ public:
 };
   
 template <typename V, typename E, typename M>
-class VertexCompensate : public VertexContext<V, E, M> {
+class VertexCompensation : public VertexContext<V, E, M> {
   friend class Worker<V, E, M>;
   
 public:

arangod/Pregel/Worker.cpp

@@ -332,41 +332,68 @@ template <typename V, typename E, typename M>
 void Worker<V, E, M>::startRecovery(VPackSlice data) {
   VPackSlice method = data.get(Utils::recoveryMethodKey);
   if (method.compareString(Utils::compensate) == 0) {
+    
+    _preRecoveryTotal = _graphStore->vertexCount();
     WorkerState nextState(_state.database(), data);
     _graphStore->loadShards(nextState);
     _state = nextState;
-    
+    compensateStep(data);
-    // TODO start compensate method
-    /*
-    ThreadPool *pool = PregelFeature::instance()->threadPool();
-    pool->enqueue([this, start, end] {
-      if (!_running) {
-        LOG(INFO) << "Execution aborted prematurely.";
-        return;
-      }
-      
-      size_t total = _graphStore->vertexCount();
-      size_t start = 0, end = delta;
-      auto vertexIterator = _graphStore->vertexIterator(start, end);
-      _executeGlobalStep(vertexIterator);
-    });
-    
-    uint64_t gss = _state.globalSuperstep();
-    std::unique_ptr<VertexComputation<V, E, M>>
-    vertexComputation(_algorithm->createComputation(gss));
-    vertexComputation->_gss = gss;
-    vertexComputation->_context = _workerContext.get();
-    vertexComputation->_graphStore = _graphStore.get();
-    vertexComputation->_outgoing = &outCache;
-    vertexComputation->_conductorAggregators = _conductorAggregators.get();
-    vertexComputation->_workerAggregators = &workerAggregator;
-    */
     
   } else if (method.compareString(Utils::rollback) == 0) {
     
   }
 }
 
+template <typename V, typename E, typename M>
+void Worker<V, E, M>::compensateStep(VPackSlice data) {
+  _conductorAggregators->resetValues();
+  VPackSlice aggValues = data.get(Utils::aggregatorValuesKey);
+  if (aggValues.isObject()) {
+    _conductorAggregators->aggregateValues(aggValues);
+  }
+  _workerAggregators->resetValues();
+  
+  ThreadPool *pool = PregelFeature::instance()->threadPool();
+  pool->enqueue([this] {
+    if (!_running) {
+      LOG(INFO) << "Compensation aborted prematurely.";
+      return;
+    }
+    
+    size_t total = _graphStore->vertexCount();
+    auto vertexIterator = _graphStore->vertexIterator(0, total);
+    
+    // TODO look if we can avoid instantiating this
+    std::unique_ptr<VertexCompensation<V, E, M>>
+    vCompensate(_algorithm->createCompensation(_state.globalSuperstep()));
+    _initializeVertexContext(vCompensate.get());
+    vCompensate->_workerAggregators = _workerAggregators.get();
+    
+    size_t i = 0;
+    for (VertexEntry *vertexEntry : vertexIterator) {
+      vCompensate->_vertexEntry = vertexEntry;
+      vCompensate->compensate(i < _preRecoveryTotal);
+      i++;
+      if (!_running) {
+        LOG(INFO) << "Execution aborted prematurely.";
+        break;
+      }
+    }
+    VPackBuilder package;
+    package.openObject();
+    package.add(Utils::senderKey, VPackValue(ServerState::instance()->getId()));
+    package.add(Utils::executionNumberKey, VPackValue(_state.executionNumber()));
+    package.add(Utils::globalSuperstepKey, VPackValue(_state.globalSuperstep()));
+    if (_workerAggregators->size() > 0) {// add aggregators
+      package.add(Utils::aggregatorValuesKey, VPackValue(VPackValueType::Object));
+      _workerAggregators->serializeValues(package);
+      package.close();
+    }
+    package.close();
+    _callConductor(Utils::finishedRecoveryPath, package.slice());
+  });
+}
+
 template <typename V, typename E, typename M>
 void Worker<V, E, M>::_callConductor(std::string path, VPackSlice message) {
   
@@ -383,6 +410,8 @@ void Worker<V, E, M>::_callConductor(std::string path, VPackSlice message) {
                    true);
 }
 
+
+
 // template types to create
 template class arangodb::pregel::Worker<int64_t, int64_t, int64_t>;
 template class arangodb::pregel::Worker<float, float, float>;

arangod/Pregel/Worker.h

@@ -37,7 +37,6 @@ class RestPregelHandler;
 namespace pregel {
 
 class IWorker {
-
  public:
   virtual ~IWorker(){};
   virtual void prepareGlobalStep(VPackSlice data) = 0;
@@ -45,6 +44,7 @@ class IWorker {
   virtual void receivedMessages(VPackSlice data) = 0;
   virtual void finalizeExecution(VPackSlice data) = 0;
   virtual void startRecovery(VPackSlice data) = 0;
+  virtual void compensateStep(VPackSlice data) = 0;
 };
 
 template <typename V, typename E>
@@ -73,6 +73,11 @@ class Worker : public IWorker {
   std::unique_ptr<Algorithm<V, E, M>> _algorithm;
   std::unique_ptr<WorkerContext> _workerContext;
   Mutex _conductorMutex;// locks callbak methods
+  mutable Mutex _threadMutex;// locks _workerThreadDone
+  
+  // only valid while recovering to determine the offset
+  // where new vertices were inserted
+  size_t _preRecoveryTotal;
  
   std::unique_ptr<GraphStore<V, E>> _graphStore;
   std::unique_ptr<IncomingCache<M>> _readCache, _writeCache;
@@ -83,7 +88,6 @@ class Worker : public IWorker {
   WorkerStats _superstepStats;
   
   size_t _runningThreads;
-  mutable Mutex _threadMutex;
   
   void _swapIncomingCaches() {
     _readCache.swap(_writeCache);
@@ -106,6 +110,7 @@ class Worker : public IWorker {
   void receivedMessages(VPackSlice data) override;
   void finalizeExecution(VPackSlice data) override;
   void startRecovery(VPackSlice data) override;
+  void compensateStep(VPackSlice data) override;
 };
 }
 }