fabian/actsim-cluster-agent
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

upload fault flags, streamline code

Browse source
This commit is contained in:
Fabian Posch 2025-01-03 12:55:44 +01:00
parent 7f8efe0e73
commit 127074f85e
2 changed files with 53 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
include/actsim_agent/uploader.hpp
View file

@ -28,6 +28,7 @@
#include <cluster/db_types.hpp> #include <cluster/db_types.hpp>
#include <cluster/db_client.hpp> #include <cluster/db_client.hpp>
#include <memory>
#include <thread> #include <thread>
#include "task_interface.hpp" #include "task_interface.hpp"
@ -44,6 +45,7 @@ class Uploader {
void thread_run(); void thread_run();
bool upload_task(std::unique_ptr<OutputType> task); bool upload_task(std::unique_ptr<OutputType> task);
std::string build_fault_flags(const std::unique_ptr<OutputType>& task);
std::unique_ptr<std::thread> uploader_thread; std::unique_ptr<std::thread> uploader_thread;
std::unique_ptr<db::Connection> conn; std::unique_ptr<db::Connection> conn;

68
src/actsim_agent/uploader.cpp
View file

@ -23,9 +23,12 @@
************************************************************************** **************************************************************************
*/ */
#include <memory>
#include <pqxx/pqxx> #include <pqxx/pqxx>
#include <functional> #include <functional>
#include <cluster/db_types.hpp> #include <cluster/db_types.hpp>
#include <sstream>
#include "task_interface.hpp"
#include "util.h" #include "util.h"
#include "uploader.hpp" #include "uploader.hpp"
@ -60,7 +63,7 @@ void Uploader::thread_run() {
// program was halted // program was halted
this->interface.wait_for_finished(); this->interface.wait_for_finished();
DEBUG_PRINT("Uploader was worken up"); DEBUG_PRINT("Uploader was woken up");
// so first we check if we should still be running // so first we check if we should still be running
if (!this->interface.running()) break; if (!this->interface.running()) break;
@ -117,29 +120,60 @@ void Uploader::thread_run() {
bool Uploader::upload_task(std::unique_ptr<OutputType> task) { bool Uploader::upload_task(std::unique_ptr<OutputType> task) {
auto&& task_id = task->id;
auto&& sim_log = task->get_content().first;
auto&& sim_error = task->get_content().second;
auto&& output_tokens = task->output_tokens;
const auto&& fault_flags = build_fault_flags(task);
// make sure any task that is uploaded isn't halted in the database // make sure any task that is uploaded isn't halted in the database
auto task_upload_lambda = []( auto task_upload_lambda = [task_id, sim_log, sim_error, output_tokens, fault_flags](
pqxx::work *txn, pqxx::work *txn
const db::uuid_t *target,
std::vector<std::string> *sim_log,
std::vector<std::string> *sim_error
) { ) {
txn->exec_params0( txn->exec_params0(
"UPDATE sim_outputs SET sim_log = $1, error_log = $2, part_status = 'finished' WHERE id = $3 AND part_status != 'halted';", "UPDATE sim_outputs SET "
*sim_log, " sim_log = $1, "
*sim_error, " error_log = $2, "
*target " output_tokens = $3, "
" fault_flags = $4, "
" part_status = 'finished' "
"WHERE id = $5 AND part_status != 'halted';",
sim_log,
sim_error,
output_tokens,
fault_flags,
task_id
); );
}; };
std::function<void( std::function<void(pqxx::work*)> task_upload_func = task_upload_lambda;
pqxx::work*,
const db::uuid_t*,
std::vector<std::string>*,
std::vector<std::string>*
)> task_upload_func = task_upload_lambda;
DEBUG_PRINT("Updating task " + db::to_string(task->id)); DEBUG_PRINT("Updating task " + db::to_string(task->id));
return this->conn->send_request(&task_upload_func, &(task->id), &(task->get_content().first), &(task->get_content().second)); return this->conn->send_request(&task_upload_func);
}
std::string Uploader::build_fault_flags(const std::unique_ptr<OutputType>& task) {
// bit mask for faults is
// 0: timing
// 1: value
// 2: coding
// 3: glitch
// 4: deadlock
// 5: token count
// 54 3210
// XX XXXX
std::stringstream flags;
flags << (task->fault_token_count ? "1" : "0");
flags << (task->fault_deadlock ? "1" : "0");
flags << (task->fault_glitch ? "1" : "0");
flags << (task->fault_coding ? "1" : "0");
flags << (task->fault_value ? "1" : "0");
flags << (task->fault_timing_deviation ? "1" : "0");
return flags.str();
} }