max/poet
1
0
Fork 0
mirror of https://git.gfz-potsdam.de/naaice/poet.git synced 2025-12-16 12:54:50 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into doc

Browse Source
This commit is contained in:
Max Lübke 2021-02-02 14:00:13 +01:00
commit a801dc7512
No known key found for this signature in database
GPG Key ID: D3201E51647D1199
15 changed files with 500 additions and 432 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -1,4 +1,4 @@

<!-- <!--
Time-stamp: "Last modified 2020-02-01 18:14:13 delucia" Time-stamp: "Last modified 2020-02-01 18:14:13 delucia"
--> -->
@ -23,7 +23,7 @@ DHT:
--dht ... enable dht (default is off) --dht ... enable dht (default is off)
--dht-log ... enable logarithm application before rounding (default is off) --dht-nolog ... enable logarithm application before rounding (default is off)
--dht-signif=<1-n> ... set rounding to number of significant digits (default 5) --dht-signif=<1-n> ... set rounding to number of significant digits (default 5)
(only used if no vector is given in setup file) (only used if no vector is given in setup file)

8
src/DHT/DHT.c
View File

@ -1,10 +1,10 @@
#include "DHT.h" #include "DHT.h"
#include <inttypes.h>
#include <math.h> #include <math.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <inttypes.h>
#include <unistd.h> #include <unistd.h>
static void determine_dest(uint64_t hash, int comm_size, static void determine_dest(uint64_t hash, int comm_size,
@ -79,7 +79,7 @@ DHT *DHT_create(MPI_Comm comm, uint64_t size, unsigned int data_size,
object->recv_entry = malloc(1 + data_size + key_size); object->recv_entry = malloc(1 + data_size + key_size);
object->send_entry = malloc(1 + data_size + key_size); object->send_entry = malloc(1 + data_size + key_size);
object->index_count = 9 - (index_bytes / 8); object->index_count = 9 - (index_bytes / 8);
object->index = (uint64_t*)malloc((object->index_count) * sizeof(uint64_t)); object->index = (uint64_t *)malloc((object->index_count) * sizeof(uint64_t));
object->mem_alloc = mem_alloc; object->mem_alloc = mem_alloc;
// if set, initialize dht_stats // if set, initialize dht_stats
@ -375,8 +375,8 @@ int DHT_free(DHT *table, int *eviction_counter, int *readerror_counter) {
return DHT_SUCCESS; return DHT_SUCCESS;
} }
#ifdef DHT_STATISTICS
int DHT_print_statistics(DHT *table) { int DHT_print_statistics(DHT *table) {
#ifdef DHT_STATISTICS
int *written_buckets; int *written_buckets;
int *read_misses, sum_read_misses; int *read_misses, sum_read_misses;
int *evictions, sum_evictions; int *evictions, sum_evictions;
@ -476,5 +476,5 @@ int DHT_print_statistics(DHT *table) {
MPI_Barrier(table->communicator); MPI_Barrier(table->communicator);
return DHT_SUCCESS; return DHT_SUCCESS;
#endif
} }
#endif

2
src/DHT/DHT.h
View File

@ -250,9 +250,7 @@ extern int DHT_free(DHT* table, int* eviction_counter, int* readerror_counter);
* @return int Returns DHT_SUCCESS on success or DHT_MPI_ERROR on internal MPI * @return int Returns DHT_SUCCESS on success or DHT_MPI_ERROR on internal MPI
* error. * error.
*/ */
#ifdef DHT_STATISTICS
extern int DHT_print_statistics(DHT* table); extern int DHT_print_statistics(DHT* table);
#endif
/** /**
* @brief Determine destination rank and index. * @brief Determine destination rank and index.

20
src/DHT/DHT_Wrapper.cpp
View File

@ -26,20 +26,22 @@ uint64_t poet::get_md5(int key_size, void *key) {
return retval; return retval;
} }
DHT_Wrapper::DHT_Wrapper(t_simparams *params, MPI_Comm dht_comm, DHT_Wrapper::DHT_Wrapper(SimParams &params, MPI_Comm dht_comm,
int buckets_per_process, int data_size, int key_size) { int buckets_per_process, int data_size, int key_size) {
// initialize DHT object // initialize DHT object
dht_object = dht_object =
DHT_create(dht_comm, buckets_per_process, data_size, key_size, &poet::get_md5); DHT_create(dht_comm, buckets_per_process, data_size, key_size, &get_md5);
// allocate memory for fuzzing buffer // allocate memory for fuzzing buffer
fuzzing_buffer = (double *)malloc(key_size); fuzzing_buffer = (double *)malloc(key_size);
// extract needed values from sim_param struct // extract needed values from sim_param struct
this->dt_differ = params->dt_differ; t_simparams tmp = params.getNumParams();
this->dht_log = params->dht_log;
this->dht_signif_vector = params->dht_signif_vector; this->dt_differ = tmp.dt_differ;
this->dht_prop_type_vector = params->dht_prop_type_vector; this->dht_log = tmp.dht_log;
this->dht_signif_vector = params.getDHTSignifVector();
this->dht_prop_type_vector = params.getDHTPropTypeVector();
} }
DHT_Wrapper::~DHT_Wrapper() { DHT_Wrapper::~DHT_Wrapper() {
@ -139,9 +141,7 @@ int DHT_Wrapper::fileToTable(const char *filename) {
void DHT_Wrapper::printStatistics() { void DHT_Wrapper::printStatistics() {
int res; int res;
#ifdef DHT_STATISTICS
res = DHT_print_statistics(dht_object); res = DHT_print_statistics(dht_object);
#endif
if (res != DHT_SUCCESS) { if (res != DHT_SUCCESS) {
// MPI ERROR ... WHAT TO DO NOW? // MPI ERROR ... WHAT TO DO NOW?
@ -209,4 +209,4 @@ void DHT_Wrapper::fuzzForDHT(int var_count, void *key, double dt) {
// if timestep differs over iterations set current current time step at the // if timestep differs over iterations set current current time step at the
// end of fuzzing buffer // end of fuzzing buffer
if (dt_differ) fuzzing_buffer[var_count] = dt; if (dt_differ) fuzzing_buffer[var_count] = dt;
} }

7
src/DHT/DHT_Wrapper.h
View File

@ -51,14 +51,14 @@ class DHT_Wrapper {
* calling DHT_create with all given parameters. Also the fuzzing buffer will * calling DHT_create with all given parameters. Also the fuzzing buffer will
* be allocated and all needed parameters extracted from simparams struct. * be allocated and all needed parameters extracted from simparams struct.
* *
* @param params Simulation parameter struct returned at initialization * @param params Simulation parameter object
* @param dht_comm Communicator which addresses all participating DHT * @param dht_comm Communicator which addresses all participating DHT
* processes * processes
* @param buckets_per_process Count of buckets to allocate for each process * @param buckets_per_process Count of buckets to allocate for each process
* @param data_size Size of data in bytes * @param data_size Size of data in bytes
* @param key_size Size of key in bytes * @param key_size Size of key in bytes
*/ */
DHT_Wrapper(t_simparams *params, MPI_Comm dht_comm, int buckets_per_process, DHT_Wrapper(SimParams &params, MPI_Comm dht_comm, int buckets_per_process,
int data_size, int key_size); int data_size, int key_size);
/** /**
* @brief Destroy the dht wrapper object * @brief Destroy the dht wrapper object
@ -82,7 +82,8 @@ class DHT_Wrapper {
* marked with false ('not to be simulated'). * marked with false ('not to be simulated').
* *
* @param length Count of grid cells inside work package * @param length Count of grid cells inside work package
* @param[out] out_result_index Indexing work packages which should be simulated * @param[out] out_result_index Indexing work packages which should be
* simulated
* @param[in,out] work_package Pointer to current work package * @param[in,out] work_package Pointer to current work package
* @param dt Current timestep of simulation * @param dt Current timestep of simulation
*/ */

158
src/kin.cpp
View File

@ -10,7 +10,6 @@
// BSD-licenced #include "dht_wrapper.h" #include "global_buffer.h" // BSD-licenced #include "dht_wrapper.h" #include "global_buffer.h"
#include <ChemSim.h> #include <ChemSim.h>
#include <Grid.h> #include <Grid.h>
#include <Parser.h>
#include <RRuntime.h> #include <RRuntime.h>
#include <SimParams.h> #include <SimParams.h>
#include <TransportSim.h> #include <TransportSim.h>
@ -85,11 +84,10 @@ int main(int argc, char *argv[]) {
double cummul_chemistry = 0.f; double cummul_chemistry = 0.f;
double cummul_master_seq = 0.f; double cummul_master_seq = 0.f;
argh::parser cmdl(argv); // argh::parser cmdl(argv);
int dht_significant_digits; // int dht_significant_digits;
// cout << "CPP: Start Init (MPI)" << endl; // cout << "CPP: Start Init (MPI)" << endl;
t_simparams params;
int world_size, world_rank; int world_size, world_rank;
MPI_Init(&argc, &argv); MPI_Init(&argc, &argv);
@ -190,8 +188,8 @@ int main(int argc, char *argv[]) {
"source('parallel_r_library.R');"; "source('parallel_r_library.R');";
R.parseEvalQ(r_load_dependencies); R.parseEvalQ(r_load_dependencies);
Parser parser(argv, world_rank, world_size); SimParams parser(world_rank, world_size);
int pret = parser.parseCmdl(); int pret = parser.parseFromCmdl(argv, R);
if (pret == PARSER_ERROR) { if (pret == PARSER_ERROR) {
MPI_Finalize(); MPI_Finalize();
@ -201,8 +199,8 @@ int main(int argc, char *argv[]) {
return EXIT_SUCCESS; return EXIT_SUCCESS;
} }
parser.parseR(R); // parser.parseR(R);
params = parser.getParams(); // params = parser.getParams();
// if (params.world_rank == 0) { // if (params.world_rank == 0) {
// cout << "CPP: Complete results storage is " << (params.store_result ? // cout << "CPP: Complete results storage is " << (params.store_result ?
@ -233,6 +231,7 @@ int main(int argc, char *argv[]) {
// "source(filesim)"); // eval the init string, ignoring any returns // "source(filesim)"); // eval the init string, ignoring any returns
// only rank 0 initializes goes through the whole initialization // only rank 0 initializes goes through the whole initialization
bool dt_differ;
if (world_rank == 0) { if (world_rank == 0) {
// cmdl(2) >> params.out_dir; // <- second positional argument // cmdl(2) >> params.out_dir; // <- second positional argument
// R["fileout"] = // R["fileout"] =
@ -248,13 +247,17 @@ int main(int argc, char *argv[]) {
std::string master_init_code = "mysetup <- master_init(setup=setup)"; std::string master_init_code = "mysetup <- master_init(setup=setup)";
R.parseEval(master_init_code); R.parseEval(master_init_code);
params.dt_differ = R.parseEval("mysetup$dt_differ"); dt_differ = R.parseEval("mysetup$dt_differ");
MPI_Bcast(&(params.dt_differ), 1, MPI_C_BOOL, 0, MPI_COMM_WORLD); MPI_Bcast(&dt_differ, 1, MPI_C_BOOL, 0, MPI_COMM_WORLD);
} else { // workers will only read the setup DataFrame defined by input file } else { // workers will only read the setup DataFrame defined by input file
R.parseEval("mysetup <- setup"); R.parseEval("mysetup <- setup");
MPI_Bcast(&(params.dt_differ), 1, MPI_C_BOOL, 0, MPI_COMM_WORLD); MPI_Bcast(&dt_differ, 1, MPI_C_BOOL, 0, MPI_COMM_WORLD);
} }
parser.setDtDiffer(dt_differ);
t_simparams params = parser.getNumParams();
if (world_rank == 0) { if (world_rank == 0) {
cout << "CPP: R init done on process with rank " << params.world_rank cout << "CPP: R init done on process with rank " << params.world_rank
<< endl; << endl;
@ -267,6 +270,8 @@ int main(int argc, char *argv[]) {
Grid grid(R); Grid grid(R);
// params.grid = &grid; // params.grid = &grid;
grid.init(); grid.init();
parser.initVectorParams(R, grid.getCols());
/* Retrieve state_C from R context for MPI buffer generation */ /* Retrieve state_C from R context for MPI buffer generation */
// Rcpp::DataFrame state_C = R.parseEval("mysetup$state_C"); // Rcpp::DataFrame state_C = R.parseEval("mysetup$state_C");
@ -296,78 +301,79 @@ int main(int argc, char *argv[]) {
// params.R = &R; // params.R = &R;
if (params.dht_enabled) { // if (params.dht_enabled) {
// cout << "\nCreating DHT\n"; // // cout << "\nCreating DHT\n";
// determine size of dht entries // // determine size of dht entries
// int dht_data_size = grid.getCols() * sizeof(double); // // int dht_data_size = grid.getCols() * sizeof(double);
// int dht_key_size = // // int dht_key_size =
// grid.getCols() * sizeof(double) + (params.dt_differ * // // grid.getCols() * sizeof(double) + (params.dt_differ *
// sizeof(double)); // // sizeof(double));
// // // determine bucket count for preset memory usage // // // // determine bucket count for preset memory usage
// // // bucket size is key + value + 1 byte for status // // // // bucket size is key + value + 1 byte for status
// int dht_buckets_per_process = // // int dht_buckets_per_process =
// params.dht_size_per_process / (1 + dht_data_size + dht_key_size); // // params.dht_size_per_process / (1 + dht_data_size + dht_key_size);
// MDL : following code moved here from worker.cpp // // MDL : following code moved here from worker.cpp
/*Load significance vector from R setup file (or set default)*/ // /*Load significance vector from R setup file (or set default)*/
bool signif_vector_exists = R.parseEval("exists('signif_vector')"); // bool signif_vector_exists = R.parseEval("exists('signif_vector')");
if (signif_vector_exists) { // if (signif_vector_exists) {
params.dht_signif_vector = as<std::vector<int>>(R["signif_vector"]); // params.dht_signif_vector = as<std::vector<int>>(R["signif_vector"]);
} else { // } else {
params.dht_signif_vector.assign(grid.getCols(), dht_significant_digits); // params.dht_signif_vector.assign(grid.getCols(),
} // params.dht_significant_digits);
// }
/*Load property type vector from R setup file (or set default)*/ // /*Load property type vector from R setup file (or set default)*/
bool prop_type_vector_exists = R.parseEval("exists('prop_type')"); // bool prop_type_vector_exists = R.parseEval("exists('prop_type')");
if (prop_type_vector_exists) { // if (prop_type_vector_exists) {
params.dht_prop_type_vector = as<std::vector<string>>(R["prop_type"]); // params.dht_prop_type_vector = as<std::vector<string>>(R["prop_type"]);
} else { // } else {
params.dht_prop_type_vector.assign(grid.getCols(), "act"); // params.dht_prop_type_vector.assign(grid.getCols(), "act");
} // }
if (params.world_rank == 0) { // if (params.world_rank == 0) {
// // print only on master, values are equal on all workes // // // print only on master, values are equal on all workes
// cout << "CPP: dht_data_size: " << dht_data_size << "\n"; // // cout << "CPP: dht_data_size: " << dht_data_size << "\n";
// cout << "CPP: dht_key_size: " << dht_key_size << "\n"; // // cout << "CPP: dht_key_size: " << dht_key_size << "\n";
// cout << "CPP: dht_buckets_per_process: " << dht_buckets_per_process // // cout << "CPP: dht_buckets_per_process: " << dht_buckets_per_process
// << endl; // // << endl;
// MDL: new output on signif_vector and prop_type // // MDL: new output on signif_vector and prop_type
if (signif_vector_exists) { // if (signif_vector_exists) {
cout << "CPP: using problem-specific rounding digits: " << endl; // cout << "CPP: using problem-specific rounding digits: " << endl;
R.parseEval( // R.parseEval(
"print(data.frame(prop=prop, type=prop_type, " // "print(data.frame(prop=prop, type=prop_type, "
"digits=signif_vector))"); // "digits=signif_vector))");
} else { // } else {
cout << "CPP: using DHT default rounding digits = " // cout << "CPP: using DHT default rounding digits = "
<< dht_significant_digits << endl; // << params.dht_significant_digits << endl;
} // }
// MDL: pass to R the DHT stuff. These variables exist // // MDL: pass to R the DHT stuff. These variables exist
// only if dht_enabled is true // // only if dht_enabled is true
R["dht_final_signif"] = params.dht_signif_vector; // R["dht_final_signif"] = params.dht_signif_vector;
R["dht_final_proptype"] = params.dht_prop_type_vector; // R["dht_final_proptype"] = params.dht_prop_type_vector;
} // }
// if (params.dht_strategy == 0) { // // if (params.dht_strategy == 0) {
// if (params.world_rank != 0) { // // if (params.world_rank != 0) {
// dht_object = DHT_create(dht_comm, dht_buckets_per_process, // // dht_object = DHT_create(dht_comm, dht_buckets_per_process,
// dht_data_size, dht_key_size, get_md5); // // dht_data_size, dht_key_size, get_md5);
// // storing for access from worker and callback functions // // // storing for access from worker and callback functions
// fuzzing_buffer = (double *)malloc(dht_key_size); // // fuzzing_buffer = (double *)malloc(dht_key_size);
// } // // }
// } else { // // } else {
// dht_object = DHT_create(MPI_COMM_WORLD, dht_buckets_per_process, // // dht_object = DHT_create(MPI_COMM_WORLD, dht_buckets_per_process,
// dht_data_size, dht_key_size, get_md5); // // dht_data_size, dht_key_size, get_md5);
// } // // }
// if (params.world_rank == 0) { // // if (params.world_rank == 0) {
// cout << "CPP: DHT successfully created!" << endl; // // cout << "CPP: DHT successfully created!" << endl;
// } // // }
}
// }
// MDL: store all parameters // MDL: store all parameters
if (params.world_rank == 0) { if (params.world_rank == 0) {
cout << "CPP: Calling R Function to store calling parameters" << endl; cout << "CPP: Calling R Function to store calling parameters" << endl;
@ -377,7 +383,7 @@ int main(int argc, char *argv[]) {
MPI_Barrier(MPI_COMM_WORLD); MPI_Barrier(MPI_COMM_WORLD);
if (params.world_rank == 0) { /* This is executed by the master */ if (params.world_rank == 0) { /* This is executed by the master */
ChemMaster master(&params, R, grid); ChemMaster master(parser, R, grid);
TransportSim trans(R); TransportSim trans(R);
Rcpp::NumericVector master_send; Rcpp::NumericVector master_send;
@ -572,14 +578,14 @@ int main(int argc, char *argv[]) {
R.parseEval(r_vis_code); R.parseEval(r_vis_code);
cout << "CPP: Done! Results are stored as R objects into <" cout << "CPP: Done! Results are stored as R objects into <"
<< params.out_dir << "/timings.rds>" << endl; << parser.getOutDir() << "/timings.rds>" << endl;
/*exporting results and profiling data*/ /*exporting results and profiling data*/
// std::string r_vis_code; // std::string r_vis_code;
// r_vis_code = "saveRDS(profiling, file=paste0(fileout,'/timings.rds'));"; // r_vis_code = "saveRDS(profiling, file=paste0(fileout,'/timings.rds'));";
// R.parseEval(r_vis_code); // R.parseEval(r_vis_code);
} else { /*This is executed by the workers*/ } else { /*This is executed by the workers*/
ChemWorker worker(&params, R, grid, dht_comm); ChemWorker worker(parser, R, grid, dht_comm);
// worker.prepareSimulation(dht_comm); // worker.prepareSimulation(dht_comm);
worker.loop(); worker.loop();
} }

13
src/model/ChemMaster.cpp
View File

@ -8,13 +8,14 @@ using namespace poet;
using namespace std; using namespace std;
using namespace Rcpp; using namespace Rcpp;
#define TAG_WORK 42 ChemMaster::ChemMaster(SimParams &params, RRuntime &R_, Grid &grid_)
ChemMaster::ChemMaster(t_simparams *params, RRuntime &R_, Grid &grid_)
: ChemSim(params, R_, grid_) { : ChemSim(params, R_, grid_) {
this->wp_size = params->wp_size; t_simparams tmp = params.getNumParams();
this->out_dir = params->out_dir;
this->dht_enabled = params->dht_enabled; this->wp_size = tmp.wp_size;
this->dht_enabled = tmp.dht_enabled;
this->out_dir = params.getOutDir();
workerlist = (worker_struct *)calloc(world_size - 1, sizeof(worker_struct)); workerlist = (worker_struct *)calloc(world_size - 1, sizeof(worker_struct));
send_buffer = (double *)calloc((wp_size * (grid.getCols())) + BUFFER_OFFSET, send_buffer = (double *)calloc((wp_size * (grid.getCols())) + BUFFER_OFFSET,

11
src/model/ChemSim.cpp
View File

@ -7,12 +7,13 @@
using namespace Rcpp; using namespace Rcpp;
using namespace poet; using namespace poet;
ChemSim::ChemSim(t_simparams *params, RRuntime &R_, Grid &grid_) ChemSim::ChemSim(SimParams &params, RRuntime &R_, Grid &grid_)
: R(R_), grid(grid_) { : R(R_), grid(grid_) {
this->world_rank = params->world_rank; t_simparams tmp = params.getNumParams();
this->world_size = params->world_size; this->world_rank = tmp.world_rank;
this->wp_size = params->wp_size; this->world_size = tmp.world_size;
this->out_dir = params->out_dir; this->wp_size = tmp.wp_size;
this->out_dir = params.getOutDir();
} }
void ChemSim::run() { void ChemSim::run() {

8
src/model/ChemSim.h
View File

@ -23,7 +23,7 @@
namespace poet { namespace poet {
class ChemSim { class ChemSim {
public: public:
ChemSim(t_simparams *params, RRuntime &R_, Grid &grid_); ChemSim(SimParams &params, RRuntime &R_, Grid &grid_);
virtual void run(); virtual void run();
virtual void end(); virtual void end();
@ -59,7 +59,7 @@ class ChemSim {
class ChemMaster : public ChemSim { class ChemMaster : public ChemSim {
public: public:
ChemMaster(t_simparams *params, RRuntime &R_, Grid &grid_); ChemMaster(SimParams &params, RRuntime &R_, Grid &grid_);
~ChemMaster(); ~ChemMaster();
void run() override; void run() override;
@ -91,8 +91,7 @@ class ChemMaster : public ChemSim {
class ChemWorker : public ChemSim { class ChemWorker : public ChemSim {
public: public:
ChemWorker(t_simparams *params_, RRuntime &R_, Grid &grid_, ChemWorker(SimParams &params, RRuntime &R_, Grid &grid_, MPI_Comm dht_comm);
MPI_Comm dht_comm);
~ChemWorker(); ~ChemWorker();
void loop(); void loop();
@ -110,7 +109,6 @@ class ChemWorker : public ChemSim {
std::string dht_file; std::string dht_file;
unsigned int dht_size_per_process; unsigned int dht_size_per_process;
std::vector<bool> dht_flags; std::vector<bool> dht_flags;
t_simparams *params;
double *mpi_buffer_results; double *mpi_buffer_results;

25
src/model/ChemWorker.cpp
View File

@ -9,14 +9,17 @@ using namespace poet;
using namespace std; using namespace std;
using namespace Rcpp; using namespace Rcpp;
ChemWorker::ChemWorker(t_simparams *params_, RRuntime &R_, Grid &grid_, ChemWorker::ChemWorker(SimParams &params, RRuntime &R_, Grid &grid_,
MPI_Comm dht_comm) MPI_Comm dht_comm)
: params(params_), ChemSim(params_, R_, grid_) { : ChemSim(params, R_, grid_) {
this->dt_differ = params->dt_differ; t_simparams tmp = params.getNumParams();
this->dht_enabled = params->dht_enabled;
this->dht_size_per_process = params->dht_size_per_process; this->dt_differ = tmp.dt_differ;
this->dht_file = params->dht_file; this->dht_enabled = tmp.dht_enabled;
this->dht_snaps = params->dht_snaps; this->dht_size_per_process = tmp.dht_size_per_process;
this->dht_snaps = tmp.dht_snaps;
this->dht_file = params.getDHTFile();
mpi_buffer = (double *)calloc((wp_size * (grid.getCols())) + BUFFER_OFFSET, mpi_buffer = (double *)calloc((wp_size * (grid.getCols())) + BUFFER_OFFSET,
sizeof(double)); sizeof(double));
@ -49,9 +52,9 @@ ChemWorker::ChemWorker(t_simparams *params_, RRuntime &R_, Grid &grid_,
if (!dht_file.empty()) readFile(); if (!dht_file.empty()) readFile();
// set size // set size
dht_flags.resize(params->wp_size, true); dht_flags.resize(wp_size, true);
// assign all elements to true (default) // assign all elements to true (default)
dht_flags.assign(params->wp_size, true); dht_flags.assign(wp_size, true);
timing[0] = 0.0; timing[0] = 0.0;
timing[1] = 0.0; timing[1] = 0.0;
@ -178,9 +181,9 @@ void ChemWorker::doWork(MPI_Status &probe_status) {
// R.parseEvalQ("print(head(work_package_full))"); // R.parseEvalQ("print(head(work_package_full))");
// R.parseEvalQ("print( c(length(dht_flags), nrow(work_package_full)) )"); // R.parseEvalQ("print( c(length(dht_flags), nrow(work_package_full)) )");
grid.importWP(mpi_buffer, params->wp_size); grid.importWP(mpi_buffer, wp_size);
if (params->dht_enabled) { if (dht_enabled) {
R.parseEvalQ("work_package <- work_package_full[dht_flags,]"); R.parseEvalQ("work_package <- work_package_full[dht_flags,]");
} else { } else {
R.parseEvalQ("work_package <- work_package_full"); R.parseEvalQ("work_package <- work_package_full");

2
src/util/CMakeLists.txt
View File

@ -1,4 +1,4 @@
add_library(POET_Util RRuntime.cpp Parser.cpp) add_library(POET_Util RRuntime.cpp SimParams.cpp)
target_include_directories(POET_Util PUBLIC ${CMAKE_CURRENT_SOURCE_DIR} ${R_INCLUDE_DIRS}) target_include_directories(POET_Util PUBLIC ${CMAKE_CURRENT_SOURCE_DIR} ${R_INCLUDE_DIRS})
target_link_libraries(POET_Util ${R_LIBRARIES}) target_link_libraries(POET_Util ${R_LIBRARIES})
target_compile_definitions(POET_Util PUBLIC STRICT_R_HEADERS) target_compile_definitions(POET_Util PUBLIC STRICT_R_HEADERS)

148
src/util/Parser.cpp
View File

@ -1,148 +0,0 @@
#include "Parser.h"
#include <Rcpp.h>
#include <iostream>
using namespace poet;
using namespace std;
using namespace Rcpp;
Parser::Parser(char *argv[], int world_rank_, int world_size_)
: cmdl(argv), world_rank(world_rank_), world_size(world_size_) {
this->simparams.world_rank = world_rank_;
this->simparams.world_size = world_size;
}
int Parser::parseCmdl() {
// if user asked for help
if (cmdl[{"help", "h"}]) {
if (world_rank == 0) {
cout << "Todo" << endl
<< "See README.md for further information." << endl;
}
return PARSER_HELP;
}
// if positional arguments are missing
else if (!cmdl(2)) {
if (world_rank == 0) {
cerr << "ERROR. Kin needs 2 positional arguments: " << endl
<< "1) the R script defining your simulation and" << endl
<< "2) the directory prefix where to save results and profiling"
<< endl;
}
return PARSER_ERROR;
}
// collect all parameters which are not known, print them to stderr and return
// with PARSER_ERROR
std::list<std::string> unknownOptions = validateOptions();
if (!unknownOptions.empty()) {
if (world_rank == 0) {
cerr << "Unrecognized option(s):\n" << endl;
for (auto option : unknownOptions) {
cerr << option << endl;
}
cerr << "\nMake sure to use available options. Exiting!" << endl;
}
return PARSER_ERROR;
}
/*Parse DHT arguments*/
simparams.dht_enabled = cmdl["dht"];
// cout << "CPP: DHT is " << ( dht_enabled ? "ON" : "OFF" ) << '\n';
if (simparams.dht_enabled) {
cmdl("dht-strategy", 0) >> simparams.dht_strategy;
// cout << "CPP: DHT strategy is " << dht_strategy << endl;
cmdl("dht-signif", 5) >> dht_significant_digits;
// cout << "CPP: DHT significant digits = " << dht_significant_digits <<
// endl;
simparams.dht_log = !(cmdl["dht-nolog"]);
// cout << "CPP: DHT logarithm before rounding: " << ( dht_logarithm ? "ON"
// : "OFF" ) << endl;
cmdl("dht-size", DHT_SIZE_PER_PROCESS) >> simparams.dht_size_per_process;
// cout << "CPP: DHT size per process (Byte) = " << dht_size_per_process <<
// endl;
cmdl("dht-snaps", 0) >> simparams.dht_snaps;
cmdl("dht-file") >> simparams.dht_file;
}
/*Parse work package size*/
cmdl("work-package-size", WORK_PACKAGE_SIZE_DEFAULT) >> simparams.wp_size;
/*Parse output options*/
simparams.store_result = !cmdl["ignore-result"];
/* rank 0 will summarize all extracted values to stdout */
if (world_rank == 0) {
cout << "CPP: Complete results storage is "
<< (simparams.store_result ? "ON" : "OFF") << endl;
cout << "CPP: Work Package Size: " << simparams.wp_size << endl;
cout << "CPP: DHT is " << (simparams.dht_enabled ? "ON" : "OFF") << '\n';
if (simparams.dht_enabled) {
cout << "CPP: DHT strategy is " << simparams.dht_strategy << endl;
cout << "CPP: DHT key default digits (ignored if 'signif_vector' is "
"defined) = "
<< dht_significant_digits << endl;
cout << "CPP: DHT logarithm before rounding: "
<< (simparams.dht_log ? "ON" : "OFF") << endl;
cout << "CPP: DHT size per process (Byte) = "
<< simparams.dht_size_per_process << endl;
cout << "CPP: DHT save snapshots is " << simparams.dht_snaps << endl;
cout << "CPP: DHT load file is " << simparams.dht_file << endl;
}
}
cmdl(1) >> simparams.filesim;
cmdl(2) >> simparams.out_dir;
return PARSER_OK;
}
void Parser::parseR(RRuntime &R) {
// if local_rank == 0 then master else worker
R["local_rank"] = simparams.world_rank;
// assign a char* (string) to 'filesim'
R["filesim"] = wrap(simparams.filesim);
// assign a char* (string) to 'fileout'
R["fileout"] = wrap(simparams.out_dir);
// pass the boolean "store_result" to the R process
R["store_result"] = simparams.store_result;
// worker count
R["n_procs"] = simparams.world_size - 1;
// work package size
R["work_package_size"] = simparams.wp_size;
// dht enabled?
R["dht_enabled"] = simparams.dht_enabled;
// log before rounding?
R["dht_log"] = simparams.dht_log;
// eval the init string, ignoring any returns
R.parseEvalQ("source(filesim)");
}
t_simparams Parser::getParams() { return this->simparams; }
std::list<std::string> Parser::validateOptions() {
/* store all unknown parameters here */
std::list<std::string> retList;
/* loop over all flags and compare to given flaglist*/
for (auto &flag : cmdl.flags()) {
if (!(flaglist.find(flag) != flaglist.end())) retList.push_back(flag);
}
/* and loop also over params and compare to given paramlist */
for (auto &param : cmdl.params()) {
if (!(paramlist.find(param.first) != paramlist.end()))
retList.push_back(param.first);
}
return retList;
}

146
src/util/Parser.h
View File

@ -1,146 +0,0 @@
#ifndef PARSER_H
#define PARSER_H
#include <RRuntime.h>
#include <string>
#include "SimParams.h"
#include "argh.h"
/** Return value if no error occured */
#define PARSER_OK 0
/** Return value if error occured during parsing of program arguments */
#define PARSER_ERROR -1
/** Return value if user asked for help message with program parameter */
#define PARSER_HELP -2
/** Standard DHT Size (Defaults to 1 GiB) */
#define DHT_SIZE_PER_PROCESS 1073741824
/** Standard work package size */
#define WORK_PACKAGE_SIZE_DEFAULT 5
namespace poet {
/**
* @brief Reads information from program arguments and R runtime
*
* Providing functions to initialize parameters of the simulation using command
* line parameters and parameters from the R runtime. This class will also parse
* arguments from the commandline and decides if argument is known or unknown.
*
*/
class Parser {
public:
/**
* @brief Construct a new Parser object
*
* With all given parameters a new instance of this class will be created.
*
* @param argv Argument value of the program
* @param world_rank Rank of process inside MPI_COMM_WORLD
* @param world_size Size of communicator MPI_COMM_WORLD
*/
Parser(char *argv[], int world_rank, int world_size);
/**
* @brief Parse program arguments
*
* This is done by the argh.h library.
*
* First, the function will check if there is a flag 'help' or 'h'. If this is
* the case a help message is printed and the function will return with
* PARSER_HELP.
*
* Second, if there are not 2 positional arguments an error will be printed to
* stderr and the function returns with PARSER_ERROR.
*
* Then all given program parameters and flags will be read and checked, if
* there are known by validateOptions. A list of all unknown options might be
* returned, printed out and the function will return with PARSER_ERROR.
* Oterhwise the function continuos.
*
* Now all program arguments will be stored inside t_simparams struct, printed
* out and the function returns with PARSER_OK.
*
* @return int Returns with 0 if no error occured, otherwise value less than 0
* is returned.
*/
int parseCmdl();
/**
* @brief Distribute all known parameters to R runtime.
*
* All stored parameters are distributed to the R runtime.
*
* @todo This function might be redundant and can be put into parseCmdl.
*
* @param R Instance of RRuntime
*/
void parseR(RRuntime &R);
/**
* @brief Get the Params object
*
* Return all parsed simulation parameters. Should be called after parseCmdl.
*
* @return t_simparams Struct of t_simparams
*/
t_simparams getParams();
private:
/**
* @brief Validate program parameters and flags
*
* Therefore this function iterates over the list of flags and parameters and
* compare them to the class member flagList and paramList. If a program
* argument is not included it is put to a list. This list will be returned.
*
* @return std::list<std::string> List with all unknown parameters. Might be
* empty.
*/
std::list<std::string> validateOptions();
/**
* @brief Contains all valid program flags.
*
*/
const std::set<std::string> flaglist{"ignore-result", "dht", "dht-nolog"};
/**
* @brief Contains all valid program parameters.
*
*/
const std::set<std::string> paramlist{"work-package-size", "dht-signif",
"dht-strategy", "dht-size",
"dht-snaps", "dht-file"};
/**
* @brief Instance of argh class
*
* This class will be instantiate inside constructor of this class object.
*
*/
argh::parser cmdl;
/**
* @brief Struct containing all simulation parameters
*
*/
t_simparams simparams;
/**
* @brief Rank of process inside MPI_COMM_WORLD
*
*/
int world_rank;
/**
* @brief Size of MPI_COMM_WORLD
*
*/
int world_size;
int dht_significant_digits;
};
} // namespace poet
#endif // PARSER_H

240
src/util/SimParams.cpp Normal file
View File

@ -0,0 +1,240 @@
#include "SimParams.h"
#include <Rcpp.h>
#include <iostream>
using namespace poet;
using namespace std;
using namespace Rcpp;
SimParams::SimParams(int world_rank_, int world_size_) {
this->simparams.world_rank = world_rank_;
this->simparams.world_size = world_size_;
}
int SimParams::parseFromCmdl(char *argv[], RRuntime &R) {
// initialize argh object
argh::parser cmdl(argv);
// if user asked for help
if (cmdl[{"help", "h"}]) {
if (simparams.world_rank == 0) {
cout << "Todo" << endl
<< "See README.md for further information." << endl;
}
return PARSER_HELP;
}
// if positional arguments are missing
else if (!cmdl(2)) {
if (simparams.world_rank == 0) {
cerr << "ERROR. Kin needs 2 positional arguments: " << endl
<< "1) the R script defining your simulation and" << endl
<< "2) the directory prefix where to save results and profiling"
<< endl;
}
return PARSER_ERROR;
}
// collect all parameters which are not known, print them to stderr and return
// with PARSER_ERROR
std::list<std::string> optionsError = validateOptions(cmdl);
if (!optionsError.empty()) {
if (simparams.world_rank == 0) {
cerr << "Unrecognized option(s):\n" << endl;
for (auto option : optionsError) {
cerr << option << endl;
}
cerr << "\nMake sure to use available options. Exiting!" << endl;
}
return PARSER_ERROR;
}
/*Parse DHT arguments*/
simparams.dht_enabled = cmdl["dht"];
// cout << "CPP: DHT is " << ( dht_enabled ? "ON" : "OFF" ) << '\n';
if (simparams.dht_enabled) {
cmdl("dht-strategy", 0) >> simparams.dht_strategy;
// cout << "CPP: DHT strategy is " << dht_strategy << endl;
cmdl("dht-signif", 5) >> simparams.dht_significant_digits;
// cout << "CPP: DHT significant digits = " << dht_significant_digits <<
// endl;
simparams.dht_log = !(cmdl["dht-nolog"]);
// cout << "CPP: DHT logarithm before rounding: " << ( dht_logarithm ? "ON"
// : "OFF" ) << endl;
cmdl("dht-size", DHT_SIZE_PER_PROCESS) >> simparams.dht_size_per_process;
// cout << "CPP: DHT size per process (Byte) = " << dht_size_per_process <<
// endl;
cmdl("dht-snaps", 0) >> simparams.dht_snaps;
cmdl("dht-file") >> dht_file;
}
/*Parse work package size*/
cmdl("work-package-size", WORK_PACKAGE_SIZE_DEFAULT) >> simparams.wp_size;
/*Parse output options*/
simparams.store_result = !cmdl["ignore-result"];
if (simparams.world_rank == 0) {
cout << "CPP: Complete results storage is "
<< (simparams.store_result ? "ON" : "OFF") << endl;
cout << "CPP: Work Package Size: " << simparams.wp_size << endl;
cout << "CPP: DHT is " << (simparams.dht_enabled ? "ON" : "OFF") << '\n';
if (simparams.dht_enabled) {
cout << "CPP: DHT strategy is " << simparams.dht_strategy << endl;
cout << "CPP: DHT key default digits (ignored if 'signif_vector' is "
"defined) = "
<< simparams.dht_significant_digits << endl;
cout << "CPP: DHT logarithm before rounding: "
<< (simparams.dht_log ? "ON" : "OFF") << endl;
cout << "CPP: DHT size per process (Byte) = "
<< simparams.dht_size_per_process << endl;
cout << "CPP: DHT save snapshots is " << simparams.dht_snaps << endl;
cout << "CPP: DHT load file is " << dht_file << endl;
}
}
cmdl(1) >> filesim;
cmdl(2) >> out_dir;
/* distribute information to R runtime */
// if local_rank == 0 then master else worker
R["local_rank"] = simparams.world_rank;
// assign a char* (string) to 'filesim'
R["filesim"] = wrap(filesim);
// assign a char* (string) to 'fileout'
R["fileout"] = wrap(out_dir);
// pass the boolean "store_result" to the R process
R["store_result"] = simparams.store_result;
// worker count
R["n_procs"] = simparams.world_size - 1;
// work package size
R["work_package_size"] = simparams.wp_size;
// dht enabled?
R["dht_enabled"] = simparams.dht_enabled;
// log before rounding?
R["dht_log"] = simparams.dht_log;
// eval the init string, ignoring any returns
R.parseEvalQ("source(filesim)");
return PARSER_OK;
}
void SimParams::initVectorParams(RRuntime &R, int col_count) {
if (simparams.dht_enabled) {
// cout << "\nCreating DHT\n";
// determine size of dht entries
// int dht_data_size = grid.getCols() * sizeof(double);
// int dht_key_size =
// grid.getCols() * sizeof(double) + (params.dt_differ *
// sizeof(double));
// // // determine bucket count for preset memory usage
// // // bucket size is key + value + 1 byte for status
// int dht_buckets_per_process =
// params.dht_size_per_process / (1 + dht_data_size + dht_key_size);
// MDL : following code moved here from worker.cpp
/*Load significance vector from R setup file (or set default)*/
bool signif_vector_exists = R.parseEval("exists('signif_vector')");
if (signif_vector_exists) {
dht_signif_vector = as<std::vector<int>>(R["signif_vector"]);
} else {
dht_signif_vector.assign(col_count, simparams.dht_significant_digits);
}
/*Load property type vector from R setup file (or set default)*/
bool prop_type_vector_exists = R.parseEval("exists('prop_type')");
if (prop_type_vector_exists) {
dht_prop_type_vector = as<std::vector<string>>(R["prop_type"]);
} else {
dht_prop_type_vector.assign(col_count, "act");
}
if (simparams.world_rank == 0) {
// // print only on master, values are equal on all workes
// cout << "CPP: dht_data_size: " << dht_data_size << "\n";
// cout << "CPP: dht_key_size: " << dht_key_size << "\n";
// cout << "CPP: dht_buckets_per_process: " << dht_buckets_per_process
// << endl;
// MDL: new output on signif_vector and prop_type
if (signif_vector_exists) {
cout << "CPP: using problem-specific rounding digits: " << endl;
R.parseEval(
"print(data.frame(prop=prop, type=prop_type, "
"digits=signif_vector))");
} else {
cout << "CPP: using DHT default rounding digits = "
<< simparams.dht_significant_digits << endl;
}
// MDL: pass to R the DHT stuff. These variables exist
// only if dht_enabled is true
R["dht_final_signif"] = dht_signif_vector;
R["dht_final_proptype"] = dht_prop_type_vector;
}
}
}
// void SimParams::parseR(RRuntime &R) {
// // if local_rank == 0 then master else worker
// R["local_rank"] = simparams.world_rank;
// // assign a char* (string) to 'filesim'
// R["filesim"] = wrap(simparams.filesim);
// // assign a char* (string) to 'fileout'
// R["fileout"] = wrap(simparams.out_dir);
// // pass the boolean "store_result" to the R process
// R["store_result"] = simparams.store_result;
// // worker count
// R["n_procs"] = simparams.world_size - 1;
// // work package size
// R["work_package_size"] = simparams.wp_size;
// // dht enabled?
// R["dht_enabled"] = simparams.dht_enabled;
// // log before rounding?
// R["dht_log"] = simparams.dht_log;
// // eval the init string, ignoring any returns
// R.parseEvalQ("source(filesim)");
// }
void SimParams::setDtDiffer(bool dt_differ) { simparams.dt_differ = dt_differ; }
t_simparams SimParams::getNumParams() { return this->simparams; }
std::vector<int> SimParams::getDHTSignifVector() {
return this->dht_signif_vector;
}
std::vector<std::string> SimParams::getDHTPropTypeVector() {
return this->dht_prop_type_vector;
}
std::string SimParams::getDHTFile() { return this->dht_file; }
std::string SimParams::getFilesim() { return this->filesim; }
std::string SimParams::getOutDir() { return this->out_dir; }
std::list<std::string> SimParams::validateOptions(argh::parser cmdl) {
/* store all unknown parameters here */
std::list<std::string> retList;
/* loop over all flags and compare to given flaglist*/
for (auto &flag : cmdl.flags()) {
if (!(flaglist.find(flag) != flaglist.end())) retList.push_back(flag);
}
/* and loop also over params and compare to given paramlist */
for (auto &param : cmdl.params()) {
if (!(paramlist.find(param.first) != paramlist.end()))
retList.push_back(param.first);
}
return retList;
}

140
src/util/SimParams.h
View File

@ -1,8 +1,22 @@
#ifndef SIMPARAMS_H #ifndef PARSER_H
#define SIMPARAMS_H #define PARSER_H
#include <string> #include <string>
#include <vector>
#include "RRuntime.h"
#include "argh.h"
/** Return value if no error occured */
#define PARSER_OK 0
/** Return value if error occured during parsing of program arguments */
#define PARSER_ERROR -1
/** Return value if user asked for help message with program parameter */
#define PARSER_HELP -2
/** Standard DHT Size (Defaults to 1 GiB) */
#define DHT_SIZE_PER_PROCESS 1073741824
/** Standard work package size */
#define WORK_PACKAGE_SIZE_DEFAULT 5
typedef struct { typedef struct {
int world_size; int world_size;
@ -14,19 +28,119 @@ typedef struct {
int dht_snaps; int dht_snaps;
int dht_strategy; int dht_strategy;
unsigned int dht_size_per_process; unsigned int dht_size_per_process;
int dht_significant_digits;
unsigned int wp_size;
bool store_result;
} t_simparams;
namespace poet {
/**
* @brief Reads information from program arguments and R runtime
*
* Providing functions to initialize parameters of the simulation using command
* line parameters and parameters from the R runtime. This class will also parse
* arguments from the commandline and decides if argument is known or unknown.
*
* Stores and distribute current simulation parameters at any time.
*
*/
class SimParams {
public:
/**
* @brief Construct a new SimParams object
*
* With all given parameters a new instance of this class will be created.
*
* @param world_rank Rank of process inside MPI_COMM_WORLD
* @param world_size Size of communicator MPI_COMM_WORLD
*/
SimParams(int world_rank, int world_size);
/**
* @brief Parse program arguments
*
* This is done by the argh.h library.
*
* First, the function will check if there is a flag 'help' or 'h'. If this is
* the case a help message is printed and the function will return with
* PARSER_HELP.
*
* Second, if there are not 2 positional arguments an error will be printed to
* stderr and the function returns with PARSER_ERROR.
*
* Then all given program parameters and flags will be read and checked, if
* there are known by validateOptions. A list of all unknown options might be
* returned, printed out and the function will return with PARSER_ERROR.
* Oterhwise the function continuos.
*
* Now all program arguments will be stored inside t_simparams struct, printed
* out and the function returns with PARSER_OK.
*
* Also, all parsed agruments are distributed to the R runtime.
*
* @param argv Argument value of the program
* @param R Instantiated R runtime
* @return int Returns with 0 if no error occured, otherwise value less than 0
* is returned.
*/
int parseFromCmdl(char *argv[], RRuntime &R);
void initVectorParams(RRuntime &R, int col_count);
void setDtDiffer(bool dt_differ);
t_simparams getNumParams();
std::vector<int> getDHTSignifVector();
std::vector<std::string> getDHTPropTypeVector();
std::string getDHTFile();
std::string getFilesim();
std::string getOutDir();
private:
/**
* @brief Validate program parameters and flags
*
* Therefore this function iterates over the list of flags and parameters and
* compare them to the class member flagList and paramList. If a program
* argument is not included it is put to a list. This list will be returned.
*
* @return std::list<std::string> List with all unknown parameters. Might be
* empty.
*/
std::list<std::string> validateOptions(argh::parser cmdl);
/**
* @brief Contains all valid program flags.
*
*/
std::set<std::string> flaglist{"ignore-result", "dht", "dht-nolog"};
/**
* @brief Contains all valid program parameters.
*
*/
std::set<std::string> paramlist{"work-package-size", "dht-signif",
"dht-strategy", "dht-size",
"dht-snaps", "dht-file"};
/**
* @brief Struct containing all simulation parameters
*
* Contains only those values which are standard arithmetic C types.
*
*/
t_simparams simparams;
std::vector<int> dht_signif_vector; std::vector<int> dht_signif_vector;
std::vector<std::string> dht_prop_type_vector; std::vector<std::string> dht_prop_type_vector;
std::string dht_file; std::string dht_file;
unsigned int wp_size;
std::string filesim; std::string filesim;
std::string out_dir; std::string out_dir;
};
bool store_result; } // namespace poet
#endif // PARSER_H
// void* R;
// void* grid;
} t_simparams;
#endif // SIMPARAMS_H