rmontanana/BayesNet
rmontanana/BayesNet
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 3 Wiki Activity

Set structure & protocol of producer-consumer

Browse Source
This commit is contained in:
Ricardo Montañana Gómez 2023-12-22 12:47:13 +01:00
parent 9b9e91e856
commit e0b7b2d316
Signed by: rmontanana
GPG Key ID: 46064262FD9A7ADE
2 changed files with 202 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

271
src/Platform/GridSearch.cc
View File

@ -149,88 +149,120 @@ namespace platform {
auto colors = { Colors::RED(), Colors::GREEN(), Colors::BLUE(), Colors::MAGENTA(), Colors::CYAN() }; auto colors = { Colors::RED(), Colors::GREEN(), Colors::BLUE(), Colors::MAGENTA(), Colors::CYAN() };
return *(colors.begin() + rank % colors.size()); return *(colors.begin() + rank % colors.size());
} }
void GridSearch::process_task_mpi(struct ConfigMPI& config_mpi, json& task, Datasets& datasets, json& results)
void GridSearch::go_producer_consumer(struct ConfigMPI& config_mpi)
{ {
// Process the task and store the result in the results json /*
Timer timer; * Each task is a json object with the following structure:
timer.start(); * {
auto grid = GridData(Paths::grid_input(config.model)); * "dataset": "dataset_name",
auto dataset = task["dataset"].get<std::string>(); * "seed": # of seed to use,
auto seed = task["seed"].get<int>(); * "model": "model_name",
auto n_fold = task["fold"].get<int>(); * "Fold": # of fold to process
// Generate the hyperparamters combinations * }
auto combinations = grid.getGrid(dataset); *
auto [X, y] = datasets.getTensors(dataset); * The overall process consists in these steps:
auto states = datasets.getStates(dataset); * 0. Create the MPI result type & tasks
auto features = datasets.getFeatures(dataset); * 0.1 Create the MPI result type
auto className = datasets.getClassName(dataset); * 0.2 Manager creates the tasks
* 1. Manager will broadcast the tasks to all the processes
* 1.1 Broadcast the number of tasks
* 1.2 Broadcast the length of the following string
* 1.2 Broadcast the tasks as a char* string
* 2. Workers will receive the tasks and start the process
* 2.1 A method will tell each worker the range of tasks to process
* 2.2 Each worker will process the tasks and generate the best score for each task
* 3. Manager gather the scores from all the workers and find out the best hyperparameters for each dataset
* 3.1 Obtain the maximum size of the results message of all the workers
* 3.2 Gather all the results from the workers into the manager
* 3.3 Compile the results from all the workers
* 3.4 Filter the best hyperparameters for each dataset
*/
// //
// Start working on task // 0.1 Create the MPI result type
// //
Fold* fold; Task_Result result;
if (config.stratified) MPI_Datatype MPI_Result;
fold = new StratifiedKFold(config.n_folds, y, seed); MPI_Datatype type[3] = { MPI_UNSIGNED, MPI_UNSIGNED, MPI_DOUBLE };
else int blocklen[3] = { 1, 1, 1 };
fold = new KFold(config.n_folds, y.size(0), seed); MPI_Aint disp[3];
auto [train, test] = fold->getFold(n_fold); disp[0] = offsetof(struct MPI_Result, idx_dataset);
auto train_t = torch::tensor(train); disp[1] = offsetof(struct MPI_Result, idx_combination);
auto test_t = torch::tensor(test); disp[2] = offsetof(struct MPI_Result, score);
auto X_train = X.index({ "...", train_t }); MPI_Type_create_struct(3, blocklen, disp, type, &MPI_Result);
auto y_train = y.index({ train_t }); MPI_Type_commit(&MPI_Result);
auto X_test = X.index({ "...", test_t }); //
auto y_test = y.index({ test_t }); // 0.2 Manager creates the tasks
auto num = 0; //
double best_fold_score = 0.0; char* msg;
json best_fold_hyper; if (config_mpi.rank == config_mpi.manager) {
for (const auto& hyperparam_line : combinations) { timer.start();
auto hyperparameters = platform::HyperParameters(datasets.getNames(), hyperparam_line); auto tasks = build_tasks_mpi();
Fold* nested_fold; auto tasks_str = tasks.dump();
if (config.stratified) tasks_size = tasks_str.size();
nested_fold = new StratifiedKFold(config.nested, y_train, seed); msg = new char[tasks_size + 1];
else strcpy(msg, tasks_str.c_str());
nested_fold = new KFold(config.nested, y_train.size(0), seed); }
double score = 0.0; //
for (int n_nested_fold = 0; n_nested_fold < config.nested; n_nested_fold++) { // 1. Manager will broadcast the tasks to all the processes
// Nested level fold //
auto [train_nested, test_nested] = nested_fold->getFold(n_nested_fold); MPI_Bcast(&tasks_size, 1, MPI_INT, config_mpi.manager, MPI_COMM_WORLD);
auto train_nested_t = torch::tensor(train_nested); if (config_mpi.rank != config_mpi.manager) {
auto test_nested_t = torch::tensor(test_nested); msg = new char[tasks_size + 1];
auto X_nested_train = X_train.index({ "...", train_nested_t }); }
auto y_nested_train = y_train.index({ train_nested_t }); MPI_Bcast(msg, tasks_size + 1, MPI_CHAR, config_mpi.manager, MPI_COMM_WORLD);
auto X_nested_test = X_train.index({ "...", test_nested_t }); json tasks = json::parse(msg);
auto y_nested_test = y_train.index({ test_nested_t }); delete[] msg;
// Build Classifier with selected hyperparameters //
auto clf = Models::instance()->create(config.model); // 2. All Workers will receive the tasks and start the process
auto valid = clf->getValidHyperparameters(); //
hyperparameters.check(valid, dataset); if (config_mpi.rank == config_mpi.manager) {
clf->setHyperparameters(hyperparameters.get(dataset)); producer(tasks, &MPI_Result);
// Train model } else {
clf->fit(X_nested_train, y_nested_train, features, className, states); consumer(tasks, &MPI_Result);
// Test model }
score += clf->score(X_nested_test, y_nested_test); }
} void producer(json& tasks, MPI_Datatpe& MPI_Result)
delete nested_fold; {
score /= config.nested; Task_Result result;
if (score > best_fold_score) { int num_tasks = tasks.size();
best_fold_score = score; for (int i = 0; i < num_tasks; ++i) {
best_fold_hyper = hyperparam_line; MPI_Status status;
} MPI_recv(&result, 1, MPI_Result, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
if (status.MPI_TAG == TAG_RESULT) {
//Store result
}
MPI_Send(&i, 1, MPI_INT, status.MPI_SOURCE, TAG_TASK, MPI_COMM_WORLD);
}
// Send end message to all workers
for (int i = 0; i < config_mpi.n_procs; ++i) {
MPI_Status status;
MPI_recv(&result, 1, MPI_Result, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
if (status.MPI_TAG == TAG_RESULT) {
//Store result
}
MPI_Send(&i, 1, MPI_INT, status.MPI_SOURCE, TAG_END, MPI_COMM_WORLD);
}
}
void consumer(json& tasks, MPI_Datatpe& MPI_Result)
{
Task_Result result;
// Anounce to the producer
MPI_Send(&result, 1, MPI_Result, config_mpi.manager, TAG_QUERY, MPI_COMM_WORLD);
int task;
while (true) {
MPI_Status status;
MPI_recv(&task, 1, MPI_INT, config_mpi.manager, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
if (status.MPI_TAG == TAG_END) {
break;
}
// Process task
process_task_mpi(config_mpi, task, datasets, results);
// Send result to producer
MPI_Send(&result, 1, MPI_Result, config_mpi.manager, TAG_RESULT, MPI_COMM_WORLD);
} }
delete fold;
// Build Classifier with the best hyperparameters to obtain the best score
auto hyperparameters = platform::HyperParameters(datasets.getNames(), best_fold_hyper);
auto clf = Models::instance()->create(config.model);
auto valid = clf->getValidHyperparameters();
hyperparameters.check(valid, dataset);
clf->setHyperparameters(best_fold_hyper);
clf->fit(X_train, y_train, features, className, states);
best_fold_score = clf->score(X_test, y_test);
// Save results
results[dataset][std::to_string(n_fold)]["score"] = best_fold_score;
results[dataset][std::to_string(n_fold)]["hyperparameters"] = best_fold_hyper;
results[dataset][std::to_string(n_fold)]["seed"] = seed;
results[dataset][std::to_string(n_fold)]["duration"] = timer.getDuration();
std::cout << get_color_rank(config_mpi.rank) << "*" << std::flush;
} }
void GridSearch::go_mpi(struct ConfigMPI& config_mpi) void GridSearch::go_mpi(struct ConfigMPI& config_mpi)
{ {
@ -555,6 +587,89 @@ namespace platform {
} }
return { goatScore, goatHyperparameters }; return { goatScore, goatHyperparameters };
} }
void GridSearch::process_task_mpi(struct ConfigMPI& config_mpi, json& task, Datasets& datasets, json& results)
{
// Process the task and store the result in the results json
Timer timer;
timer.start();
auto grid = GridData(Paths::grid_input(config.model));
auto dataset = task["dataset"].get<std::string>();
auto seed = task["seed"].get<int>();
auto n_fold = task["fold"].get<int>();
// Generate the hyperparamters combinations
auto combinations = grid.getGrid(dataset);
auto [X, y] = datasets.getTensors(dataset);
auto states = datasets.getStates(dataset);
auto features = datasets.getFeatures(dataset);
auto className = datasets.getClassName(dataset);
//
// Start working on task
//
Fold* fold;
if (config.stratified)
fold = new StratifiedKFold(config.n_folds, y, seed);
else
fold = new KFold(config.n_folds, y.size(0), seed);
auto [train, test] = fold->getFold(n_fold);
auto train_t = torch::tensor(train);
auto test_t = torch::tensor(test);
auto X_train = X.index({ "...", train_t });
auto y_train = y.index({ train_t });
auto X_test = X.index({ "...", test_t });
auto y_test = y.index({ test_t });
auto num = 0;
double best_fold_score = 0.0;
json best_fold_hyper;
for (const auto& hyperparam_line : combinations) {
auto hyperparameters = platform::HyperParameters(datasets.getNames(), hyperparam_line);
Fold* nested_fold;
if (config.stratified)
nested_fold = new StratifiedKFold(config.nested, y_train, seed);
else
nested_fold = new KFold(config.nested, y_train.size(0), seed);
double score = 0.0;
for (int n_nested_fold = 0; n_nested_fold < config.nested; n_nested_fold++) {
// Nested level fold
auto [train_nested, test_nested] = nested_fold->getFold(n_nested_fold);
auto train_nested_t = torch::tensor(train_nested);
auto test_nested_t = torch::tensor(test_nested);
auto X_nested_train = X_train.index({ "...", train_nested_t });
auto y_nested_train = y_train.index({ train_nested_t });
auto X_nested_test = X_train.index({ "...", test_nested_t });
auto y_nested_test = y_train.index({ test_nested_t });
// Build Classifier with selected hyperparameters
auto clf = Models::instance()->create(config.model);
auto valid = clf->getValidHyperparameters();
hyperparameters.check(valid, dataset);
clf->setHyperparameters(hyperparameters.get(dataset));
// Train model
clf->fit(X_nested_train, y_nested_train, features, className, states);
// Test model
score += clf->score(X_nested_test, y_nested_test);
}
delete nested_fold;
score /= config.nested;
if (score > best_fold_score) {
best_fold_score = score;
best_fold_hyper = hyperparam_line;
}
}
delete fold;
// Build Classifier with the best hyperparameters to obtain the best score
auto hyperparameters = platform::HyperParameters(datasets.getNames(), best_fold_hyper);
auto clf = Models::instance()->create(config.model);
auto valid = clf->getValidHyperparameters();
hyperparameters.check(valid, dataset);
clf->setHyperparameters(best_fold_hyper);
clf->fit(X_train, y_train, features, className, states);
best_fold_score = clf->score(X_test, y_test);
// Save results
results[dataset][std::to_string(n_fold)]["score"] = best_fold_score;
results[dataset][std::to_string(n_fold)]["hyperparameters"] = best_fold_hyper;
results[dataset][std::to_string(n_fold)]["seed"] = seed;
results[dataset][std::to_string(n_fold)]["duration"] = timer.getDuration();
std::cout << get_color_rank(config_mpi.rank) << "*" << std::flush;
}
json GridSearch::initializeResults() json GridSearch::initializeResults()
{ {
// Load previous results // Load previous results

9
src/Platform/GridSearch.h
View File

@ -30,6 +30,15 @@ namespace platform {
int n_procs; int n_procs;
int manager; int manager;
}; };
typedef struct {
uint idx_dataset;
uint idx_combination;
double score;
} Task_Result;
const TAG_QUERY = 1;
const TAG_RESULT = 2;
const TAG_TASK = 3;
const TAG_END = 4;
class GridSearch { class GridSearch {
public: public:
explicit GridSearch(struct ConfigGrid& config); explicit GridSearch(struct ConfigGrid& config);