3. Implement the solve member function

4. Create a driver

#include "utils.hpp"
#include "wc.hpp"
DEFINE_string(server_info);
DEFINE_string(cfg_file);
int main(int argc, char *argv[])
{
  // init worker environment
  paracel::main_env comm_main_env(argc, argv);
  paracel::Comm comm(MPI_COMM_WORLD);
  // init parameter server environment
  google::SetUsageMessage("[options]\n\t--server_info\n");
  google::ParseCommandLineFlags(&argc, &argv, true);
  
  string input = "/alg/data/*.csv"; string output = "result.txt";
  int a = 7;
  // create a instance
  HOF solver(comm, FLAGS_server_info, input, output, a);
  // solve problems
  solver.solve();
  return 0;
}
  

specify LD_LIBRARY_PATH, then use prun.py

Options:
  -h, --help                        show this help message and exit
  -p PARASRV_NUM, --snum=PARASRV_NUM
                                    number of parameter servers
  --m_server=local | mesos | mpi
                                    running method for parameter servers. If not given,
                                    set with the same value of -m or --method
  --ppn_server=PPN_SERVER           mesos case: procs number per node of parameter
                                    servers. If not given, set with the same value of --ppn
  --mem_limit_server=MEM_LIMIT_SERVER 
                                    mesos case: memory size of each task in parameter
                                    servers. If not given, set with the same value of --mem_limit
  --hostfile_server=HOSTFILE_SERVER mpi case: hostfile for mpirun of parameter servers
  -w WORKER_NUM, --wnum=WORKER_NUM  number of workers for learning
  -m local | mesos | mpi, --method=local | mesos | mpi
                                    running method for workers
  --ppn=PPN                         mesos case: procs number per node for workers
  --mem_limit=MEM_LIMIT             mesos case: memory size of each task of workers
  --hostfile=HOSTFILE               mpi case: hostfile for mpirun for workers
  -c CONFIG, --cfg_file=CONFIG      config file in json fmt, for alg usage
  

class WordCount : public paralg {
 private:
  vector<pair<string, int> > result;
};

void WordCount::learning(const vector<string> & lines) {
  for(auto & line : lines) {
    auto word_lst = parser(line);
    for(auto & word : word_lst) {
      paracel_bupdate(word, 1,
                      "/local/lib/libwc_update.so",
                      "wc_updater");
    } // word_lst                                     
  } // lines                                        // update function
  sync();                                           int local_update(int a, int b) {
  paracel_read_topk(topk, result)                     return a + b;
}                                                   }

virtual void WordCount::solve() {                   paracel::update_result wc_updater = 
  auto lines = paracel_load(input);                   paracel::update_proxy(local_update);
  sync();
  learning(lines); 
  sync();
}
  

void logistic_regression::solve() {
  auto lines = paracel_load(input); local_parser(lines); sync(); // init data
  set_total_iters(rounds); 
  dgd_learning();
}

void logistic_regression::dgd_learning() {
  theta = paracel::random_double_list(data_dim); 
  paracel_write("theta", theta); // init push
  for(int rd = 0; rd < rounds; ++rd) {
    for(int i = 0; i < data_dim; ++i) { delta[i] = 0.; }
    random_shuffle(idx.begin(), idx.end()); 
    theta = paracel_read<vector<double> >("theta"); // pull thera
    for(auto sample_id : idx) {
      for(int i = 0; i < data_dim; ++i) {
        delta[i] += coff1 * samples[sample_id][i] - coff2 * theta[i];
      }
    } // traverse
    paracel_bupdate("theta", delta, "update.so", "lg_theta_update"); // update with delta
    iter_commit();
  }
  theta = paracel_read<vector<double> >("theta"); // last pull
}