rmontanana/Platform
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Continue grid Experiment

Browse Source
This commit is contained in:
Ricardo Montañana Gómez
2025-01-14 22:04:23 +01:00
parent 386faf960e
commit 9a9a9fb17a
7 changed files with 226 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
src/grid/GridBase.cpp
View File

@@ -26,39 +26,9 @@ namespace platform {
std::string id = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
auto idx = rank % id.size();
return *(colors.begin() + rank % colors.size()) + id[idx];
};
json GridBase::build_tasks()
}
void GridBase::shuffle_and_progress_bar(json& tasks)
{
/*
* Each task is a json object with the following structure:
* {
* "dataset": "dataset_name",
* "idx_dataset": idx_dataset, // used to identify the dataset in the results
* // this index is relative to the list of used datasets in the actual run not to the whole datasets list
* "seed": # of seed to use,
* "fold": # of fold to process
* }
*/
auto tasks = json::array();
auto grid = GridData(Paths::grid_input(config.model));
auto datasets = Datasets(false, Paths::datasets());
auto all_datasets = datasets.getNames();
auto datasets_names = filterDatasets(datasets);
for (int idx_dataset = 0; idx_dataset < datasets_names.size(); ++idx_dataset) {
auto dataset = datasets_names[idx_dataset];
for (const auto& seed : config.seeds) {
auto combinations = grid.getGrid(dataset);
for (int n_fold = 0; n_fold < config.n_folds; n_fold++) {
json task = {
{ "dataset", dataset },
{ "idx_dataset", idx_dataset},
{ "seed", seed },
{ "fold", n_fold},
};
tasks.push_back(task);
}
}
}
// Shuffle the array so heavy datasets are eas ier spread across the workers
std::mt19937 g{ 271 }; // Use fixed seed to obtain the same shuffle
std::shuffle(tasks.begin(), tasks.end(), g);
@@ -71,7 +41,6 @@ namespace platform {
std::cout << (i + 1) % 10;
}
std::cout << separator << std::endl << separator << std::flush;
return tasks;
}
void GridBase::summary(json& all_results, json& tasks, struct ConfigMPI& config_mpi)
{
@@ -135,25 +104,16 @@ namespace platform {
total += task["time"].get<double>();
}
if (num_tasks > 1) {
std::cout << Colors::MAGENTA() << setw(3) << std::right << num_tasks;
std::cout << setw(max_dataset) << " Total..." << std::string(10, '.');
std::cout << setw(15) << std::setprecision(7) << std::fixed << total << std::endl;
std::cout << Colors::MAGENTA() << " ";
std::cout << setw(max_dataset) << "Total (" << setw(2) << std::right << num_tasks << ")" << std::string(7, '.');
std::cout << " " << setw(15) << std::setprecision(7) << std::fixed << total << std::endl;
}
}
}
void GridBase::go(struct ConfigMPI& config_mpi)
{
/*
* Each task is a json object with the following structure:
* {
* "dataset": "dataset_name",
* "idx_dataset": idx_dataset, // used to identify the dataset in the results
* // this index is relative to the list of used datasets in the actual run not to the whole datasets list
* "seed": # of seed to use,
* "fold": # of fold to process
* }
*
* This way a task consists in process all combinations of hyperparameters for a dataset, seed and fold
* Each task is a json object with the data needed by the process
*
* The overall process consists in these steps:
* 0. Create the MPI result type & tasks
@@ -170,7 +130,7 @@ namespace platform {
* 2b.1 Consumers announce to the producer that they are ready to receive a task
* 2b.2 Consumers receive the task from the producer and process it
* 2b.3 Consumers send the result to the producer
* 3. Manager select the bests scores for each dataset
* 3. Manager compile results for each dataset
* 3.1 Loop thru all the results obtained from each outer fold (task) and select the best
* 3.2 Save the results
* 3.3 Summary of jobs done
@@ -201,9 +161,11 @@ namespace platform {
//
char* msg;
json tasks;
auto env = platform::DotEnv();
auto datasets = Datasets(config.discretize, Paths::datasets(), env.get("discretize_algo"));
if (config_mpi.rank == config_mpi.manager) {
timer.start();
tasks = build_tasks();
tasks = build_tasks(datasets);
auto tasks_str = tasks.dump();
tasks_size = tasks_str.size();
msg = new char[tasks_size + 1];
@@ -219,8 +181,7 @@ namespace platform {
MPI_Bcast(msg, tasks_size + 1, MPI_CHAR, config_mpi.manager, MPI_COMM_WORLD);
tasks = json::parse(msg);
delete[] msg;
auto env = platform::DotEnv();
auto datasets = Datasets(config.discretize, Paths::datasets(), env.get("discretize_algo"));
if (config_mpi.rank == config_mpi.manager) {
//
@@ -230,10 +191,10 @@ namespace platform {
json all_results = producer(datasets_names, tasks, config_mpi, MPI_Result);
std::cout << separator << std::endl;
//
// 3. Manager select the bests sccores for each dataset
// 3. Manager compile results for each dataset
//
auto results = initializeResults();
select_best_results_folds(results, all_results, config.model);
compile_results(results, all_results, config.model);
//
// 3.2 Save the results
//
@@ -250,5 +211,61 @@ namespace platform {
consumer(datasets, tasks, config, config_mpi, MPI_Result);
}
}
json GridBase::producer(std::vector<std::string>& names, json& tasks, struct ConfigMPI& config_mpi, MPI_Datatype& MPI_Result)
{
Task_Result result;
json results;
int num_tasks = tasks.size();
//
// 2a.1 Producer will loop to send all the tasks to the consumers and receive the results
//
for (int i = 0; i < num_tasks; ++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
store_result(names, result, results);
}
MPI_Send(&i, 1, MPI_INT, status.MPI_SOURCE, TAG_TASK, MPI_COMM_WORLD);
}
//
// 2a.2 Producer will send the end message to all the consumers
//
for (int i = 0; i < config_mpi.n_procs - 1; ++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
store_result(names, result, results);
}
MPI_Send(&i, 1, MPI_INT, status.MPI_SOURCE, TAG_END, MPI_COMM_WORLD);
}
return results;
}
void GridBase::consumer(Datasets& datasets, json& tasks, struct ConfigGrid& config, struct ConfigMPI& config_mpi, MPI_Datatype& MPI_Result)
{
Task_Result result;
//
// 2b.1 Consumers announce to the producer that they are ready to receive a task
//
MPI_Send(&result, 1, MPI_Result, config_mpi.manager, TAG_QUERY, MPI_COMM_WORLD);
int task;
while (true) {
MPI_Status status;
//
// 2b.2 Consumers receive the task from the producer and process it
//
MPI_Recv(&task, 1, MPI_INT, config_mpi.manager, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
if (status.MPI_TAG == TAG_END) {
break;
}
consumer_go(config, config_mpi, tasks, task, datasets, &result);
//
// 2b.3 Consumers send the result to the producer
//
MPI_Send(&result, 1, MPI_Result, config_mpi.manager, TAG_RESULT, MPI_COMM_WORLD);
}
}
}