text:
#include <random>
#include <fstream>

#include <diy/master.hpp>
#include <diy/reduce.hpp>
#include <diy/partners/merge.hpp>
#include <diy/decomposition.hpp>
#include <diy/assigner.hpp>
#include <diy/mpi.hpp>
#include <diy/serialization.hpp>
#include <diy/partners/broadcast.hpp>
#include <diy/reduce-operations.hpp>

#include <Eigen/Eigen>
#include <highfive/H5File.hpp>
#include <highfive/H5Easy.hpp>
#include <highfive/H5DataSet.hpp>
#include <highfive/H5DataSpace.hpp>

#include "AppEval.h"
#include "SignalGenerator.h"
#include "ProblemSBN.h"
#include "optim.hpp"

#include "opts.h"

using namespace Eigen;
using namespace Apprentice;
using namespace SignalGenerator;
using namespace ProblemSBN;

using namespace std;

std::tuple<std::vector<std::vector<size_t> >, size_t, size_t>  mkRankWork(size_t nPoints, size_t nUniverses, int thisRank, int worldSize) {
    size_t _S(nPoints*nUniverses);
    std::vector<size_t> _L;
    size_t maxwork = size_t(ceil(_S/worldSize));
    for (size_t r=0; r <             _S%worldSize; ++r) _L.push_back(maxwork + 1);
    for (size_t r=0; r < worldSize - _S%worldSize; ++r) _L.push_back(maxwork    );

    std::vector<size_t> _bp, _bu;
    _bp.push_back(0);
    _bu.push_back(0);

    size_t _n(0), _temp(0);
    for (size_t i=0; i<nPoints;++i) {
       for (size_t j=0; j<nUniverses;++j) {
          if (_temp == _L[_n]) {
             _bp.push_back(i);
             _bu.push_back(j);
             _temp = 0;
             _n+=1;
          }
          _temp+=1;
       }
    }
    _bp.push_back(nPoints-1);
    _bu.push_back(nUniverses);

    std::vector<size_t> rankworkbl;
    //std::cerr << thisRank << " : " << _bu.size() -1 << " sets of work " << _bp.size() -1 << "\n";
    rankworkbl.push_back(_bp[thisRank]);
    rankworkbl.push_back(_bu[thisRank]);
    rankworkbl.push_back(_bp[thisRank+1]);
    rankworkbl.push_back(_bu[thisRank+1]);

    size_t pStart = rankworkbl[0];
    size_t uStart = rankworkbl[1];
    size_t pLast  = rankworkbl[2];
    size_t uLast  = rankworkbl[3];

    size_t i_begin = pStart * nUniverses + uStart;
    size_t i_end   = pLast  * nUniverses + uLast;

    //std::cerr << "[" << i << "] " << pStart << " " <<  uStart << " " << pLast << " " << uLast << " starts: " << i_begin << " " << i_end << " " <<i_end-i_begin << "\n";
    size_t lenDS = i_end - i_begin;
    size_t d_bgn = i_begin;

    std::vector<std::vector<size_t> > RW;
    if (pStart == pLast) RW.push_back({pStart, uStart, uLast});
    else {
      RW.push_back({pStart, uStart, nUniverses});
       for (size_t _p = pStart+1; _p<pLast;++_p) {
          RW.push_back({_p, 0, nUniverses});
       }
       if (uLast>0) RW.push_back({pLast, 0, uLast});

    }

    return std::make_tuple(RW, i_begin, lenDS);
}

void loadData(const char* fname, std::string what, std::vector<double> & v_buffer, int & n_rows, int & n_cols) {
    H5Easy::File file(fname, H5Easy::File::ReadOnly);
    MatrixXd _mat      = H5Easy::load<MatrixXd>(file, what);
    n_rows = _mat.rows();
    n_cols = _mat.cols();
    v_buffer = std::vector<double>(_mat.data(), _mat.data() + _mat.rows() * _mat.cols());
}

void loadData(const char* fname, std::string what, std::vector<int> & v_buffer, int & n_rows, int & n_cols) {
    H5Easy::File file(fname, H5Easy::File::ReadOnly);
    MatrixXi _mat      = H5Easy::load<MatrixXi>(file, what);
    n_rows = _mat.rows();
    n_cols = _mat.cols();
    v_buffer = std::vector<int>(_mat.data(), _mat.data() + _mat.rows() * _mat.cols());
}

MatrixXd bcMatrixXd(diy::mpi::communicator world, std::vector<double>  v_buffer, int  n_rows, int  n_cols) {
    diy::mpi::broadcast(world, v_buffer, 0);
    diy::mpi::broadcast(world, n_rows,   0);
    diy::mpi::broadcast(world, n_cols,   0);

    Map<MatrixXd> mat(v_buffer.data(), n_rows, n_cols);
    return mat;
}

MatrixXi bcMatrixXi(diy::mpi::communicator world, std::vector<int>  v_buffer, int  n_rows, int  n_cols) {
    diy::mpi::broadcast(world, v_buffer, 0);
    diy::mpi::broadcast(world, n_rows,   0);
    diy::mpi::broadcast(world, n_cols,   0);

    Map<MatrixXi> mat(v_buffer.data(), n_rows, n_cols);
    return mat;
}

AppEval mkAppEval(diy::mpi::communicator world, const char* fname) {
    std::vector<double> v_buff;
    std::vector<int> v_buff_i;
    int ncols, nrows;
    if (world.rank()==0) loadData(fname, "pcoeff",      v_buff,   nrows, ncols);
    MatrixXd PC = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(fname, "qcoeff",      v_buff,   nrows, ncols);
    MatrixXd QC = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(fname, "a",           v_buff,   nrows, ncols);
    VectorXd a = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(fname, "xmin",        v_buff,   nrows, ncols);
    VectorXd xmin = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(fname, "scaleterm",   v_buff,   nrows, ncols);
    VectorXd scaleterm = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(fname, "structure",   v_buff_i, nrows, ncols);
    MatrixXi structure = bcMatrixXi(world, v_buff_i, nrows, ncols);

    return AppEval(PC, QC, structure, a, xmin, scaleterm);
}

MyConfig mkConfig() {
    std::vector<int> num_bins = {19};
    std::vector<int> num_subchannels = {2};
    std::vector<std::vector<int> > subchannel_osc_patterns = { {22, 0} };
    return MyConfig{3, 38, 1, 19, 57, 114, num_bins, num_subchannels, subchannel_osc_patterns};
}

typedef diy::DiscreteBounds Bounds;

struct FitBlock {
    static void*    create()            { return new FitBlock; }
    static void     destroy(void* b)    { delete static_cast<FitBlock*>(b); }

    std::vector<double> last_chi_min, delta_chi;
    std::vector<int> n_iter, i_grid, i_univ;
};

struct MyGrid {
  ArrayXXd points;
  Array2d MIN;
  Array2d MAX;
};

MyGrid mkGrid(int nx, double xmin, double xmax, int ny, double ymin, double ymax) {

  int N=nx*ny;
  ArrayXXd ret(N,2);
  auto M = ArrayXd::LinSpaced(nx, xmin, xmax);
  auto P = ArrayXd::LinSpaced(ny, ymin, ymax);
  for (auto m=0; m<nx;++m) {
    for (auto p=0;p<ny;++p) {
       int idx = m*nx +p;
       ret(idx,0) = M[m];
       ret(idx,1) = P[p];
    }
  }
  Array2d minbound;minbound<<xmin,ymin;
  Array2d maxbound;maxbound<<xmax,ymax;
  return {ret, minbound, maxbound};
}

void writeGrid(HighFive::File* file, ArrayXXd  const & coords) {
    size_t npoints(coords.rows()), dim(coords.cols());
    file->createDataSet<double>("grid", HighFive::DataSpace( {dim, npoints} ));
    diy::mpi::communicator world;
    if (world.rank()==0) {
       HighFive::DataSet d_grid = file->getDataSet("grid");
       d_grid.select({0, 0}, {dim,npoints}).write(coords.data());
    }
}

