Homec4science
Diffusion blackdynamite (features/zeo) Browse

R3127/src/pusher.cc
37bdc45cff01features/zeo

Last Change
Raw File

pusher.cc
View Options

/*
author : Guillaume ANCIAUX (anciaux@labri.fr, g.anciaux@laposte.net)
author : Till JUNGE <till.junge@epfl.ch>
*/
/* -------------------------------------------------------------------------- */
#include "pusher.hh"
#include "sql_pusher.hh"
#include "sql_state_updater.hh"
#include <sstream>
#include <cmath>
/* -------------------------------------------------------------------------- */
namespace BlackDynamite {
Pusher::Pusher() {
this->connection_counter++;
this->run_id = 0;
this->sql_mode = false;
}
/* -------------------------------------------------------------------------- */
Pusher::~Pusher() {
this->connection_counter--;
if (this->connection_counter == 0) {
delete this->sql_connection;
this->sql_connection = NULL;
}
}
/* -------------------------------------------------------------------------- */
inline std::string Pusher::request_increment(std::string keyword, std::string parameter) {
if (!keyword.empty())
return keyword + "=" + parameter + " ";
return "";
}
/* -------------------------------------------------------------------------- */
void Pusher::init() {
char * ptr = NULL;
ptr = getenv("USER");if (!ptr) FATAL("undefined env var");
std::string user = ptr;
ptr = getenv("DBNAME");
std::string dbname;
if (!ptr) dbname = user;
else dbname = ptr;
ptr = getenv("HOST");if (!ptr) FATAL("undefined env var");
std::string host = ptr;
ptr = getenv("SCHEMA");if (!ptr) FATAL("undefined env var");
std::string schema = ptr;
ptr = getenv("RUN_ID");if (!ptr) FATAL("undefined env var");
UInt run_id = atoi(ptr);
// ptr = getenv("HOSTADDR");if (!ptr) FATAL("undefined env var");
// std::string hostaddr = ptr;
// ptr = getenv("PORT");if (!ptr) FATAL("undefined env var");
// std::string port = ptr;
// ptr = getenv("CONNECT_TIMEOUT");if (!ptr) FATAL("undefined env var");
// std::string connect_timeout = ptr;
// ptr = getenv("SLMODE");if (!ptr) FATAL("undefined env var");
// std::string slmode = ptr;
// ptr = getenv("SERVICE");if (!ptr) FATAL("undefined env var");
// std::string service = ptr;
// init(dbname,user,host,hostaddr,port,connect_timeout,slmode,service);
init(dbname,user,host,schema,run_id);
}
/* -------------------------------------------------------------------------- */
void Pusher::init(const std::string & dbname,
const std::string & user,
const std::string & host,
const std::string & schema,
UInt run_id
//const std::string & hostaddr,
//const std::string & port,
//const std::string & connect_timeout,
//const std::string & slmode,
//const std::string & service
) {
std::string options("");
if (this->sql_connection == NULL) {
options += request_increment("dbname" , dbname );
options += request_increment("user" , user );
options += request_increment("host" , host );
// options += request_increment("hostaddr" , hostaddr );
// options += request_increment("port" , port );
//options += request_increment("connect_timeout", connect_timeout);
//options += request_increment("slmode" , slmode );
//options += request_increment("service" , service );
try {
this->sql_connection = new pqxx::connection(options);
this->sql_connection->activate();
} catch (std::runtime_error & e) {
FATAL ("Database connection failed : " << e.what());
} catch (std::exception & e) {
FATAL (e.what());
} catch (...) {
FATAL ("Unknown exception during connection to database");
}
}
this->sql_schema = schema;
if (this->sql_schema.empty()) {
this->scalar_integer.tablename = "scalar_integer";
this->scalar_real.tablename = "scalar_real";
this->vector_real.tablename = "vector_real";
this->vector_integer.tablename = "vector_integer";
this->quantities_tablename = "quantities";
} else {
this->scalar_integer.tablename = this->sql_schema + "." + "scalar_integer";
this->scalar_real.tablename = this->sql_schema + "." + "scalar_real";
this->vector_real.tablename = this->sql_schema + "." + "vector_real";
this->vector_integer.tablename = this->sql_schema + "." + "vector_integer";
this->quantities_tablename = this->sql_schema + ".quantities";
}
this->run_id = run_id;
if (this->run_id <1) {
FATAL("The run id required for the sql dumps has not been specified");
}
this->scalar_integer.run_id = this->run_id;
this->scalar_real.run_id = this->run_id;
this->vector_real.run_id = this->run_id;
this->vector_integer.run_id = this->run_id;
state_updater = new SqlStateUpdater(this->sql_schema, this->run_id);
state_updater->changeState(STARTED);
this->sql_connection->perform(*state_updater);
}
/* -------------------------------------------------------------------------- */
void Pusher::endRun(){
state_updater->changeState(ENDED);
this->sql_connection->perform(*state_updater);
}
/* -------------------------------------------------------------------------- */
void Pusher::push(const int & value, const UInt& quantity_id, const UInt& step) {
this->scalar_integer.value = &value;
this->scalar_integer.run_id = this->run_id;
this->scalar_integer.quantity_id = quantity_id;
this->scalar_integer.step = step;
this->sql_connection->perform(SqlPusher<int, true>(this->scalar_integer));
}
/* -------------------------------------------------------------------------- */
void Pusher::push(const Real & value, const UInt& quantity_id, const UInt& step) {
this->scalar_real.value = &value;
this->scalar_real.run_id = this->run_id;
this->scalar_real.quantity_id = quantity_id;
this->scalar_real.step = step;
this->sql_connection->perform(SqlPusher<Real, true>(this->scalar_real));
}
/* -------------------------------------------------------------------------- */
void Pusher::push(const std::vector<Real> & values, const UInt& quantity_id, const UInt& step) {
this->vector_real.value = &values;
this->vector_real.run_id = this->run_id;
this->vector_real.quantity_id = quantity_id;
this->vector_real.step = step;
this->sql_connection->perform(SqlPusher<Real, false>(this->vector_real));
}
/* -------------------------------------------------------------------------- */
void Pusher::push(const std::vector<int> & values, const UInt& quantity_id, const UInt& step) {
this->vector_integer.value = &values;
this->vector_integer.run_id = this->run_id;
this->vector_integer.quantity_id = quantity_id;
this->vector_integer.step = step;
this->sql_connection->perform(SqlPusher<int, false>(this->vector_integer));
}
/* -------------------------------------------------------------------------- */
UInt Pusher::getQuantityID(const std::string & name) {
UInt return_id = 0;
std::stringstream command;
command << "SELECT id FROM " << this->quantities_tablename
<< " WHERE name = '" << pqxx::to_string(name) << "';";
pqxx::transaction<pqxx::serializable> transaction(*this->sql_connection);
try {
pqxx::result result= transaction.exec(command);
if (result.size() == 0) {
FATAL("There is no quantity named '" << name
<< " defined in the database. ( The failed query was " << command.str() << ")");
}
return_id = result.at(0).at(0).as<UInt>(UInt(0));
if (return_id == 0) {
FATAL("There appears to be a problem, query for quantity id of " << name);
}
} catch (std::runtime_error & e) {
FATAL("Connection failed with " << e.what());
} catch (std::exception & e) {
FATAL("exception occured when executing sql request: "<< command.str() << std::endl << e.what());
} catch (...) {
FATAL("Unknown error occured while trying to query quantity_id for " << name);
}
return return_id;
}
/* -------------------------------------------------------------------------- */
UInt Pusher::connection_counter = 0;
pqxx::connection * Pusher::sql_connection = NULL;
/* -------------------------------------------------------------------------- */
}
c4science · Help