NDimensionalExecutor.h

#ifndef N_DIMENSIONAL_EXECUTOR_H
#define N_DIMENSIONAL_EXECUTOR_H
 
#include <iomanip>
#include <sstream>
#include <set>
#include <vector>
#include <functional>
#include <cstddef>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <queue>
#include <stdexcept>
#include <utility>
#include <exception>
#include <memory>
#include <string>
#include <Rtypes.h>
#include <THnSparse.h>
#include "NLogger.h"
#include "NThreadData.h"
#include "NTaskStateManager.h"
 
namespace Ndmspc {

struct ExecutionProgress {
  size_t tasksAcked{0};     
  size_t tasksPending{0};   
  size_t tasksRunning{0};   
  size_t tasksDone{0};      
  size_t activeWorkers{0};  
};

class NDimensionalExecutor {
  public:
  NDimensionalExecutor(const std::vector<int> & minBounds, const std::vector<int> & maxBounds);

  NDimensionalExecutor(THnSparse * hist, bool onlyfilled = false);
 
  ~NDimensionalExecutor();
 
  void SetBounds(const std::vector<int> & minBounds, const std::vector<int> & maxBounds);

  void Execute(const std::function<void(const std::vector<int> & coords)> & func);

  template <typename TObject>
  void ExecuteParallel(const std::function<void(const std::vector<int> & coords, TObject & thread_object)> & func,
                       std::vector<TObject> & thread_objects);

  size_t ExecuteParallelProcessIpc(std::vector<NThreadData *> & workerObjects, size_t processCount);
  void   StartProcessIpc(std::vector<NThreadData *> & workerObjects, size_t processCount,
                         const std::string & tcpBindEndpoint = "", const std::string & jobDir = "",
                         const std::string & treeName = "ngnt", const std::string & macroList = "",
                         const std::string & tmpDir = "", const std::string & tmpResultsDir = "",
                         const std::string & macroParams = "");
  size_t ExecuteCurrentBoundsProcessIpc(const std::string & definitionName = "",
                                        const std::vector<Long64_t> * definitionIds = nullptr,
                                        const std::function<void(const ExecutionProgress&)> & progressCallback = nullptr);
  void   FinishProcessIpc(bool abort = false);
 
  std::set<size_t> GetRegisteredWorkerIndices() const { return fRegisteredWorkerIndices; }

  size_t Dimensions() const { return fNumDimensions; }

  const std::vector<int> & GetMinBounds() const { return fMinBounds; }

  const std::vector<int> & GetMaxBounds() const { return fMaxBounds; }
 
  private:
  size_t           fNumDimensions; 
  std::vector<int> fMinBounds;     
  std::vector<int> fMaxBounds;     
  std::vector<int> fCurrentCoords; 

  bool Increment();

  bool InitTcpWorker(const std::string & identity);

  bool HandleBootstrap(const std::string & identity);

  size_t HandleWorkerFailure(const std::string & failedIdentity,
                             const std::string & failureReason,
                             size_t & outstanding,
                             size_t & acked);
 
  struct IpcSession;
  std::unique_ptr<IpcSession> fIpcSession;
  std::set<size_t>            fRegisteredWorkerIndices; 
};
 
 
 
} // namespace Ndmspc
 
#endif