void createDataSets(HighFive::File* file, size_t nPoints, size_t nUniverses) {
   file->createDataSet<size_t>("point",    HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<size_t>("universe", HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("xmin",     HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("ymin",     HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("thischi2",     HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("minchi2",     HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("minchi2pf",     HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("xtrue",    HighFive::DataSpace( {1, nPoints*nUniverses} ));
   file->createDataSet<double>("ytrue",    HighFive::DataSpace( {1, nPoints*nUniverses} ));
}

struct MyFitData
{
    VectorXd data;
    MatrixXd covmat;
    MatrixXd invcovbg;
    SignalGenerator::SignalGenerator sig;
    MyConfig config;
};

double prefit_fn(const arma::vec& vals_inp, arma::vec* grad_out, void * opt_data) {
  MyFitData* mfd  = reinterpret_cast<MyFitData*>(opt_data);
  Vector3d xeval; xeval << vals_inp[0], 1, vals_inp[1];
  auto D     = mfd->sig.predict(xeval, 1);
  double obj =  calcChi(mfd->data - collapseVector(D, mfd->config), mfd->invcovbg);

  if (grad_out) {
    const double h=1e-5;
    double g[2];

#pragma omp parallel for
    for (unsigned int i=0;i<2;++i) {
      Vector3d xgrad0; xgrad0 << vals_inp[0]+h, 1, vals_inp[1];
      auto D0     = mfd->sig.predict(xgrad0, 1);
      double obj0 =  calcChi(mfd->data - collapseVector(D0, mfd->config), mfd->invcovbg);
      g[i] = (obj0-obj)/h;
    }
    *grad_out = {g[0], g[1]};
  }
  return obj;
}

double fit_fn(const arma::vec& vals_inp, arma::vec* grad_out, void * opt_data) {
  MyFitData* mfd  = reinterpret_cast<MyFitData*>(opt_data);
  Vector3d xeval; xeval << vals_inp[0], 1, vals_inp[1];
  auto D     = mfd->sig.predict(xeval, 1);
  auto CINV  = updateInvCov(mfd->covmat, D, mfd->config);
  double obj =  calcChi(mfd->data - collapseVector(D, mfd->config), CINV);

  if (grad_out) {
    const double h=1e-5;
    double g[2];

#pragma omp parallel for
    for (unsigned int i=0;i<2;++i) {
      Vector3d xgrad0; xgrad0 << vals_inp[0]+h, 1, vals_inp[1];
      auto D0     = mfd->sig.predict(xgrad0, 1);
      auto CINV0  = updateInvCov(mfd->covmat, D0, mfd->config);
      double obj0 =  calcChi(mfd->data - collapseVector(D0, mfd->config), CINV0);
      g[i] = (obj0-obj)/h;
    }
    *grad_out = {g[0], g[1]};
  }
  return obj;
}

inline Array2d unscale(Array2d const & X, Array2d const & xmin, Array2d const & scale) {
  return xmin + (X + Array2d::Ones()) / scale;
}

ArrayXXd mkPoints(Array2d const & p0, Array2d const & pmin, Array2d const & pmax, int nPoints, double dist) {
    Array2d scale = 2./(pmax-pmin);
    int good(0);
    ArrayXXd ret(nPoints, 2);
    double d2 = dist*dist;
    while(good<nPoints) {
        Array2d tt = unscale(Array2d::Random(), pmin, scale);

        double l2 = (p0(0) - tt(0)) * (p0(0) - tt(0)) + (p0(1) - tt(1)) * (p0(1) - tt(1));
        if (l2 < d2) {
            ret.row(good) = tt;
            good++;
        }
    }
    return ret;
}

std::tuple<double, int> universeChi2(MyFitData & mfd, MyGrid const & MG) {
   double chimin=std::numeric_limits<double>::infinity();

   int bestP(0);
   for (size_t i=0; i<MG.points.rows(); ++i) {

      double chi = prefit_fn({MG.points(i,0), MG.points(i,1)}, nullptr, &mfd);
       if (chi<chimin) {
          chimin = chi;
          bestP=i;
       }
   }
   return {chimin, bestP};
}

void runMin(FitBlock* b, diy::Master::ProxyWithLink const& cp, SignalGenerator::SignalGenerator  sig, const MatrixXd & covmat, const MatrixXd & invcovbg, MyConfig const & config, MyGrid const & MG, size_t nUniverses, HighFive::File* file) {

  optim::algo_settings_t settings;
  settings.err_tol = 1e-6;
  settings.iter_max=100;
  settings.vals_bound=true;
  settings.lower_bounds = {MG.MIN(0), MG.MIN(1)};
  settings.upper_bounds = {MG.MAX(0), MG.MAX(1)};

  Array2d pmin; pmin <<settings.lower_bounds[0], settings.lower_bounds[1];
  Array2d pmax; pmax <<settings.upper_bounds[0], settings.upper_bounds[1];

  std::mt19937 rng(cp.gid());
  diy::mpi::communicator world;
  const auto  [rankwork, ds_bgn, _ds_len] = mkRankWork(MG.points.rows(), nUniverses, world.rank(),  world.size());
  const size_t ds_len(_ds_len);
  ArrayXXi res_meta(ds_len, 2);
  ArrayXXd res_val( ds_len, 7);

  bool success;
  arma::vec  xwork = arma::vec(2);
  arma::vec _xwork = arma::vec(2);
  size_t idx(0);

  double minValue(1e40);

  for (auto r : rankwork) {
     size_t i_grid = r[0];

     Array3d xeval; xeval << MG.points(i_grid, 0), 1, MG.points(i_grid, 1);
     auto const D  = sig.predict(xeval, 1);
     Eigen::MatrixXd const & LMAT = cholD(covmat, D);

     auto const Dc = collapseVector(D, config);
     MyFitData mfd = {Dc, covmat, invcovbg, sig, config};

     for (size_t uu=r[1]; uu<r[2];++uu) {
        VectorXd const fake_data  = ProblemSBN::poisson_fluctuate(sample(D, LMAT, rng), rng);
        VectorXd const fake_dataC = collapseVector(fake_data, config);
        mfd.data = std::move(fake_dataC);

        xwork = { MG.points(i_grid,0),  MG.points(i_grid,1) };

        Vector2d ewk; ewk<< xwork[0], xwork[1];
        auto VIC = mkPoints(ewk, pmin, pmax, 10, 0.1);

        optim::lbfgs(xwork, prefit_fn, &mfd, settings);
        double minValuePrefit = prefit_fn(xwork, nullptr, &mfd);
        //std::cerr << "[" << world.rank() <<"] "  << "WINNERpre: " << minValuePrefit << " at " << xwork[0] << " " << xwork[1] << "(" << G(i_grid,0) << ", "<<  G(i_grid,1) << ")" << "\n";
        //success = optim::cg(xwork, fit_fn, &mfd, settings);

        for (int ii=0;ii<VIC.rows();++ii) {
           _xwork[0] = MG.points(ii, 0);
           _xwork[1] = MG.points(ii, 1);
           optim::lbfgs(_xwork, prefit_fn, &mfd, settings);
           double _minValue = prefit_fn(_xwork, nullptr, &mfd);

           if (_minValue < minValuePrefit) {
              minValuePrefit = _minValue;
              xwork = _xwork;
           }
        }

        //std::cerr << "[" << world.rank() <<"] "  << "WINNERpre MS: " << minValuePrefit << " at " << xwork[0] << " " << xwork[1] << "(" << G(i_grid,0) << ", "<<  G(i_grid,1) << ")" << "\n";
        Vector2d ewk2; ewk2<< xwork[0], xwork[1];
        auto VIC2 = mkPoints(ewk, pmin, pmax, 5, 0.1);

        optim::lbfgs(xwork, fit_fn, &mfd, settings);
        double minValue = fit_fn(xwork, nullptr, &mfd);
        //std::cerr << "[" << world.rank() <<"] "  << "WINNER:   " << minValue << " at " << xwork[0] << " " << xwork[1] << "(" << G(i_grid,0) << ", "<<  G(i_grid,1) << ")" << "\n";
        //  Multistart
        for (int ii=0;ii<VIC2.rows();++ii) {
           _xwork[0] = MG.points(ii, 0);
           _xwork[1] = MG.points(ii, 1);
           success = optim::lbfgs(_xwork, fit_fn, &mfd, settings);
           double _minValue = fit_fn(_xwork, nullptr, &mfd);

           if (_minValue < minValue) {
              minValue = _minValue;
              xwork = _xwork;
           }
        }

        xwork = { MG.points(i_grid,0),  MG.points(i_grid,1) };
        //std::cerr << "WINNERMS: " << minValue << " at " << xwork[0] << " " << xwork[1] << "\n\n";

        Vector3d xmineval; xmineval << xwork[0], 1, xwork[1];
        VectorXd sigmin = sig.predict(xmineval, 1);
        MatrixXd invcovatmin  = updateInvCov(covmat, sigmin, config);
        double this_chi = calcChi(fake_dataC, Dc, invcovatmin);

        _xwork = { MG.points(i_grid,0),  MG.points(i_grid,1) };
        double thus_chi = fit_fn(_xwork, nullptr, &mfd);

        //float this_chi = this->CalcChi(fake_data, true_spec->collapsed_vector,inverse_current_collapsed_covariance_matrix);

        res_meta(idx,0) = r[0];
        res_meta(idx,1) = uu;
        res_val(idx,0) = xwork[0];
        res_val(idx,1) = xwork[1];
        res_val(idx,2) = this_chi;
        res_val(idx,3) = minValue;
        res_val(idx,4) = thus_chi;
        //res_val(idx,3) = this_chi - globmin;
        res_val(idx,5) = MG.points(i_grid, 0);
        res_val(idx,6) = MG.points(i_grid, 1);
        idx++;
       }
       if (world.rank()==0 && idx%100==0) fmt::print(stderr, "[{}] progress: {}/{}\n",cp.gid(), idx, ds_len);
    }

    HighFive::DataSet d_point    = file->getDataSet("point");
    HighFive::DataSet d_universe = file->getDataSet("universe");
    d_point   .select({0, ds_bgn}, {1,idx}).write(res_meta.col(0).data());
    d_universe.select({0, ds_bgn}, {1,idx}).write(res_meta.col(1).data());

    HighFive::DataSet d_xmin     = file->getDataSet("xmin");
    HighFive::DataSet d_ymin     = file->getDataSet("ymin");
    HighFive::DataSet d_chi2     = file->getDataSet("thischi2");
    HighFive::DataSet d_minchi2  = file->getDataSet("minchi2");
    HighFive::DataSet d_minchi2pf= file->getDataSet("minchi2pf");
    HighFive::DataSet d_xtrue    = file->getDataSet("xtrue");
    HighFive::DataSet d_ytrue    = file->getDataSet("ytrue");

    d_xmin     .select({0, ds_bgn}, {1,idx}).write(res_val.col(0).data());
    d_ymin     .select({0, ds_bgn}, {1,idx}).write(res_val.col(1).data());
    d_chi2     .select({0, ds_bgn}, {1,idx}).write(res_val.col(2).data());
    d_minchi2  .select({0, ds_bgn}, {1,idx}).write(res_val.col(3).data());
    d_minchi2pf.select({0, ds_bgn}, {1,idx}).write(res_val.col(4).data());

    d_xtrue    .select({0, ds_bgn}, {1,idx}).write(res_val.col(5).data());
    d_ytrue    .select({0, ds_bgn}, {1,idx}).write(res_val.col(6).data());
}

void runGrid(FitBlock* b, diy::Master::ProxyWithLink const& cp, SignalGenerator::SignalGenerator  sig, const MatrixXd & covmat, const MatrixXd & invcovbg, MyConfig const & config, MyGrid const & MG, size_t nUniverses, HighFive::File* file) {

  std::mt19937 rng(cp.gid());
  diy::mpi::communicator world;
  const auto  [rankwork, ds_bgn, _ds_len] = mkRankWork(MG.points.rows(), nUniverses, world.rank(),  world.size());
  const size_t ds_len(_ds_len);
  ArrayXXi res_meta(ds_len, 2);
  ArrayXXd res_val( ds_len, 7);

  size_t idx(0);

  for (auto r : rankwork) {
     size_t i_grid = r[0];

     Array3d xeval; xeval << MG.points(i_grid, 0), 1, MG.points(i_grid, 1);
     auto const D  = sig.predict(xeval, 1);
     auto const Dc = collapseVector(D, config);

     Eigen::MatrixXd const & LMAT = cholD(covmat, D);

     for (size_t uu=r[1]; uu<r[2];++uu) {
        VectorXd const fake_data  = ProblemSBN::poisson_fluctuate(sample(D, LMAT, rng), rng);
        VectorXd const fake_dataC = collapseVector(fake_data, config);
        MyFitData mfd = {fake_dataC, covmat, invcovbg, sig, config};

        size_t n_iter(0);
        double last_chi_min = 1e40;
        int    pbest = -99;
        int _pbest(-99);

        for(n_iter = 0; n_iter < 5; n_iter++){
            if(n_iter!=0){
                Vector3d xeval; xeval << MG.points(pbest,0), 1, MG.points(pbest,1);
                auto D     = mfd.sig.predict(xeval, 1);
                auto CINV  = updateInvCov(mfd.covmat, D, mfd.config);
                mfd.invcovbg = std::move(CINV);
            }
            auto [chi_min, _pbest]  = universeChi2(mfd, MG);//, goodpoints);
            if(n_iter!=0){
                if(fabs(chi_min-last_chi_min) < .001){
                    last_chi_min = chi_min;
                    pbest = _pbest;
                    break;
                }
            }
            last_chi_min = chi_min;
                    pbest = _pbest;
        }

        Vector3d xmineval; xmineval << MG.points(pbest,0), 1, MG.points(pbest,1);
        double this_chi = calcChi(fake_dataC, Dc, mfd.invcovbg);

        res_meta(idx,0) = r[0];
        res_meta(idx,1) = uu;
        res_val(idx,0) = MG.points(pbest,0);
        res_val(idx,1) = MG.points(pbest,1);
        res_val(idx,2) = this_chi;
        res_val(idx,3) = last_chi_min;
        res_val(idx,4) = last_chi_min;
        res_val(idx,5) = MG.points(i_grid, 0);
        res_val(idx,6) = MG.points(i_grid, 1);
        idx++;
       }
       if (world.rank()==0 && idx%100==0) fmt::print(stderr, "[{}] progress: {}/{}\n",cp.gid(), idx, ds_len);
    }

    HighFive::DataSet d_point    = file->getDataSet("point");
    HighFive::DataSet d_universe = file->getDataSet("universe");
    d_point   .select({0, ds_bgn}, {1,idx}).write(res_meta.col(0).data());
    d_universe.select({0, ds_bgn}, {1,idx}).write(res_meta.col(1).data());

    HighFive::DataSet d_xmin     = file->getDataSet("xmin");
    HighFive::DataSet d_ymin     = file->getDataSet("ymin");
    HighFive::DataSet d_chi2     = file->getDataSet("thischi2");
    HighFive::DataSet d_minchi2  = file->getDataSet("minchi2");
    HighFive::DataSet d_minchi2pf= file->getDataSet("minchi2pf");
    HighFive::DataSet d_xtrue    = file->getDataSet("xtrue");
    HighFive::DataSet d_ytrue    = file->getDataSet("ytrue");

    d_xmin     .select({0, ds_bgn}, {1,idx}).write(res_val.col(0).data());
    d_ymin     .select({0, ds_bgn}, {1,idx}).write(res_val.col(1).data());
    d_chi2     .select({0, ds_bgn}, {1,idx}).write(res_val.col(2).data());
    d_minchi2  .select({0, ds_bgn}, {1,idx}).write(res_val.col(3).data());
    d_minchi2pf.select({0, ds_bgn}, {1,idx}).write(res_val.col(4).data());

    d_xtrue    .select({0, ds_bgn}, {1,idx}).write(res_val.col(5).data());
    d_ytrue    .select({0, ds_bgn}, {1,idx}).write(res_val.col(6).data());
}

void readConfig(const char* fname, int & nbx, double & xmin, double & xmax, int & nby, double & ymin, double & ymax) {
    std::ifstream f(fname);
    f >> nbx;
    f >> xmin;
    f >> xmax;
    f >> nby;
    f >> ymin;
    f >> ymax;
}

int main(int argc, char* argv[]) {
    diy::mpi::environment env(argc, argv);
    diy::mpi::communicator world;

    std::vector<double> v_buff;
    int ncols, nrows;
    if (world.rank()==0) loadData(argv[1], "covmat",      v_buff,   nrows, ncols);
    MatrixXd covmat = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(argv[1], "core",        v_buff,   nrows, ncols);
    VectorXd core = bcMatrixXd(world, v_buff, nrows, ncols);
    if (world.rank()==0) loadData(argv[1], "invcovbg",      v_buff,   nrows, ncols);
    MatrixXd invcovbg = bcMatrixXd(world, v_buff, nrows, ncols);

    auto config = mkConfig();
    SignalGenerator::SignalGenerator sig(mkAppEval(world, argv[2]), mkAppEval(world, argv[3]), core, config);

    double xmin, xmax, ymin, ymax;
    int nbx, nby;
    readConfig(argv[4], nbx, xmin, xmax, nby, ymin, ymax);
    MyGrid MG = mkGrid(nbx, xmin, xmax, nby, ymin, ymax);

    int nuniv = atoi(argv[5]);
    std::string out_file=argv[6];

    HighFive::File* f_out  = new HighFive::File(out_file,
                        HighFive::File::ReadWrite|HighFive::File::Create|HighFive::File::Truncate,
                        HighFive::MPIOFileDriver(MPI_COMM_WORLD,MPI_INFO_NULL));

    writeGrid(f_out, MG.points);
    createDataSets(f_out, MG.points.rows(), nuniv);

    size_t blocks = world.size();//nPoints;//*nUniverses;
    if (world.rank()==0) fmt::print(stderr, "FC will be done on {} blocks, distributed over {} ranks\n", blocks, world.size());
    Bounds fc_domain(1);
    fc_domain.min[0] = 0;
    fc_domain.max[0] = blocks-1;
    diy::RoundRobinAssigner        fc_assigner(world.size(), blocks);
    diy::Master                    fc_master(world, 1, -1, &FitBlock::create, &FitBlock::destroy);
    diy::decompose(1, world.rank(), fc_domain, fc_assigner, fc_master);

    fc_master.foreach([world, sig, covmat, invcovbg, config, MG, nuniv, f_out](FitBlock* b, const diy::Master::ProxyWithLink& cp) {runMin(b, cp, sig, covmat, invcovbg, config, MG, nuniv, f_out); });
    //fc_master.foreach([world, sig, covmat, invcovbg, config, MG, nuniv, f_out](FitBlock* b, const diy::Master::ProxyWithLink& cp) {runGrid(b, cp, sig, covmat, invcovbg, config, MG, nuniv, f_out); });

    delete f_out;
    return 0;
}

ext: cxx

max_stars_repo_name: iamholger/ProblemSBN

max_issues_repo_name: iamholger/ProblemSBN

max_forks_repo_name: iamholger/ProblemSBN


text:
#pragma once

#include <boost/asio/ssl/context.hpp>

/*  Load a signed certificate into the ssl context, and configure
    the context for use with a server.
*/
namespace test::helpers
{
void load_server_certificate(boost::asio::ssl::context& ctx);
}

ext: hpp

max_stars_repo_name: TankerHQ/fetchpp

max_issues_repo_name: TankerHQ/fetchpp

max_forks_repo_name: TankerHQ/fetchpp


text:
#ifndef PICSAR_MULTIPHYSICS_SPECIAL_FUNCTIONS
#define PICSAR_MULTIPHYSICS_SPECIAL_FUNCTIONS

//Should be included by all the src files of the library
#include "picsar_qed/qed_commons.h"

//This .hpp file is an extremely thin wrapper around special functions
//(Bessel functions for now) defined either in the STL (if C++17 is available)
//or in Boost library as a fallback.
#ifdef PXRMP_USE_CXX17_FOR_SPECIAL_FUNCTIONS
    #include <cmath>
#else
    #include <boost/math/special_functions/bessel.hpp>
#endif

namespace picsar{
namespace multi_physics{
namespace math{

    /**
    * This function is a wrapper around the Bessel function
    * of the second kind defined either in the STL (if C++17 is available)
    * or in Boost library as a fallback (not usable on GPUs).
    *
    * @tparam RealType the floating point type to be used
    * @param[in] v order of the function
    * @param[in] x argument of the function
    * @return K_v(x)
    */
    template<typename RealType>
    inline RealType k_v(RealType v, RealType x)
    {
#ifdef PXRMP_USE_CXX17_FOR_SPECIAL_FUNCTIONS
        return std::cyl_bessel_k(v, x);
#else
        return boost::math::cyl_bessel_k(v, x);
#endif
    }

}
}
}

#endif //PICSAR_MULTIPHYSICS_SPECIAL_FUNCTIONS


text:
/*************************************************************************
 * Copyright (C) 2017-2019 Barcelona Supercomputing Center               *
 *                         Centro Nacional de Supercomputacion           *
 * All rights reserved.                                                  *
 *                                                                       *
 * This file is part of NORNS, a service that allows other programs to   *
 * start, track and manage asynchronous transfers of data resources      *
 * between different storage backends.                                   *
 *                                                                       *
 * See AUTHORS file in the top level directory for information regarding *
 * developers and contributors.                                          *
 *                                                                       *
 * This software was developed as part of the EC H2020 funded project    *
 * NEXTGenIO (Project ID: 671951).                                       *
 *     www.nextgenio.eu                                                  *
 *                                                                       *
 * Permission is hereby granted, free of charge, to any person obtaining *
 * a copy of this software and associated documentation files (the       *
 * "Software"), to deal in the Software without restriction, including   *
 * without limitation the rights to use, copy, modify, merge, publish,   *
 * distribute, sublicense, and/or sell copies of the Software, and to    *
 * permit persons to whom the Software is furnished to do so, subject to *
 * the following conditions:                                             *
 *                                                                       *
 * The above copyright notice and this permission notice shall be        *
 * included in all copies or substantial portions of the Software.       *
 *                                                                       *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,       *
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF    *
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND                 *
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS   *
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN    *
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN     *
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE      *
 * SOFTWARE.                                                             *
 *************************************************************************/

#include <stdio.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <cstdlib>
#include <exception>
#include <dirent.h>
#include <fstream>
#include <fcntl.h>
#include <list>
#include <signal.h>
#include <string.h>
#include <ctime>

#ifdef __LOGGER_ENABLE_DEBUG__
#include <sys/prctl.h>
#endif

#include <boost/optional.hpp>
#include <boost/optional/optional_io.hpp>
#include <boost/atomic.hpp>
#include <functional>

#include "common.hpp"
#include "api.hpp"
#include "backends.hpp"
#include "logger.hpp"
#include "job.hpp"
#include "resources.hpp"
#include "io.hpp"
#include "namespaces.hpp"
#include "fmt.hpp"
#include "hermes.hpp"
#include "rpcs.hpp"
#include "context.hpp"
#include "urd.hpp"

namespace norns {

urd::urd() :
    m_is_paused(false),
    m_settings(std::make_shared<config::settings>()) {}

urd::~urd() {}

pid_t urd::daemonize() {
    /*
     * --- Daemonize structure ---
     *  Check if this is already a daemon
     *  Fork off praent process
     *  Obtain new process group
     *  Close all descriptors
     *  Handle standard IO
     *  Change file mode mask
     *  Change the current working directory
     *  Check if daemon already exists
     *  Manage signals
     */

    pid_t pid, sid;

    /* Check if this is already a daemon */
    if(getppid() == 1) {
        return 0;
    }

    // We need to destroy the global logger before calling fork. Otherwise the
    // logger will not function properly since its internal thread will not
    // be duplicated by fork(). Furthermore, if we don't destroy pre-fork()
    // and attempt to replace it post-fork(), the logger destructor will attempt
    // to join the (now invalid) thread and end up blocking forever. To avoid
    // this (and since we want to be able to output messages from all
    // processes), we destroy it now and recreate it post-fork() both in the
    // parent process and in the child.
    logger::destroy_global_logger();

    /* Fork off the parent process */
    pid = fork();

    // re-initialize logging facilities (post-fork)
    init_logger();

    if(pid < 0) {
        LOGGER_ERRNO("Failed to create child process");
        exit(EXIT_FAILURE);
    }

    /* Parent returns to caller */
    if(pid != 0) {
        return pid;
    }

    /* Become a session and process group leader with no controlling tty */
    if((sid = setsid()) < 0) {
        /* Log failure */
        LOGGER_ERRNO("Failed to disassociate controlling tty");
        exit(EXIT_FAILURE);
    }

    /* Handle standard IO: discard data to/from stdin, stdout and stderr */
    int dev_null;

    if((dev_null = open("/dev/null", O_RDWR)) == -1) {
        LOGGER_ERRNO("Failed to open \"/dev/null\"");
        exit(EXIT_FAILURE);
    }

    if(dup2(dev_null, STDIN_FILENO) == -1) {
        LOGGER_ERRNO("Failed to dup \"/dev/null\" onto stdin");
        exit(EXIT_FAILURE);
    }

    if(dup2(dev_null, STDOUT_FILENO) == -1) {
        LOGGER_ERRNO("Failed to dup \"/dev/null\" onto stdout");
        exit(EXIT_FAILURE);
    }

    if(dup2(dev_null, STDERR_FILENO) == -1) {
        LOGGER_ERRNO("Failed to dup \"/dev/null\" onto stderr");
        exit(EXIT_FAILURE);
    }

    /* Change the file mode creation mask */
    umask(0);

    /* ensure the process does not keep a directory in use,
     * avoid relative paths beyond this point! */
    if(chdir("/") < 0) {
        LOGGER_ERRNO("Failed to change working directory to root directory");
        exit(EXIT_FAILURE);
    }

    /* Check if daemon already exists:
     * First instance of the daemon will lock the file so that other
     * instances understand that an instance is already running.
     */
    int pfd;

    if((pfd = open(m_settings->pidfile().c_str(), O_RDWR | O_CREAT, 0640)) == -1) {
        LOGGER_ERRNO("Failed to create daemon lock file");
        exit(EXIT_FAILURE);
    }

    if(lockf(pfd, F_TLOCK, 0) < 0) {
        LOGGER_ERRNO("Failed to acquire lock on pidfile");
        LOGGER_ERROR("Another instance of this daemon may already be running");
        exit(EXIT_FAILURE);
    }

    /* record pid in lockfile */
    std::string pidstr(std::to_string(getpid()));

    if(write(pfd, pidstr.c_str(), pidstr.length()) !=
            static_cast<ssize_t>(pidstr.length())) {
        LOGGER_ERRNO("Failed to write pidfile");
        exit(EXIT_FAILURE);
    }

    close(pfd);
    close(dev_null);

    /* Manage signals */
    signal(SIGCHLD, SIG_IGN); /* ignore child */
    signal(SIGTSTP, SIG_IGN); /* ignore tty signals */
    signal(SIGTTOU, SIG_IGN);
    signal(SIGTTIN, SIG_IGN);
//  signal(SIGHUP, signal_handler); /* catch hangup signal */
//  signal(SIGTERM, signal_handler); /* catch kill signal */

    return 0;
}

urd_error urd::validate_iotask_args(iotask_type type,
                                    const resource_info_ptr& src_rinfo,
                                    const resource_info_ptr& dst_rinfo) const {

    if(type != iotask_type::copy &&
       type != iotask_type::move &&
       type != iotask_type::remove) {
        return urd_error::bad_args;
    }

    if(src_rinfo->is_remote() && dst_rinfo->is_remote()) {
        return urd_error::not_supported;
    }

//    // dst_resource cannot be a memory region
//    if(dst_rinfo->type() == data::resource_type::memory_region) {
//        return urd_error::not_supported;
//    }

    return urd_error::success;
}

///////////////////////////////////////////////////////////////////////////////
//                 handlers for user requests
///////////////////////////////////////////////////////////////////////////////

response_ptr
urd::iotask_create_handler(const request_ptr base_request) {

    // downcast the generic request to the concrete implementation
    auto request =
        utils::static_unique_ptr_cast<api::iotask_create_request>(
                std::move(base_request));

    const auto type = request->get<0>();
    const auto src_rinfo = request->get<1>();
    const auto dst_rinfo = request->get<2>().get_value_or(nullptr);

    std::vector<std::string> nsids;
    std::vector<bool> remotes;
    std::vector<std::shared_ptr<storage::backend>> backend_ptrs;
    std::vector<std::shared_ptr<data::resource_info>> rinfo_ptrs;
    boost::optional<iotask_id> tid;
    boost::optional<auth::credentials> auth;
    boost::optional<io::generic_task> t;
    response_ptr resp;
    urd_error rv = urd_error::success;

    if(m_is_paused) {
        rv = urd_error::accept_paused;
        goto log_and_return;
    }

    auth = request->credentials();

    if(!auth) {
        LOGGER_CRITICAL("Request without credentials");
        rv = urd_error::snafu; // TODO: invalid_credentials? eaccess? eperm?
        goto log_and_return;
    }

//    if(src_rinfo->is_remote()) {
//        rv = urd_error::not_supported;
//        goto log_and_return;
//    }

    for(const auto& rinfo : {src_rinfo, dst_rinfo}) {
        if(rinfo) {
            nsids.push_back(rinfo->nsid());
            remotes.push_back(rinfo->is_remote());
            rinfo_ptrs.emplace_back(rinfo);
        }
    }

#ifdef __LOGGER_ENABLE_DEBUG__
    LOGGER_DEBUG("Request metadata:");
    LOGGER_DEBUG("  rinfos: {} {}", src_rinfo, dst_rinfo);

    LOGGER_DEBUG("  nsids: {");
    for(std::size_t i=0; i<nsids.size(); ++i) {
        LOGGER_DEBUG("    \"{}\"", nsids[i]);
    }
    LOGGER_DEBUG("  }");

    LOGGER_DEBUG("  remotes: {");
    for(std::size_t i=0; i<remotes.size(); ++i) {
        LOGGER_DEBUG("    {}", remotes[i]);
    }
    LOGGER_DEBUG("  }");
#endif

    {
        bool all_found = false;
        boost::shared_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);
        std::tie(all_found, backend_ptrs) = m_namespace_mgr->find(nsids, remotes);

        if(!all_found) {
            rv = urd_error::no_such_namespace;
            goto log_and_return;
        }
    }

#ifdef __LOGGER_ENABLE_DEBUG__
    for(std::size_t i=0; i<nsids.size(); ++i) {
        LOGGER_DEBUG("nsid: {}, remote?: {}, bptr: {}", nsids[i], remotes[i], backend_ptrs[i]);
    }
#endif

    //FIXME: use appropriate args for each task rather than a vector of nullptrs
    switch(type) {
        case iotask_type::move:
        case iotask_type::copy:
            std::tie(rv, t) =
                m_task_mgr->create_local_initiated_task(type, *auth, backend_ptrs, rinfo_ptrs);
            break;
        case iotask_type::remove:
            std::tie(rv, t) =
                m_task_mgr->create_local_initiated_task(type, *auth, backend_ptrs, rinfo_ptrs);
            break;
        case iotask_type::noop:
            std::tie(rv, t) =
                m_task_mgr->create_local_initiated_task(type, *auth, backend_ptrs, rinfo_ptrs);
            break;
        default:
            rv = urd_error::bad_args;
            goto log_and_return;
    }

    if(rv == urd_error::success) {
        tid = t->id();

        // enqueue task so that it's eventually run by a worker thread
        m_task_mgr->enqueue_task(std::move(*t));
    }

//    std::tie(rv, tid) = m_task_mgr->create_task(type, *auth, backend_ptrs, rinfo_ptrs);

log_and_return:
    resp = std::make_unique<api::iotask_create_response>(tid.get_value_or(0));
    resp->set_error_code(rv);

    LOGGER_INFO("IOTASK_CREATE({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr
urd::iotask_status_handler(const request_ptr base_request) const {

    auto resp = std::make_unique<api::iotask_status_response>();

    // downcast the generic request to the concrete implementation
    auto request =
        utils::static_unique_ptr_cast<api::iotask_status_request>(
                std::move(base_request));

    auto task_info_ptr = m_task_mgr->find(request->get<0>());

    if(task_info_ptr) {
        resp->set_error_code(urd_error::success);

        // stats provides a thread-safe view of a task status
        // (locking is done internally)
        resp->set<0>(task_info_ptr->stats());

        if(task_info_ptr->status() == io::task_status::finished ||
           task_info_ptr->status() == io::task_status::finished_with_error) {
            m_task_mgr->erase(request->get<0>());
        }

    }
    else {
        resp->set_error_code(urd_error::no_such_task);
    }

    LOGGER_INFO("IOTASK_STATUS({}) = {}",
                request->to_string(), resp->to_string());

    return std::move(resp);
}

///////////////////////////////////////////////////////////////////////////////
//                 handlers for control requests
///////////////////////////////////////////////////////////////////////////////

response_ptr urd::ping_handler(const request_ptr /*base_request*/) {
    response_ptr resp = std::make_unique<api::ping_response>();

    resp->set_error_code(urd_error::success);

    LOGGER_INFO("PING_REQUEST() = {}", resp->to_string());
    return resp;
}

response_ptr urd::job_register_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::job_register_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::job_register_request>(std::move(base_request));

    uint32_t jobid = request->get<0>();
    auto hosts = request->get<1>();

    {
        boost::unique_lock<boost::shared_mutex> lock(m_jobs_mutex);

        if(m_jobs.find(jobid) != m_jobs.end()) {
            resp->set_error_code(urd_error::job_exists);
        }
        else {
            m_jobs.emplace(jobid, std::make_shared<job>(jobid, hosts));
            resp->set_error_code(urd_error::success);
        }
    }

    LOGGER_INFO("REGISTER_JOB({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr urd::job_update_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::job_update_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::job_update_request>(std::move(base_request));

    uint32_t jobid = request->get<0>();
    auto hosts = request->get<1>();

    {
        boost::unique_lock<boost::shared_mutex> lock(m_jobs_mutex);

        const auto& it = m_jobs.find(jobid);

        if(it == m_jobs.end()) {
            resp->set_error_code(urd_error::no_such_job);
        }
        else {
            it->second->update(hosts);
            resp->set_error_code(urd_error::success);
        }
    }

    LOGGER_INFO("UPDATE_JOB({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr urd::job_remove_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::job_unregister_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::job_unregister_request>(std::move(base_request));

    uint32_t jobid = request->get<0>();

    {
        boost::unique_lock<boost::shared_mutex> lock(m_jobs_mutex);

        const auto& it = m_jobs.find(jobid);

        if(it == m_jobs.end()) {
            resp->set_error_code(urd_error::no_such_job);
        }
        else {
            m_jobs.erase(it);
            resp->set_error_code(urd_error::success);
        }
    }

    LOGGER_INFO("UNREGISTER_JOB({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr urd::process_add_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::process_register_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::process_register_request>(std::move(base_request));

    uint32_t jobid = request->get<0>();
    //pid_t uid = request->get<1>();
    gid_t gid = request->get<2>();
    pid_t pid = request->get<3>();

    boost::unique_lock<boost::shared_mutex> lock(m_jobs_mutex);

    const auto& it = m_jobs.find(jobid);

    if(it == m_jobs.end()) {
        resp->set_error_code(urd_error::no_such_job);
        goto log_and_return;
    }

    it->second->add_process(pid, gid);
    resp->set_error_code(urd_error::success);

log_and_return:
    LOGGER_INFO("ADD_PROCESS({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr urd::process_remove_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::process_unregister_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::process_unregister_request>(std::move(base_request));

    uint32_t jobid = request->get<0>();
    //pid_t uid = request->get<1>();
    gid_t gid = request->get<2>();
    pid_t pid = request->get<3>();

    boost::unique_lock<boost::shared_mutex> lock(m_jobs_mutex);

    const auto& it = m_jobs.find(jobid);

    if(it == m_jobs.end()) {
        resp->set_error_code(urd_error::no_such_job);
        goto log_and_return;
    }

    if(it->second->find_and_remove_process(pid, gid)) {
        resp->set_error_code(urd_error::success);
    }
    else {
        resp->set_error_code(urd_error::no_such_process);
    }

log_and_return:
    LOGGER_INFO("REMOVE_PROCESS({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

urd_error urd::create_namespace(const config::namespace_def& nsdef) {

    const auto type = storage::backend_factory::get().
        get_type(nsdef.alias());

    if(type == backend_type::unknown) {
        return urd_error::bad_args;
    }

    return create_namespace(nsdef.nsid(), type, nsdef.track(),
                            nsdef.mountpoint(), nsdef.capacity());
}

urd_error urd::create_namespace(const std::string& nsid, backend_type type,
                                bool track, const bfs::path& mount,
                                uint32_t quota) {

    boost::unique_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);

    if(m_namespace_mgr->contains(nsid)) {
        return urd_error::namespace_exists;
    }

    if(auto bptr = storage::backend_factory::create_from(
                type, nsid, track, mount, quota)) {
        m_namespace_mgr->add(nsid, bptr);
        return urd_error::success;
    }

    return urd_error::bad_args;
}

response_ptr urd::namespace_register_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::backend_register_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::backend_register_request>(std::move(base_request));

    std::string nsid = request->get<0>();
    backend_type type = request->get<1>();
    bool track = request->get<2>();
    std::string mount = request->get<3>();
    int32_t quota = request->get<4>();

    resp->set_error_code(create_namespace(nsid, type, track, mount, quota));

    LOGGER_INFO("REGISTER_NAMESPACE({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

/* XXX not supported yet
response_ptr urd::namespace_update_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::backend_update_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::backend_update_request>(std::move(base_request));

    resp->set_error_code(urd_error::success);

log_and_return:
    LOGGER_INFO("UPDATE_NAMESPACE({}) = {}", request->to_string(), resp->to_string());
    return resp;
}*/

response_ptr urd::namespace_remove_handler(const request_ptr base_request) {

    response_ptr resp = std::make_unique<api::backend_unregister_response>();

    // downcast the generic request to the concrete implementation
    auto request = utils::static_unique_ptr_cast<api::backend_unregister_request>(std::move(base_request));

    std::string nsid = request->get<0>();

    {
        boost::unique_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);

        if(m_namespace_mgr->remove(nsid)) {
            resp->set_error_code(urd_error::success);
        }
        else {
            resp->set_error_code(urd_error::no_such_namespace);
        }
    }

    LOGGER_INFO("UNREGISTER_NAMESPACE({}) = {}", request->to_string(), resp->to_string());
    return resp;
}

response_ptr urd::global_status_handler(const request_ptr /*base_request*/) {

    auto resp = std::make_unique<api::global_status_response>();

    resp->set_error_code(urd_error::success);
    resp->set<0>(m_task_mgr->global_stats());

    LOGGER_INFO("GLOBAL_STATUS() = {}", resp->to_string());
    return std::move(resp);
}

response_ptr
urd::command_handler(const request_ptr base_request) {

    // downcast the generic request to the concrete implementation
    auto request =
        utils::static_unique_ptr_cast<api::command_request>(
                std::move(base_request));
    auto resp = std::make_unique<api::command_response>();
    resp->set_error_code(urd_error::success);

    switch(request->get<0>()) {
        case command_type::ping:
            break; // nothing special to do here
        case command_type::pause_listen:
            pause_listening();
            break;
        case command_type::resume_listen:
            resume_listening();
            break;
        case command_type::shutdown:
        {
            LOGGER_WARN("Shutdown requested!");
            pause_listening();

            const auto rv = check_shutdown();
            resp->set_error_code(rv);

            if(rv != urd_error::success) {
                resume_listening();
                break;
            }
            shutdown();
            break;
        }
        case command_type::unknown:
            resp->set_error_code(urd_error::bad_args);
            break;
    }

    LOGGER_INFO("COMMAND({}) = {}", request->to_string(), resp->to_string());
    return std::move(resp);
}

response_ptr urd::unknown_request_handler(const request_ptr /*base_request*/) {
    response_ptr resp = std::make_unique<api::bad_request_response>();

    resp->set_error_code(urd_error::bad_request);

    LOGGER_INFO("UNKNOWN_REQUEST() = {}", resp->to_string());
    return resp;
}

// N.B. This function is called by the progress thread internal to
// m_network_service rather than by the main execution thread
void
urd::push_resource_handler(hermes::request<rpc::push_resource>&& req) {

    const auto args = req.args();

    LOGGER_WARN("incoming rpc::push_resource(from: \"{}@{}:{}\", to: \"{}:{}\")",
                args.in_nsid(),
                args.in_address(),
                args.in_resource_name(),
                args.out_nsid(),
                args.out_resource_name());

    urd_error rv = urd_error::success;
    boost::optional<io::generic_task> t;
    std::shared_ptr<storage::backend> src_backend;
    boost::optional<std::shared_ptr<storage::backend>> dst_backend;
    auto dst_rtype = static_cast<data::resource_type>(args.out_resource_type());
    auth::credentials auth; //XXX fake credentials for now

    const auto create_rinfo =
        [&](const data::resource_type& rtype) ->
            std::shared_ptr<data::resource_info> {

        switch(rtype) {
            case data::resource_type::remote_resource:
                return std::make_shared<data::remote_resource_info>(
                        args.in_address(), args.in_nsid(),
                        args.in_is_collection(), args.in_resource_name(),
                        args.in_buffers());
            case data::resource_type::local_posix_path:
            case data::resource_type::shared_posix_path:
                return std::make_shared<data::local_path_info>(
                        args.out_nsid(), args.out_resource_name());
            default:
                rv = urd_error::not_supported;
                return {};
        }
    };

    if(m_is_paused) {
        rv = urd_error::accept_paused;
        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                    static_cast<uint32_t>(io::task_status::finished_with_error),
                    static_cast<uint32_t>(rv),
                    0,
                    0);
        return;
    }

    auto src_rinfo = create_rinfo(data::resource_type::remote_resource);
    auto dst_rinfo = create_rinfo(dst_rtype);

    //TODO: check src_rinfo and dst_rinfo

    // TODO: actually retrieve and validate credentials, etc

    src_backend =
        std::make_shared<storage::detail::remote_backend>(args.in_nsid());

    {
        boost::shared_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);
        dst_backend = m_namespace_mgr->find(args.out_nsid());
    }

    if(!dst_backend) {
        rv = urd_error::no_such_namespace;
        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                static_cast<uint32_t>(io::task_status::finished_with_error),
                static_cast<int32_t>(rv),
                0,
                0);
        return;
    }

    LOGGER_DEBUG("nsid: {}, bptr: {}", args.in_nsid(), dst_backend);

    auto ctx =
        std::make_shared<hermes::request<rpc::push_resource>>(std::move(req));

    std::tie(rv, t) =
        m_task_mgr->create_remote_initiated_task(
                iotask_type::remote_transfer, auth,
                ctx, src_backend, src_rinfo,
                *dst_backend, dst_rinfo);

    if(rv == urd_error::success) {
        // run the task and check that it started correctly
        (*t)();

        auto req = std::move(*ctx);

        if(t->info()->status() == io::task_status::finished_with_error) {
            rv = t->info()->task_error();
            LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
            m_network_service->respond(std::move(req),
                    static_cast<uint32_t>(io::task_status::finished_with_error),
                    static_cast<int32_t>(t->info()->task_error()),
                    static_cast<int32_t>(t->info()->sys_error().value()),
                    0);
        }
    }
}

void
urd::pull_resource_handler(hermes::request<rpc::pull_resource>&& req) {

    const auto args = req.args();

    LOGGER_WARN("incoming rpc::push_request(from: \"{}:{}\", to: \"{}@{}:{}\")",
                args.in_nsid(),
                args.in_resource_name(),
                args.out_address(),
                args.out_nsid(),
                args.out_resource_name());

    urd_error rv = urd_error::success;
    boost::optional<io::generic_task> tsk;
    boost::optional<std::shared_ptr<storage::backend>> src_backend;
    std::shared_ptr<storage::backend> dst_backend;

    /// XXX this information should be retrievable from the backend
    auto src_rtype = static_cast<data::resource_type>(args.in_resource_type());

    auth::credentials auth; //XXX fake credentials for now

    const auto create_rinfo =
        [&](const data::resource_type& rtype) ->
            std::shared_ptr<data::resource_info> {

        switch(rtype) {
            case data::resource_type::remote_resource:
                return std::make_shared<data::remote_resource_info>(
                        args.out_address(), args.out_nsid(), false,
                        args.out_resource_name(), args.out_buffers());
            case data::resource_type::local_posix_path:
            case data::resource_type::shared_posix_path:
                return std::make_shared<data::local_path_info>(
                        args.in_nsid(), args.in_resource_name());
            default:
                rv = urd_error::not_supported;
                return {};
        }
    };

    if(m_is_paused) {
        rv = urd_error::accept_paused;
        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                    static_cast<uint32_t>(io::task_status::finished_with_error),
                    static_cast<uint32_t>(rv),
                    0,
                    0);
        return;
    }

    auto src_rinfo = create_rinfo(src_rtype);
    auto dst_rinfo = create_rinfo(data::resource_type::remote_resource);

    //TODO: check src_rinfo and dst_rinfo

    // TODO: actually retrieve and validate credentials, etc

    {
        boost::shared_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);
        src_backend = m_namespace_mgr->find(args.in_nsid());
    }

    if(!src_backend) {
        rv = urd_error::no_such_namespace;
        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                static_cast<uint32_t>(io::task_status::finished_with_error),
                static_cast<int32_t>(rv),
                0,
                0);
        return;
    }

    LOGGER_DEBUG("nsid: {}, bptr: {}", args.in_nsid(), src_backend);

    dst_backend =
        std::make_shared<storage::detail::remote_backend>(args.out_nsid());

    auto ctx =
        std::make_shared<hermes::request<rpc::pull_resource>>(std::move(req));

    std::tie(rv, tsk) =
        m_task_mgr->create_remote_initiated_task(
                iotask_type::remote_transfer, auth,
                ctx, *src_backend, src_rinfo,
                dst_backend, dst_rinfo);

    if(rv == urd_error::success) {
        // run the task and check that it started correctly
        (*tsk)();

        auto req = std::move(*ctx);

        if(tsk->info()->status() == io::task_status::finished_with_error) {
            rv = tsk->info()->task_error();
            LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
            m_network_service->respond(std::move(req),
                    static_cast<uint32_t>(io::task_status::finished_with_error),
                    static_cast<int32_t>(tsk->info()->task_error()),
                    static_cast<int32_t>(tsk->info()->sys_error().value()),
                    0);
        }
    }
}

void
urd::stat_resource_handler(hermes::request<rpc::stat_resource>&& req) {

    const auto args = req.args();

    LOGGER_WARN("incoming rpc::stat_resource(\"{}:{}\")",
                args.nsid(),
                args.resource_name());

    urd_error rv = urd_error::success;
    boost::optional<std::shared_ptr<storage::backend>> dst_backend;

    {
        boost::shared_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);
        dst_backend = m_namespace_mgr->find(args.nsid());
    }

    if(!dst_backend) {
        rv = urd_error::no_such_namespace;
        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                                    static_cast<uint32_t>(rv),
                                    false,
                                    //XXX ENOENT should not be required:
                                    // the transfer() interface should be
                                    // richer rather than returning only a
                                    // std::error_code
                                    ENOENT,
                                    0);
        return;
    }

    auto rtype = static_cast<data::resource_type>(args.resource_type());

    const auto create_rinfo =
        [&](const data::resource_type& rtype) ->
            std::shared_ptr<data::resource_info> {

        switch(rtype) {
            case data::resource_type::local_posix_path:
            case data::resource_type::shared_posix_path:
                return std::make_shared<data::local_path_info>(
                        args.nsid(), args.resource_name());
            default:
                return {};
        }
    };

    auto rinfo = create_rinfo(rtype);

    if(!rinfo) {
        LOGGER_ERROR("Failed to access resource {}", rinfo->to_string());
        rv = urd_error::not_supported;

        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                                    static_cast<uint32_t>(rv),
                                    //XXX EOPNOTSUPP should not be required:
                                    // the transfer() interface should be
                                    // richer rather than returning only a
                                    // std::error_code
                                    EOPNOTSUPP,
                                    false,
                                    0);
        return;
    }

    std::error_code ec;
    auto rsrc = (*dst_backend)->get_resource(rinfo, ec);

    if(ec) {
        LOGGER_ERROR("Failed to access resource {}", rinfo->to_string());
        rv = urd_error::no_such_resource;

        LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
        m_network_service->respond(std::move(req),
                                    static_cast<uint32_t>(rv),
                                    ec.value(),
                                    false,
                                    0);
        return;
    }

    LOGGER_INFO("IOTASK_RECEIVE() = {}", utils::to_string(rv));
    m_network_service->respond(std::move(req),
            static_cast<uint32_t>(urd_error::success),
            0,
            rsrc->is_collection(),
            rsrc->packed_size());
}

void urd::configure(const config::settings& settings) {
    m_settings = std::make_shared<config::settings>(settings);
}

config::settings urd::get_configuration() const {
    return *m_settings;
}

void urd::signal_handler(int signum){

    switch(signum) {

        case SIGINT:
            LOGGER_WARN("A signal (SIGINT) occurred.");
            shutdown();
            break;

        case SIGTERM:
            LOGGER_WARN("A signal (SIGTERM) occurred.");
            shutdown();
            break;

        case SIGHUP:
            LOGGER_WARN("A signal (SIGHUP) occurred.");
            break;
    }
}

void urd::init_logger() {

    if(m_settings->use_console()) {
        logger::create_global_logger(m_settings->progname(), "console color");
        return;;
    }

    if(m_settings->use_syslog()) {
        logger::create_global_logger(m_settings->progname(), "syslog");

        if(!m_settings->daemonize()) {
            fmt::print(stderr, "PSA: Output sent to syslog while in "
                               "non-daemon mode\n");
        }

        return;
    }

    if(!m_settings->log_file().empty()) {
        logger::create_global_logger(m_settings->progname(), "file",
                m_settings->log_file());
        return;
    }

    logger::create_global_logger(m_settings->progname(), "console color");
}

void urd::init_event_handlers() {

    LOGGER_INFO(" * Creating event listener...");

    // create (but not start) the API listener
    // and register handlers for each request type
    try {
        m_ipc_service = std::make_unique<api_listener>();
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create the event listener. This should "
                     "not happen under normal conditions.");
        exit(EXIT_FAILURE);
    }

    boost::system::error_code ec;
    mode_t old_mask = ::umask(0);

    // setup socket for control API
    // (note that if we find any socket files at this point, where no pidfile
    // was found during initialization, they must be stale sockets from
    // another run. Just remove them).
    if(bfs::exists(m_settings->control_socket())) {
        bfs::remove(m_settings->control_socket(), ec);

        if(ec) {
            LOGGER_ERROR("Failed to remove stale control API socket: {}",
                    ec.message());
            teardown();
            exit(EXIT_FAILURE);
        }
    }

    ::umask(S_IXUSR | S_IRWXG | S_IRWXO); // u=rw-, g=---, o=---

    try {
        m_ipc_service->register_endpoint(m_settings->control_socket());
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create control API socket: {}", e.what());
        teardown();
        exit(EXIT_FAILURE);
    }

    // setup socket for user API
    if(bfs::exists(m_settings->global_socket())) {
        bfs::remove(m_settings->global_socket(), ec);

        if(ec) {
            LOGGER_ERROR("Failed to remove stale user API socket: {}",
                    ec.message());
            teardown();
            exit(EXIT_FAILURE);
        }
    }

    ::umask(S_IXUSR | S_IXGRP | S_IXOTH); // u=rw-, g=rw-, o=rw-

    try {
        m_ipc_service->register_endpoint(m_settings->global_socket());
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create user API socket: {}", e.what());
        teardown();
        exit(EXIT_FAILURE);
    }

    // restore the umask
    ::umask(old_mask);

    try {

        const std::string bind_address =
            m_settings->bind_address() + ":" +
            std::to_string(m_settings->remote_port());

        m_network_service =
            std::make_shared<hermes::async_engine>(
                    hermes::transport::ofi_tcp,
                    bind_address,
                    true);
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create remote listener: {}", e.what());
        teardown();
        exit(EXIT_FAILURE);
    }

#if 0
    // setup socket for remote connections
    try {
        m_ipc_service->register_endpoint(m_settings->remote_port());
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create socket for remote connections: {}",
                e.what());
        teardown();
        exit(EXIT_FAILURE);
    }
#endif

    LOGGER_INFO(" * Installing message handlers...");

    /* user-level functionalities */
    m_ipc_service->register_callback(
            api::request_type::iotask_create,
            std::bind(&urd::iotask_create_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::iotask_status,
            std::bind(&urd::iotask_status_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::ping,
            std::bind(&urd::ping_handler, this, std::placeholders::_1));

    /* admin-level functionalities */
    m_ipc_service->register_callback(
            api::request_type::job_register,
            std::bind(&urd::job_register_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::job_update,
            std::bind(&urd::job_update_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::job_unregister,
            std::bind(&urd::job_remove_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::process_register,
            std::bind(&urd::process_add_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::process_unregister,
            std::bind(&urd::process_remove_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
        api::request_type::backend_register,
        std::bind(&urd::namespace_register_handler, this,
                  std::placeholders::_1));

    /*    m_ipc_service->register_callback(
                api::request_type::backend_update,
                std::bind(&urd::namespace_update_handler, this,
       std::placeholders::_1));*/

    m_ipc_service->register_callback(
            api::request_type::backend_unregister,
            std::bind(&urd::namespace_remove_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::global_status,
            std::bind(&urd::global_status_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::command,
            std::bind(&urd::command_handler, this, std::placeholders::_1));

    m_ipc_service->register_callback(
            api::request_type::bad_request,
            std::bind(&urd::unknown_request_handler, this, std::placeholders::_1));

    /* remote event handlers */
    m_network_service->register_handler<rpc::push_resource>(
            std::bind(&urd::push_resource_handler, this,
                      std::placeholders::_1));

    m_network_service->register_handler<rpc::pull_resource>(
            std::bind(&urd::pull_resource_handler, this,
                      std::placeholders::_1));

    m_network_service->register_handler<rpc::stat_resource>(
            std::bind(&urd::stat_resource_handler, this,
                      std::placeholders::_1));

    // signal handlers must be installed AFTER daemonizing
    LOGGER_INFO(" * Installing signal handlers...");

    m_ipc_service->set_signal_handler(
        std::bind(&urd::signal_handler, this, std::placeholders::_1),
        SIGHUP, SIGTERM, SIGINT);
}

void urd::init_namespace_manager() {

    LOGGER_INFO(" * Creating namespace manager...");

    try {
        m_namespace_mgr = std::make_unique<ns::namespace_manager>();
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create the namespace manager. This should "
                     "not happen under normal conditions.");
        exit(EXIT_FAILURE);
    }

    // pre-register special backends
    //TODO define constants
    if(!m_namespace_mgr->add("[[internal::memory]]",
                             storage::process_memory_backend)) {
        LOGGER_ERROR("Failed to register internal memory namespace");
        exit(EXIT_FAILURE);
    }
}

void urd::init_task_manager() {

    LOGGER_INFO(" * Creating task manager...");

    try {
        m_task_mgr = std::make_unique<io::task_manager>(m_settings->workers_in_pool(),
                                                        m_settings->backlog_size(),
                                                        m_settings->dry_run(),
                                                        m_settings->dry_run_duration());
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create the task manager. This should "
                     "not happen under normal conditions.");
        exit(EXIT_FAILURE);
    }
}

void urd::load_backend_plugins() {

    // register POSIX filesystem backend plugin
    storage::backend_factory::get().
        register_backend<storage::posix_filesystem>(
            backend_type::posix_filesystem,
            [](const std::string& nsid, bool track,
               const bfs::path& mount, uint32_t quota) {
                return std::shared_ptr<storage::posix_filesystem>(
                        new storage::posix_filesystem(nsid, track, mount, quota));
            });

    storage::backend_factory::get().
        register_alias("POSIX/LOCAL", backend_type::posix_filesystem);
    storage::backend_factory::get().
        register_alias("POSIX/SHARED", backend_type::posix_filesystem);

    // register NVML-DAX filesystem backend plugin
    storage::backend_factory::get().
        register_backend<storage::nvml_dax>(
            backend_type::nvml,
            [](const std::string& nsid, bool track,
               const bfs::path& mount, uint32_t quota) {
                return std::shared_ptr<storage::nvml_dax>(
                        new storage::nvml_dax(nsid, track, mount, quota));
            });

    storage::backend_factory::get().
        register_alias("NVML", backend_type::nvml);

    // register Lustre backend plugin
    storage::backend_factory::get().
        register_backend<storage::lustre>(
            backend_type::lustre,
            [](const std::string& nsid, bool track,
               const bfs::path& mount, uint32_t quota) {
                return std::shared_ptr<storage::lustre>(
                        new storage::lustre(nsid, track, mount, quota));
            });
    storage::backend_factory::get().
        register_alias("Lustre", backend_type::lustre);
}

void urd::load_transfer_plugins() {

    LOGGER_INFO(" * Loading data transfer plugins...");

    try {
        m_transferor_registry = std::make_unique<io::transferor_registry>();
    }
    catch(const std::exception& e) {
        LOGGER_ERROR("Failed to create transfer plugin registry. This should "
                     "not happen under normal conditions.");
        exit(EXIT_FAILURE);
    }

    const auto load_plugin = [&](const data::resource_type t1,
                                 const data::resource_type t2,
                          std::shared_ptr<io::transferor>&& trp) {

#if 0
        // if(!m_transferor_registry->add(t1, t2,
        //             std::forward<std::shared_ptr<io::transferor>>(trp))) {

        //     LOGGER_WARN("    Failed to load transfer plugin ({} to {}):\n"
        //                 "    Another plugin was already registered for this "
        //                 "    combination (Plugin ignored)",
        //                 utils::to_string(t1), utils::to_string(t2));
        //     return;
        // }

#else
        if(!m_task_mgr->register_transfer_plugin(t1, t2,
                    std::forward<std::shared_ptr<io::transferor>>(trp))) {
            LOGGER_WARN("    Failed to load transfer plugin ({} to {}):\n"
                        "    Another plugin was already registered for this\n"
                        "    combination (Plugin ignored)",
                        utils::to_string(t1), utils::to_string(t2));
            return;
        }
#endif

        LOGGER_INFO("    Loaded transfer plugin ({} => {})",
                    utils::to_string(t1), utils::to_string(t2));
    };

    context ctx(m_settings->staging_directory(),
                m_network_service);

    // memory region -> local path
    load_plugin(
        data::resource_type::memory_region,
        data::resource_type::local_posix_path,
        std::make_shared<io::memory_region_to_local_path_transferor>(ctx));

    // memory region -> shared path
    load_plugin(
        data::resource_type::memory_region,
        data::resource_type::shared_posix_path,
        std::make_shared<io::memory_region_to_shared_path_transferor>(ctx));

    // memory region -> remote path
    load_plugin(
        data::resource_type::memory_region,
        data::resource_type::remote_resource,
        std::make_shared<io::memory_region_to_remote_resource_transferor>(ctx));

    // local path -> local path
    load_plugin(
        data::resource_type::local_posix_path,
        data::resource_type::local_posix_path,
        std::make_shared<io::local_path_to_local_path_transferor>(ctx));

    // local path -> shared path
    load_plugin(
        data::resource_type::local_posix_path,
        data::resource_type::shared_posix_path,
        std::make_shared<io::local_path_to_shared_path_transferor>(ctx));

    // local path -> remote resource
    load_plugin(
        data::resource_type::local_posix_path,
        data::resource_type::remote_resource,
        std::make_shared<io::local_path_to_remote_resource_transferor>(ctx));

    // remote resource -> local path
    load_plugin(
        data::resource_type::remote_resource,
        data::resource_type::local_posix_path,
        std::make_shared<io::remote_resource_to_local_path_transferor>(ctx));
}

void urd::load_default_namespaces() {

    LOGGER_INFO(" * Loading default namespaces...");

    for(const auto& nsdef: m_settings->default_namespaces()) {
        if(create_namespace(nsdef) != urd_error::success) {
            LOGGER_WARN("    Failed to load namespace \"{}://\" -> {} "
                        "of type {}: Ignored", nsdef.nsid(),
                        nsdef.mountpoint(), nsdef.alias());
            continue;
        }
        LOGGER_INFO("    Loaded namespace \"{}://\" -> {} (type: {}, {})",
                    nsdef.nsid(), nsdef.mountpoint(), nsdef.alias(),
                    (nsdef.track() ? "tracked" : "untracked"));
    }
}

void
urd::check_configuration() {

    // check that the staging directory exists and that we can write to it
    if(!bfs::exists(m_settings->staging_directory())) {
        LOGGER_ERROR("Staging directory {} does not exist",
                     m_settings->staging_directory());
        teardown_and_exit();
    }

    auto s = bfs::status(m_settings->staging_directory());

    auto expected_perms = bfs::perms::owner_read |
                          bfs::perms::owner_write;

    if((s.permissions() & expected_perms) != expected_perms) {
        LOGGER_ERROR("Unable to read from/write to staging directory {}",
                     m_settings->staging_directory());
        teardown_and_exit();
    }
}

void urd::print_greeting() {
    const char greeting[] = "Starting {} daemon (pid {})";
    const auto gsep = std::string(sizeof(greeting) - 4 +
                                  m_settings->progname().size() +
                                  std::to_string(getpid()).size(), '=');

    LOGGER_INFO("{}", gsep);
    LOGGER_INFO(greeting, m_settings->progname(), getpid());
    LOGGER_INFO("{}", gsep);
}

void urd::print_configuration() {
    LOGGER_INFO("");
    LOGGER_INFO("[[ Configuration ]]");
    LOGGER_INFO("  - running as daemon?: {}", (m_settings->daemonize() ? "yes" : "no"));

    if(!m_settings->log_file().empty()) {
        LOGGER_INFO("  - log file: {}", m_settings->log_file());
        LOGGER_INFO("  - log file maximum size: {}", m_settings->log_file_max_size());
    }
    else {
        LOGGER_INFO("  - log file: none");
    }

    LOGGER_INFO("  - dry run?: {} [duration: {} microseconds]",
            (m_settings->dry_run() ? "yes" : "no"), m_settings->dry_run_duration());
    LOGGER_INFO("  - pidfile: {}", m_settings->pidfile());
    LOGGER_INFO("  - control socket: {}", m_settings->control_socket());
    LOGGER_INFO("  - global socket: {}", m_settings->global_socket());
    LOGGER_INFO("  - staging directory: {}", m_settings->staging_directory());
    LOGGER_INFO("  - port for remote requests: {}", m_settings->remote_port());
    LOGGER_INFO("  - workers: {}", m_settings->workers_in_pool());
    LOGGER_INFO("");
}

void urd::print_farewell() {
    const char farewell[] = "Stopping {} daemon (pid {})";
    const auto fsep = std::string(sizeof(farewell) - 4 +
                                  m_settings->progname().size() +
                                  std::to_string(getpid()).size(), '=');

    LOGGER_INFO("{}", fsep);
    LOGGER_INFO(farewell, m_settings->progname(), getpid());
    LOGGER_INFO("{}", fsep);
}

void urd::pause_listening() {
    bool expected = false;
    while(!m_is_paused.compare_exchange_weak(expected, true) && !expected);

    LOGGER_WARN("Daemon locked: incoming requests will be rejected");
}

void urd::resume_listening() {
    bool expected = true;
    while(!m_is_paused.compare_exchange_weak(expected, false) && expected);

    LOGGER_WARN("Daemon unlocked: incoming requests will be processed");
}

urd_error urd::check_shutdown() {
    // - if there are active tasks (i.e. pending or running), we let
    // the client know by returning urd_error::tasks_pending
    const auto task_is_active =
        [](const std::shared_ptr<io::task_info>& ti) {
            // make sure no modifications can happen
            // to the task metadata while we examine it
            const auto lock = ti->lock_shared();
            return (ti->status() == io::task_status::pending ||
                    ti->status() == io::task_status::running);
        };

    if(m_task_mgr->count_if(task_is_active) != 0) {
        return urd_error::tasks_pending;
    }

    // - if there are no active tasks but non-empty tracked backends
    // remain, we return urd_error::namespace_not_empty
    const auto tracked_namespace_not_empty =
        [](const std::shared_ptr<storage::backend>& b) {
            // no need to filter out the process_memory and remote backends,
            // since they will never be tracked
            return (b->is_tracked() && !b->is_empty());
        };

    boost::shared_lock<boost::shared_mutex> lock(m_namespace_mgr_mutex);
    if(m_namespace_mgr->count_if(tracked_namespace_not_empty) != 0) {
        return urd_error::namespace_not_empty;
    }

    // - otherwise, we return urd_error::success
    return urd_error::success;
}

int urd::run() {

    // initialize logging facilities (pre-fork)
    init_logger();

    // validate settings
    check_configuration();

#ifdef __LOGGER_ENABLE_DEBUG__
    if(::prctl(PR_SET_DUMPABLE, 1) != 0) {
        LOGGER_WARN("Failed to set PR_SET_DUMPABLE flag for process. "
                    "Daemon will not produce core dumps.");
    }
#endif

    // daemonize if needed
    if(m_settings->daemonize() && daemonize() != 0) {
        /* parent clean ups and exits, child continues */
        teardown();
        return EXIT_SUCCESS;
    }

    // print useful information
    print_greeting();
    print_configuration();

    LOGGER_INFO("[[ Starting up ]]");

    init_task_manager();
    init_event_handlers();
    init_namespace_manager();
    load_backend_plugins();
    load_default_namespaces();

    // load plugins now so that when we propagate the daemon context to them
    // everything is set up
    load_transfer_plugins();

    // start the listener for remote transfers
    // N.B. This call returns immediately
    m_network_service->run();

    LOGGER_INFO("");
    LOGGER_INFO("[[ Start up successful, awaiting requests... ]]");

    // N.B. This call blocks here, which means that everything after it
    // will only run when a shutdown command is received
    m_ipc_service->run();

    print_farewell();
    teardown();

    LOGGER_INFO("");
    LOGGER_INFO("[Stopped]");

    return EXIT_SUCCESS;
}

void urd::teardown() {

//XXX deprecated, signals_are now managed by api_listener
//    if(m_signal_listener) {
//        LOGGER_INFO("* Stopping signal listener...");
//        m_signal_listener->stop();
//        //m_signal_listener.reset();
//    }

    if(m_ipc_service) {
        LOGGER_INFO("* Stopping API listener...");
        m_ipc_service->stop();
        m_ipc_service.reset();
    }

    api_listener::cleanup();

    if(m_settings) {
        boost::system::error_code ec;

        bfs::remove(m_settings->pidfile(), ec);

        if(ec) {
            LOGGER_ERROR("Failed to remove pidfile {}: {}",
                    m_settings->pidfile(), ec.message());
        }

        m_settings.reset();
    }

    if(m_task_mgr) {
        LOGGER_INFO("* Stopping task manager...");
        m_task_mgr->stop_all_tasks();
        m_task_mgr.reset();
    }
}

void
urd::teardown_and_exit() {
    teardown();
    exit(EXIT_FAILURE);
}

void urd::shutdown() {
    if(m_ipc_service) {
        m_ipc_service->stop();
    }
}

} // namespace norns

ext: cpp

max_stars_repo_name: bsc-ssrg/NORNS

max_issues_repo_name: bsc-ssrg/NORNS

max_forks_repo_name: bsc-ssrg/NORNS


text:
/**
 *          Copyright Matthias Walter 2010.
 * Distributed under the Boost Software License, Version 1.0.
 *    (See accompanying file LICENSE_1_0.txt or copy at
 *          http://www.boost.org/LICENSE_1_0.txt)
 **/

#ifndef MATROID_DECOMPOSITION_HPP_
#define MATROID_DECOMPOSITION_HPP_

#include <set>

#include <boost/graph/adjacency_list.hpp>
#include <boost/numeric/ublas/matrix.hpp>
#include <boost/graph/properties.hpp>

#include "matroid_graph.hpp"

namespace unimod
{

  /**
   * A set of matroid elements. Here, the original row elements are numbered -1, ..., -h
   * and the original column elements +1, ..., +w
   */

  typedef std::set <int> matroid_element_set;

  /**
   * Abstract node in a decomposition tree
   */

  class decomposed_matroid
  {
  public:
    /**
     * Constructs the tree node
     *
     * @param elements Set of matroid elements
     * @param extra_elements Set of extra elements, i.e. those that pivots were made upon
     */

    explicit decomposed_matroid(const matroid_element_set& elements, const matroid_element_set& extra_elements);

    /**
     * Destructor
     */

    virtual ~decomposed_matroid();

    /**
     * @return true if and only if this node is a leaf
     */

    virtual bool is_leaf() const = 0;

    /**
     * @return true if and only if the matroid is graphic
     */

    virtual bool is_regular() const = 0;

    /**
     * @return A read-only reference to the elements
     */

    inline const matroid_element_set& elements() const
    {
      return _elements;
    }

    /**
     * @return A read-only reference to the extra elements
     */

    inline const matroid_element_set& extra_elements() const
    {
      return _extra_elements;
    }

  private:
    matroid_element_set _elements;
    matroid_element_set _extra_elements;
  };

  /**
   * Leaf node in a decomposition tree
   */

  class decomposed_matroid_leaf: public decomposed_matroid
  {
  public:

    /**
     * Constructs the node
     *
     * @param graph A corresponding graph or NULL if not graphic
     * @param cograph A corresponding cograph or NULL if not cographic
     * @param is_R10 Whether this matroid is isomorphic to R10
     * @param elements Set of elements
     * @param extra_elements Set of extra elements
     */

    explicit decomposed_matroid_leaf(matroid_graph* graph, matroid_graph* cograph, bool is_R10, const std::set <int>& elements,
        const matroid_element_set& extra_elements);

    /**
     * Destructor
     */

    virtual ~decomposed_matroid_leaf();

    /**
     * @return true
     */

    virtual bool is_leaf() const
    {
      return true;
    }

    /**
     * @return A corresponding graph or NULL if the matroid is not graphic
     */

    const matroid_graph* graph() const
    {
      return _graph;
    }

    /**
     * @returnA corresponding cograph or NULL if the matroid is not cographic
     */

    const matroid_graph* cograph() const
    {
      return _cograph;
    }

    /**
     * @return true if and only if the matroid is isomorphic to R10
     */

    bool is_R10() const
    {
      return _is_R10;
    }

    /**
     * @return true if and only if the matroid is graphic
     */

    bool is_graphic() const
    {
      return graph() != NULL;
    }

    /**
     * @return true if and only if the matroid is cographic
     */

    bool is_cographic() const
    {
      return cograph() != NULL;
    }

    /**
     * @return true if and only if the matroid is regular
     */

    virtual bool is_regular() const
    {
      return is_graphic() || is_cographic() || is_R10();
    }

  protected:
    matroid_graph* _graph;
    matroid_graph* _cograph;
    bool _is_R10;
  };

  /**
   * Separator node in a decomposition tree
   */

  class decomposed_matroid_separator: public decomposed_matroid
  {
  public:

    /**
     * Constructs the node.
     *
     * @param first First component
     * @param second Second component
     * @param type Type of the separation
     * @param elements Set of elements
     * @param extra_elements Set of extra elements
     * @return
     */

    explicit decomposed_matroid_separator(decomposed_matroid* first, decomposed_matroid* second, int type, const std::set <int>& elements,
        const matroid_element_set& extra_elements);

    /**
     * Destructor
     */

    virtual ~decomposed_matroid_separator();

    enum { ONE_SEPARATION = 1 };
    enum { TWO_SEPARATION = 2 };
    enum { THREE_SEPARATION = 3 };

    /**
     * @return Type of the separation
     */

    inline int separation_type() const
    {
      return _type;
    }

    /**
     * @return First component
     */

    inline decomposed_matroid* first() const
    {
      return _first;
    }

    /**
     * @return Second component
     */

    inline decomposed_matroid* second() const
    {
      return _second;
    }

    /**
     * @return false
     */

    virtual bool is_leaf() const
    {
      return false;
    }

    /**
     * @return true if and only both components are regular
     */

    virtual bool is_regular() const
    {
      return _first->is_regular() && _second->is_regular();
    }

  protected:
    decomposed_matroid* _first;
    decomposed_matroid* _second;
    int _type;
  };

}

#endif /* MATROID_DECOMPOSITION_HPP_ */

max_stars_repo_name: vios-fish/CompetitiveProgramming

max_issues_repo_name: vbraun/unimodularity-library

max_forks_repo_name: vbraun/unimodularity-library


text:
// Copyright 2014 Cloudera, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "kudu/tablet/tablet-test-util.h"

#include <boost/assign/list_of.hpp>
#include <boost/foreach.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <tr1/memory>

#include "kudu/common/partial_row.h"
#include "kudu/common/row_operations.h"
#include "kudu/common/schema.h"
#include "kudu/consensus/consensus_meta.h"
#include "kudu/consensus/log.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/opid_util.h"
#include "kudu/fs/block_id.h"
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/fastmem.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/rpc/messenger.h"
#include "kudu/tserver/remote_bootstrap_session.h"
#include "kudu/tablet/tablet_peer.h"
#include "kudu/util/crc.h"
#include "kudu/util/metrics.h"
#include "kudu/util/test_util.h"
#include "kudu/util/threadpool.h"

METRIC_DECLARE_entity(tablet);

using std::string;
using std::tr1::shared_ptr;

namespace kudu {
namespace tserver {

using consensus::ConsensusMetadata;
using consensus::OpId;
using consensus::RaftConfigPB;
using consensus::RaftPeerPB;
using fs::ReadableBlock;
using log::Log;
using log::LogOptions;
using log::LogAnchorRegistry;
using rpc::Messenger;
using rpc::MessengerBuilder;
using strings::Substitute;
using tablet::ColumnDataPB;
using tablet::DeltaDataPB;
using tablet::KuduTabletTest;
using tablet::RowSetDataPB;
using tablet::TabletPeer;
using tablet::TabletSuperBlockPB;
using tablet::WriteTransactionState;

class RemoteBootstrapTest : public KuduTabletTest {
 public:
  RemoteBootstrapTest()
    : KuduTabletTest(Schema(boost::assign::list_of
                              (ColumnSchema("key", STRING))
                              (ColumnSchema("val", INT32)),
                              1)) {
    CHECK_OK(ThreadPoolBuilder("test-exec").Build(&apply_pool_));
  }

  virtual void SetUp() OVERRIDE {
    KuduTabletTest::SetUp();
    SetUpTabletPeer();
    ASSERT_NO_FATAL_FAILURE(PopulateTablet());
    InitSession();
  }

  virtual void TearDown() OVERRIDE {
    session_.reset();
    tablet_peer_->Shutdown();
    KuduTabletTest::TearDown();
  }

 protected:
  void SetUpTabletPeer() {
    scoped_refptr<Log> log;
    CHECK_OK(Log::Open(LogOptions(), fs_manager(), tablet()->tablet_id(),
                       *tablet()->schema(),
                       0, // schema_version
                       NULL, &log));

    scoped_refptr<MetricEntity> metric_entity =
      METRIC_ENTITY_tablet.Instantiate(&metric_registry_, CURRENT_TEST_NAME());

    RaftPeerPB config_peer;
    config_peer.set_permanent_uuid(fs_manager()->uuid());
    config_peer.set_member_type(RaftPeerPB::VOTER);

    tablet_peer_.reset(
        new TabletPeer(tablet()->metadata(),
                       config_peer,
                       apply_pool_.get(),
                       Bind(&RemoteBootstrapTest::TabletPeerStateChangedCallback,
                            Unretained(this),
                            tablet()->tablet_id())));

    // TODO similar to code in tablet_peer-test, consider refactor.
    RaftConfigPB config;
    config.set_local(true);
    config.add_peers()->CopyFrom(config_peer);
    config.set_opid_index(consensus::kInvalidOpIdIndex);

    gscoped_ptr<ConsensusMetadata> cmeta;
    CHECK_OK(ConsensusMetadata::Create(tablet()->metadata()->fs_manager(),
                                       tablet()->tablet_id(), fs_manager()->uuid(),
                                       config, consensus::kMinimumTerm, &cmeta));

    shared_ptr<Messenger> messenger;
    MessengerBuilder mbuilder(CURRENT_TEST_NAME());
    mbuilder.Build(&messenger);

    log_anchor_registry_.reset(new LogAnchorRegistry());
    tablet_peer_->SetBootstrapping();
    CHECK_OK(tablet_peer_->Init(tablet(),
                                clock(),
                                messenger,
                                log,
                                metric_entity));
    consensus::ConsensusBootstrapInfo boot_info;
    CHECK_OK(tablet_peer_->Start(boot_info));

    ASSERT_OK(tablet_peer_->WaitUntilConsensusRunning(MonoDelta::FromSeconds(2)));
  }

  void TabletPeerStateChangedCallback(const string& tablet_id, const string& reason) {
    LOG(INFO) << "Tablet peer state changed for tablet " << tablet_id << ". Reason: " << reason;
  }

  void PopulateTablet() {
    for (int32_t i = 0; i < 1000; i++) {
      WriteRequestPB req;
      req.set_tablet_id(tablet_peer_->tablet_id());
      ASSERT_OK(SchemaToPB(client_schema_, req.mutable_schema()));
      RowOperationsPB* data = req.mutable_row_operations();
      RowOperationsPBEncoder enc(data);
      KuduPartialRow row(&client_schema_);

      string key = Substitute("key$0", i);
      ASSERT_OK(row.SetString(0, key));
      ASSERT_OK(row.SetInt32(1, i));
      enc.Add(RowOperationsPB::INSERT, row);

      WriteResponsePB resp;
      CountDownLatch latch(1);

      WriteTransactionState* state =
          new WriteTransactionState(tablet_peer_.get(), &req, &resp);
      state->set_completion_callback(gscoped_ptr<tablet::TransactionCompletionCallback>(
          new tablet::LatchTransactionCompletionCallback<WriteResponsePB>(&latch, &resp)).Pass());
      ASSERT_OK(tablet_peer_->SubmitWrite(state));
      latch.Wait();
      ASSERT_FALSE(resp.has_error()) << "Request failed: " << resp.error().ShortDebugString();
      ASSERT_EQ(0, resp.per_row_errors_size()) << "Insert error: " << resp.ShortDebugString();
    }
    ASSERT_OK(tablet()->Flush());
  }

  void InitSession() {
    session_.reset(new RemoteBootstrapSession(tablet_peer_.get(), "TestSession", "FakeUUID",
                   fs_manager()));
    CHECK_OK(session_->Init());
  }

  // Read the specified BlockId, via the RemoteBootstrapSession, into a file.
  // 'path' will be populated with the name of the file used.
  // 'file' will be set to point to the SequentialFile containing the data.
  void FetchBlockToFile(const BlockId& block_id,
                        string* path,
                        gscoped_ptr<SequentialFile>* file) {
    string data;
    int64_t block_file_size = 0;
    RemoteBootstrapErrorPB::Code error_code;
    CHECK_OK(session_->GetBlockPiece(block_id, 0, 0, &data, &block_file_size, &error_code));
    if (block_file_size > 0) {
      CHECK_GT(data.size(), 0);
    }

    // Write the file to a temporary location.
    WritableFileOptions opts;
    string path_template = GetTestPath(Substitute("test_block_$0.tmp.XXXXXX", block_id.ToString()));
    gscoped_ptr<WritableFile> writable_file;
    CHECK_OK(Env::Default()->NewTempWritableFile(opts, path_template, path, &writable_file));
    CHECK_OK(writable_file->Append(Slice(data.data(), data.size())));
    CHECK_OK(writable_file->Close());

    CHECK_OK(Env::Default()->NewSequentialFile(*path, file));
  }

  MetricRegistry metric_registry_;
  scoped_refptr<LogAnchorRegistry> log_anchor_registry_;
  gscoped_ptr<ThreadPool> apply_pool_;
  scoped_refptr<TabletPeer> tablet_peer_;
  scoped_refptr<RemoteBootstrapSession> session_;
};

// Ensure that the serialized SuperBlock included in the RemoteBootstrapSession is
// equal to the serialized live superblock (on a quiesced tablet).
TEST_F(RemoteBootstrapTest, TestSuperBlocksEqual) {
  // Compare content of superblocks.
  faststring session_buf;
  faststring tablet_buf;

  {
    const TabletSuperBlockPB& session_superblock = session_->tablet_superblock();
    int size = session_superblock.ByteSize();
    session_buf.resize(size);
    uint8_t* session_dst = session_buf.data();
    session_dst = session_superblock.SerializeWithCachedSizesToArray(session_dst);
  }

  {
    TabletSuperBlockPB tablet_superblock;
    ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
    int size = tablet_superblock.ByteSize();
    tablet_buf.resize(size);
    uint8_t* tablet_dst = tablet_buf.data();
    tablet_dst = tablet_superblock.SerializeWithCachedSizesToArray(tablet_dst);
  }

  ASSERT_EQ(session_buf.size(), tablet_buf.size());
  int size = tablet_buf.size();
  ASSERT_EQ(0, strings::fastmemcmp_inlined(session_buf.data(), tablet_buf.data(), size));
}

// Test fetching all files from tablet server, ensure the checksums for each
// chunk and the total file sizes match.
TEST_F(RemoteBootstrapTest, TestBlocksEqual) {
  TabletSuperBlockPB tablet_superblock;
  ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
  for (int i = 0; i < tablet_superblock.rowsets_size(); i++) {
    const RowSetDataPB& rowset = tablet_superblock.rowsets(i);
    for (int j = 0; j < rowset.columns_size(); j++) {
      const ColumnDataPB& column = rowset.columns(j);
      const BlockIdPB& block_id_pb = column.block();
      BlockId block_id = BlockId::FromPB(block_id_pb);

      string path;
      gscoped_ptr<SequentialFile> file;
      FetchBlockToFile(block_id, &path, &file);
      uint64_t session_block_size = 0;
      ASSERT_OK(Env::Default()->GetFileSize(path, &session_block_size));
      faststring buf;
      buf.resize(session_block_size);
      Slice data;
      ASSERT_OK(file->Read(session_block_size, &data, buf.data()));
      uint32_t session_crc = crc::Crc32c(data.data(), data.size());
      LOG(INFO) << "session block file has size of " << session_block_size
                << " and CRC32C of " << session_crc << ": " << path;

      gscoped_ptr<ReadableBlock> tablet_block;
      ASSERT_OK(fs_manager()->OpenBlock(block_id, &tablet_block));
      uint64_t tablet_block_size = 0;
      ASSERT_OK(tablet_block->Size(&tablet_block_size));
      buf.resize(tablet_block_size);
      ASSERT_OK(tablet_block->Read(0, tablet_block_size, &data, buf.data()));
      uint32_t tablet_crc = crc::Crc32c(data.data(), data.size());
      LOG(INFO) << "tablet block file has size of " << tablet_block_size
                << " and CRC32C of " << tablet_crc
                << ": " << block_id;

      // Compare the blocks.
      ASSERT_EQ(tablet_block_size, session_block_size);
      ASSERT_EQ(tablet_crc, session_crc);
    }
  }
}

// Ensure that blocks are still readable through the open session even
// after they've been deleted.
TEST_F(RemoteBootstrapTest, TestBlocksAreFetchableAfterBeingDeleted) {
  TabletSuperBlockPB tablet_superblock;
  ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));

  // Gather all the blocks.
  vector<BlockId> data_blocks;
  BOOST_FOREACH(const RowSetDataPB& rowset, tablet_superblock.rowsets()) {
    BOOST_FOREACH(const DeltaDataPB& redo, rowset.redo_deltas()) {
      data_blocks.push_back(BlockId::FromPB(redo.block()));
    }
    BOOST_FOREACH(const DeltaDataPB& undo, rowset.undo_deltas()) {
      data_blocks.push_back(BlockId::FromPB(undo.block()));
    }
    BOOST_FOREACH(const ColumnDataPB& column, rowset.columns()) {
      data_blocks.push_back(BlockId::FromPB(column.block()));
    }
    if (rowset.has_bloom_block()) {
      data_blocks.push_back(BlockId::FromPB(rowset.bloom_block()));
    }
    if (rowset.has_adhoc_index_block()) {
      data_blocks.push_back(BlockId::FromPB(rowset.adhoc_index_block()));
    }
  }

  // Delete them.
  BOOST_FOREACH(const BlockId& block_id, data_blocks) {
    ASSERT_OK(fs_manager()->DeleteBlock(block_id));
  }

  // Read them back.
  BOOST_FOREACH(const BlockId& block_id, data_blocks) {
    ASSERT_TRUE(session_->IsBlockOpenForTests(block_id));
    string data;
    RemoteBootstrapErrorPB::Code error_code;
    int64_t piece_size;
    ASSERT_OK(session_->GetBlockPiece(block_id, 0, 0,
                                      &data, &piece_size, &error_code));
  }
}

}  // namespace tserver
}  // namespace kudu

ext: cc

max_stars_repo_name: kv-zuiwanyuan/kudu

max_issues_repo_name: kv-zuiwanyuan/kudu

max_forks_repo_name: kv-zuiwanyuan/kudu