From 8363150dda11e1fe853b3237bae1e991de4ba1dc Mon Sep 17 00:00:00 2001
From: KyleKlenk <kyle.c.klenk@gmail.com>
Date: Wed, 5 Oct 2022 20:42:22 +0000
Subject: [PATCH] applied settings changes to the server and client.
Server now sends the client the proper settings.
Client only needs the hostname and the port number given to it by the user.
---
Distributed-Documentation.md | 6 +-
build/includes/summa_actor/summa_client.hpp | 6 ++
build/includes/summa_actor/summa_server.hpp | 16 ++++-
build/source/actors/main.cpp | 51 ++++++-------
.../actors/summa_actor/summa_client.cpp | 28 ++++++--
.../actors/summa_actor/summa_server.cpp | 72 ++++++++++---------
6 files changed, 111 insertions(+), 68 deletions(-)
diff --git a/Distributed-Documentation.md b/Distributed-Documentation.md
index 4f5fb78..1cc65d6 100644
--- a/Distributed-Documentation.md
+++ b/Distributed-Documentation.md
@@ -7,7 +7,11 @@ To use distributed mode. Set the "distributed-mode" setting to true in the Summa
## SUMMA-Server
Run the server with:
- - summaMain -s -c /path/to/config
+ - summaMain -s -c /path/to/config/Summa_Actors_Settings.json
+
+The server gets its settings from Summa_Actors_Settings.json where the user can configure how many HRUs to compute and how many HRUs should be inside a batch.
## SUMMA-Client
Run the cleint with
+ - summaMain -c /path/to/config/Summa_Actors_Settings.json
+ The client needs the Distributed_Settings from the Summa_Actors_Settings.json file. The rest are set by the server.
\ No newline at end of file
diff --git a/build/includes/summa_actor/summa_client.hpp b/build/includes/summa_actor/summa_client.hpp
index 6b8132d..37f171d 100644
--- a/build/includes/summa_actor/summa_client.hpp
+++ b/build/includes/summa_actor/summa_client.hpp
@@ -2,6 +2,7 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
+#include "settings_functions.hpp"
#include <string>
#include <optional>
@@ -15,6 +16,11 @@ struct summa_client_state {
actor summa_actor_ref;
int batch_id;
int client_id; // id held by server
+
+ Summa_Actor_Settings summa_actor_settings;
+ File_Access_Actor_Settings file_access_actor_settings;
+ Job_Actor_Settings job_actor_settings;
+ HRU_Actor_Settings hru_actor_settings;
};
behavior summa_client(stateful_actor<summa_client_state>* self, std::optional<std::string> config_path);
behavior unconnected(stateful_actor<summa_client_state>*);
diff --git a/build/includes/summa_actor/summa_server.hpp b/build/includes/summa_actor/summa_server.hpp
index 2a082b3..69282b1 100644
--- a/build/includes/summa_actor/summa_server.hpp
+++ b/build/includes/summa_actor/summa_server.hpp
@@ -3,6 +3,7 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "batch_manager.hpp"
+#include "settings_functions.hpp"
#include "client.hpp"
#include <string>
#include <optional>
@@ -11,9 +12,7 @@
namespace caf {
struct summa_server_state {
- int total_hru_count;
int num_clients;
- int num_hru_per_batch;
int batches_remaining = 0;
int batches_solved = 0;
std::string config_path;
@@ -22,9 +21,20 @@ struct summa_server_state {
std::vector<Batch> failed_batches;
std::vector<Client> client_list;
std::string csv_output_name;
+
+
+ Distributed_Settings distributed_settings;
+ Summa_Actor_Settings summa_actor_settings;
+ File_Access_Actor_Settings file_access_actor_settings;
+ Job_Actor_Settings job_actor_settings;
+ HRU_Actor_Settings hru_actor_settings;
+
};
-behavior summa_server(stateful_actor<summa_server_state>* self, std::string config_path);
+behavior summa_server(stateful_actor<summa_server_state>* self, Distributed_Settings distributed_settings,
+ Summa_Actor_Settings summa_actor_settings, File_Access_Actor_Settings file_access_actor_settings,
+ Job_Actor_Settings job_actor_settings, HRU_Actor_Settings hru_actor_settings);
int assembleBatches(stateful_actor<summa_server_state>* self);
std::optional<int> getUnsolvedBatchID(stateful_actor<summa_server_state>* self);
+void initializeCSVOutput(std::string csv_output_name);
}
\ No newline at end of file
diff --git a/build/source/actors/main.cpp b/build/source/actors/main.cpp
index 02da6f2..20bfd59 100644
--- a/build/source/actors/main.cpp
+++ b/build/source/actors/main.cpp
@@ -37,41 +37,39 @@ class config : public actor_system_config {
}
};
-void run_client(actor_system& system, const config& cfg) {
+void run_client(actor_system& system, const config& cfg, Distributed_Settings distributed_settings) {
scoped_actor self{system};
- std::string host;
- uint16_t port;
aout(self) << "Starting SUMMA-Client in Distributed Mode\n";
- host = getSettings(cfg.config_file, "DistributedSettings", "host", host).value_or("");
- port = getSettings(cfg.config_file, "DistributedSettings", "port", port).value_or(-1);
- if (host == "" || port == -1) {
+ if (distributed_settings.hostname == "" || distributed_settings.port == -1) {
aout(self) << "ERROR: run_client() host and port - CHECK SETTINGS FILE\n";
return;
}
std::optional<std::string> path = cfg.config_file;
auto c = system.spawn(summa_client, path);
- if (!host.empty() && port > 0) {
- anon_send(c, connect_atom_v, host, port);
+ if (!distributed_settings.hostname.empty() && distributed_settings.port > 0) {
+ anon_send(c, connect_atom_v, distributed_settings.hostname , (uint16_t) distributed_settings.port );
} else {
aout(self) << "No Server Config" << std::endl;
}
}
-void run_server(actor_system& system, const config& cfg) {
+void run_server(actor_system& system, const config& cfg, Distributed_Settings distributed_settings,
+ Summa_Actor_Settings summa_actor_settings, File_Access_Actor_Settings file_access_actor_settings,
+ Job_Actor_Settings job_actor_settings, HRU_Actor_Settings hru_actor_settings) {
scoped_actor self{system};
- uint16_t port;
+ int err;
- port = getSettings(cfg.config_file, "DistributedSettings", "port", port).value_or(-1);
- if (port == -1) {
+ if (distributed_settings.port == -1) {
aout(self) << "ERROR: run_server() port - CHECK SETTINGS FILE\n";
return;
}
- auto server = system.spawn(summa_server, cfg.config_file);
- aout(self) << "Attempting to publish summa_server_actor on port " << port << std::endl;
- auto is_port = io::publish(server, port);
+ auto server = system.spawn(summa_server, distributed_settings,
+ summa_actor_settings, file_access_actor_settings, job_actor_settings, hru_actor_settings);
+ aout(self) << "Attempting to publish summa_server_actor on port " << distributed_settings.port << std::endl;
+ auto is_port = io::publish(server, distributed_settings.port);
if (!is_port) {
std::cerr << "********PUBLISH FAILED*******" << to_string(is_port.error()) << "\n";
return;
@@ -80,22 +78,21 @@ void run_server(actor_system& system, const config& cfg) {
}
-
-
void caf_main(actor_system& sys, const config& cfg) {
scoped_actor self{sys};
int err;
+
Distributed_Settings distributed_settings;
Summa_Actor_Settings summa_actor_settings;
File_Access_Actor_Settings file_access_actor_settings;
Job_Actor_Settings job_actor_settings;
HRU_Actor_Settings hru_actor_settings;
err = read_settings_from_json(cfg.config_file,
- distributed_settings,
- summa_actor_settings,
- file_access_actor_settings,
- job_actor_settings,
- hru_actor_settings);
+ distributed_settings,
+ summa_actor_settings,
+ file_access_actor_settings,
+ job_actor_settings,
+ hru_actor_settings);
aout(self) << "Printing Settings For SUMMA Simulation\n";
@@ -105,8 +102,12 @@ void caf_main(actor_system& sys, const config& cfg) {
if (distributed_settings.distributed_mode) {
// only command line arguments needed are config_file and server-mode
- auto system = cfg.server_mode ? run_server : run_client;
- system(sys, cfg);
+ if (cfg.server_mode) {
+ run_server(sys, cfg, distributed_settings, summa_actor_settings,
+ file_access_actor_settings, job_actor_settings, hru_actor_settings);
+ } else {
+ run_client(sys, cfg, distributed_settings);
+ }
} else {
// Configure command line arguments
@@ -127,7 +128,7 @@ void caf_main(actor_system& sys, const config& cfg) {
"fileManger is set with the \"-c\" option\n";
aout(self) << "EXAMPLE: ./summaMain -g 1 -n 10 -c location/of/config \n";
return;
- }
+ }
auto summa = sys.spawn(summa_actor, cfg.startGRU, cfg.countGRU, summa_actor_settings,
file_access_actor_settings, job_actor_settings, hru_actor_settings, self);
diff --git a/build/source/actors/summa_actor/summa_client.cpp b/build/source/actors/summa_actor/summa_client.cpp
index f22ae2f..89606b3 100644
--- a/build/source/actors/summa_actor/summa_client.cpp
+++ b/build/source/actors/summa_actor/summa_client.cpp
@@ -73,6 +73,21 @@ behavior running(stateful_actor<summa_client_state>* self, const actor& server_a
self->send(server_actor, connect_to_server_v, self, self->state.hostname);
return {
+ // Response from the server on successful connection
+ [=](connect_to_server, int client_id, Summa_Actor_Settings summa_actor_settings, File_Access_Actor_Settings file_access_actor_settings,
+ Job_Actor_Settings job_actor_settings, HRU_Actor_Settings hru_actor_settings) {
+
+ aout(self) << "Successfully Connected to Server Actor \n";
+
+ self->state.client_id = client_id;
+ self->state.summa_actor_settings = summa_actor_settings;
+ self->state.file_access_actor_settings = file_access_actor_settings;
+ self->state.job_actor_settings = job_actor_settings;
+ self->state.hru_actor_settings = hru_actor_settings;
+
+ },
+
+
[=](batch, int client_id, int batch_id, int start_hru, int num_hru, std::string config_path) {
aout(self) << "\nReceived batch to compute" << "\n";
aout(self) << "BatchID = " << batch_id << "\n";
@@ -83,11 +98,14 @@ behavior running(stateful_actor<summa_client_state>* self, const actor& server_a
self->state.batch_id = batch_id;
- // self->state.summa_actor_ref = self->spawn(summa_actor,
- // start_hru,
- // num_hru,
- // self->state.config_path.value_or(config_path),
- // self);
+ self->state.summa_actor_ref = self->spawn(summa_actor,
+ start_hru,
+ num_hru,
+ self->state.summa_actor_settings,
+ self->state.file_access_actor_settings,
+ self->state.job_actor_settings,
+ self->state.hru_actor_settings,
+ self);
},
diff --git a/build/source/actors/summa_actor/summa_server.cpp b/build/source/actors/summa_actor/summa_server.cpp
index d6f8039..abd6176 100644
--- a/build/source/actors/summa_actor/summa_server.cpp
+++ b/build/source/actors/summa_actor/summa_server.cpp
@@ -10,37 +10,20 @@
namespace caf {
-behavior summa_server(stateful_actor<summa_server_state>* self, std::string config_path) {
+behavior summa_server(stateful_actor<summa_server_state>* self, Distributed_Settings distributed_settings,
+ Summa_Actor_Settings summa_actor_settings, File_Access_Actor_Settings file_access_actor_settings,
+ Job_Actor_Settings job_actor_settings, HRU_Actor_Settings hru_actor_settings) {
+
aout(self) << "Summa Server has Started \n";
- self->state.config_path = config_path;
- // --------------------------Initalize Settings --------------------------
- self->state.total_hru_count = getSettings(self->state.config_path, "SimulationSettings", "total_hru_count",
- self->state.total_hru_count).value_or(-1);
- if (self->state.total_hru_count == -1) {
- aout(self) << "ERROR: With total_hru_count - CHECK Summa_Actors_Settings.json\n";
- }
- self->state.num_hru_per_batch = getSettings(self->state.config_path, "SimulationSettings", "num_hru_per_batch",
- self->state.num_hru_per_batch).value_or(-1);
- if (self->state.num_hru_per_batch == -1) {
- aout(self) << "ERROR: With num_hru_per_batch - CHECK Summa_Actors_Settings.json\n";
- }
- // --------------------------Initalize Settings --------------------------
+ self->state.distributed_settings = distributed_settings;
+ self->state.summa_actor_settings = summa_actor_settings;
+ self->state.file_access_actor_settings = file_access_actor_settings;
+ self->state.job_actor_settings = job_actor_settings;
+ self->state.hru_actor_settings = hru_actor_settings;
+
- // -------------------------- Initalize CSV ------------------------------
- std::ofstream csv_output;
self->state.csv_output_name = "Batch_Results.csv";
- csv_output.open(self->state.csv_output_name, std::ios_base::out);
- csv_output <<
- "Batch_ID," <<
- "Start_HRU," <<
- "Num_HRU," <<
- "Hostname," <<
- "Run_Time," <<
- "Read_Time," <<
- "Write_Time,"<<
- "Status\n";
- csv_output.close();
- // -------------------------- Initalize CSV ------------------------------
+ initializeCSVOutput(self->state.csv_output_name);
aout(self) << "Assembling HRUs into Batches\n";
if (assembleBatches(self) == -1) {
@@ -60,6 +43,12 @@ behavior summa_server(stateful_actor<summa_server_state>* self, std::string conf
int client_id = self->state.client_list.size(); // So we can lookup the client in O(1) time
self->state.client_list.push_back(Client(client_id, client, hostname));
+ // Tell client they are connected
+ self->send(client, connect_to_server_v, client_id, self->state.summa_actor_settings,
+ self->state.file_access_actor_settings, self->state.job_actor_settings, self->state.hru_actor_settings);
+
+
+
std::optional<int> batch_id = getUnsolvedBatchID(self);
if (batch_id.has_value()) {
// update the batch in the batch list with the host and actor_ref
@@ -141,21 +130,21 @@ behavior summa_server(stateful_actor<summa_server_state>* self, std::string conf
int assembleBatches(stateful_actor<summa_server_state>* self) {
- int remaining_hru_to_batch = self->state.total_hru_count;
- int count_index = 0; // this is like the offset for slurm bash scripts
+ int remaining_hru_to_batch = self->state.distributed_settings.total_hru_count;
+ int count_index = 0;
int start_hru = 1;
while(remaining_hru_to_batch > 0) {
- if (self->state.num_hru_per_batch > remaining_hru_to_batch) {
+ if (self->state.distributed_settings.num_hru_per_batch > remaining_hru_to_batch) {
self->state.batch_list.push_back(Batch(count_index, start_hru,
remaining_hru_to_batch));
remaining_hru_to_batch = 0;
} else {
self->state.batch_list.push_back(Batch(count_index, start_hru,
- self->state.num_hru_per_batch));
+ self->state.distributed_settings.num_hru_per_batch));
- remaining_hru_to_batch -= self->state.num_hru_per_batch;
- start_hru += self->state.num_hru_per_batch;
+ remaining_hru_to_batch -= self->state.distributed_settings.num_hru_per_batch;
+ start_hru += self->state.distributed_settings.num_hru_per_batch;
count_index += 1;
}
}
@@ -172,4 +161,19 @@ std::optional<int> getUnsolvedBatchID(stateful_actor<summa_server_state>* self)
return {};
}
+void initializeCSVOutput(std::string csv_output_name) {
+ std::ofstream csv_output;
+ csv_output.open(csv_output_name, std::ios_base::out);
+ csv_output <<
+ "Batch_ID," <<
+ "Start_HRU," <<
+ "Num_HRU," <<
+ "Hostname," <<
+ "Run_Time," <<
+ "Read_Time," <<
+ "Write_Time,"<<
+ "Status\n";
+ csv_output.close();
+}
+
} // end namespace
--
GitLab