DatabaseCampaign: abstract campain for interaction with MySQL Database
The DatabaseCampaign interacts with the MySQL tables that are created
by the import-trace and prune-trace tools. It does offer all
unfinished experiment pilots from the database to the
fail-clients. Those clients send back a (by the experiment) defined
protobuf message as a result. The custom protobuf message does have to
need the form:
import "DatabaseCampaignMessage.proto";
message ExperimentMsg {
required DatabaseCampaignMessage fsppilot = 1;
repeated group Result = 2 {
// custom fields
required int32 bitoffset = 1;
optional int32 result = 2;
}
}
The DatabaseCampaignMessage is the pilot identifier from the
database. For each of the repeated result entries a row in a table is
allocated. The structure of this table is constructed (by protobuf
reflection) from the description of the message. Each field in the
Result group becomes a column in the result table. For the given
example it would be:
CREATE TABLE result_ExperimentMessage(
pilot_id INT,
bitoffset INT NOT NULL,
result INT,
PRIMARY_KEY(pilot_id)
)
Change-Id: I28fb5488e739d4098b823b42426c5760331027f8
This commit is contained in:
Christian Dietrich
465
src/core/util/DatabaseProtobufAdapter.cc
Normal file
465
src/core/util/DatabaseProtobufAdapter.cc
Normal file
@ -0,0 +1,465 @@
|
||||
#include <iostream>
|
||||
#include <sstream>
|
||||
#include <assert.h>
|
||||
#include "DatabaseProtobufAdapter.hpp"
|
||||
#include "util/Logger.hpp"
|
||||
#include "util/StringJoiner.hpp"
|
||||
//FIXME
|
||||
#include "../experiments/dciao-kernelstructs/dciao_kernel.pb.h"
|
||||
|
||||
|
||||
static fail::Logger LOG("DatabaseProtobufAdapter", true);
|
||||
|
||||
|
||||
using namespace fail;
|
||||
using namespace google::protobuf;
|
||||
|
||||
DatabaseProtobufAdapter::TypeBridge::TypeBridge(const FieldDescriptor *desc)
|
||||
: desc(desc) {
|
||||
/* We get the value of the field option extension in_primary_key
|
||||
to determine which fields should be added additionally to the
|
||||
primary key, besides pilot_id */
|
||||
if (desc) { // top-level type bridge has no desc pointer
|
||||
const FieldOptions& field_options = desc->options();
|
||||
this->primary_key = field_options.GetExtension(sql_primary_key);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
std::string DatabaseProtobufAdapter::TypeBridge_enum::sql_type() {
|
||||
/* enum types are mapped onto SQL enum types, therefore we have to
|
||||
gather all the stringified names for the enum, values */
|
||||
const google::protobuf::EnumDescriptor *e = desc->enum_type();
|
||||
std::stringstream ss;
|
||||
ss << "ENUM (";
|
||||
for (int i = 0; i < e->value_count(); i++) {
|
||||
ss << "'" << e->value(i)->name() << "'";
|
||||
if (i != e->value_count() - 1)
|
||||
ss << ", ";
|
||||
}
|
||||
ss << ")";
|
||||
return ss.str();
|
||||
}
|
||||
|
||||
std::string DatabaseProtobufAdapter::TypeBridge_message::sql_create_stmt() {
|
||||
/* The create statement for a message TypeBridge just propagates
|
||||
the call onto the enclosed types (including inner
|
||||
TypeBridge_message objects */
|
||||
std::stringstream ss;
|
||||
for (std::vector<TypeBridge *>::iterator it = types.begin();
|
||||
it != types.end(); ++it) {
|
||||
TypeBridge *bridge = *it;
|
||||
ss << bridge->sql_create_stmt();
|
||||
if (it+1 != types.end())
|
||||
ss << ", ";
|
||||
}
|
||||
return ss.str();
|
||||
}
|
||||
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_int64::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_LONGLONG;
|
||||
bind->is_unsigned = 0;
|
||||
buffer = ref->GetInt64(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_uint64::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_LONGLONG;
|
||||
bind->is_unsigned = 1;
|
||||
buffer = ref->GetUInt64(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_int32::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_LONG;
|
||||
bind->is_unsigned = 0;
|
||||
buffer = ref->GetInt32(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_uint32::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_LONG;
|
||||
bind->is_unsigned = 1;
|
||||
buffer = ref->GetUInt32(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_float::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_FLOAT;
|
||||
buffer = ref->GetFloat(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_double::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_DOUBLE;
|
||||
buffer = ref->GetDouble(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_bool::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_TINY;
|
||||
bind->is_unsigned = 1;
|
||||
buffer = ref->GetBool(*msg, desc);
|
||||
bind->buffer = &buffer;
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_string::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_STRING;
|
||||
bind->buffer = (void *) ref->GetString(*msg, desc).c_str();
|
||||
bind->buffer_length = ref->GetString(*msg, desc).length();
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_enum::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
/* Handle the NULL case */
|
||||
if (insert_null(bind, msg)) return;
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_STRING;
|
||||
bind->buffer = (void *) ref->GetEnum(*msg, desc)->name().c_str();
|
||||
bind->buffer_length = ref->GetEnum(*msg, desc)->name().length();
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_message::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
std::stringstream ss;
|
||||
for (std::vector<TypeBridge *>::iterator it = types.begin();
|
||||
it != types.end(); ++it) {
|
||||
TypeBridge *bridge = *it;
|
||||
TypeBridge_message *msg_bridge = dynamic_cast<TypeBridge_message *> (bridge);
|
||||
if (msg_bridge != 0) {
|
||||
const Message *inner_msg = 0;
|
||||
if (msg_bridge->desc->is_repeated()) {
|
||||
std::vector<int> &selector = *top_level_msg()->selector;
|
||||
inner_msg = &ref->GetRepeatedMessage(*msg, msg_bridge->desc, selector[nesting_level+1]);
|
||||
} else {
|
||||
inner_msg = &ref->GetMessage(*msg, msg_bridge->desc);
|
||||
}
|
||||
msg_bridge->bind(bind, inner_msg);
|
||||
/* Increment bind pointer */
|
||||
bind = &bind[msg_bridge->field_count];
|
||||
} else {
|
||||
/* Bind the plain field and increment the binding pointer */
|
||||
bridge->bind(bind, msg);
|
||||
bind ++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_repeated::bind(MYSQL_BIND *bind, const google::protobuf::Message *msg) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
size_t count = ref->FieldSize(*msg,desc);
|
||||
if (count == 0) {
|
||||
bind->buffer_type = MYSQL_TYPE_NULL;
|
||||
return;
|
||||
}
|
||||
|
||||
if (buffer) delete[] buffer;
|
||||
buffer = new char[count * inner->element_size()];
|
||||
assert (inner->element_size() > 0);
|
||||
|
||||
char *p = buffer;
|
||||
for (unsigned i = 0; i < count; i++) {
|
||||
inner->copy_to(msg, i, p);
|
||||
p += inner->element_size();
|
||||
}
|
||||
|
||||
bind->buffer_type = MYSQL_TYPE_BLOB;
|
||||
bind->buffer = buffer;
|
||||
bind->buffer_length = count * inner->element_size();
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::TypeBridge_int32::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(int32_t *) p = ref->GetRepeatedInt32(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_uint32::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(uint32_t *) p = ref->GetRepeatedUInt32(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_int64::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(int64_t *) p = ref->GetRepeatedInt64(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_uint64::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(uint64_t *) p = ref->GetRepeatedUInt64(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_double::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(double *) p = ref->GetRepeatedDouble(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_float::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(float *) p = ref->GetRepeatedFloat(*msg, desc, i);
|
||||
}
|
||||
void DatabaseProtobufAdapter::TypeBridge_bool::copy_to(const google::protobuf::Message *msg, int i, void *p) {
|
||||
const google::protobuf::Reflection *ref = msg->GetReflection();
|
||||
*(char *) p = ref->GetRepeatedBool(*msg, desc, i) ? '1' : '0';
|
||||
}
|
||||
|
||||
void DatabaseProtobufAdapter::error_create_table() {
|
||||
std::cerr << "ERROR: Cannot create the result table from message description" << std::endl;
|
||||
std::cerr << " The form only from of messages is: [required DatabaseCampaignMessage, repeated group X {}]" << std::endl;
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
int DatabaseProtobufAdapter::TypeBridge_message::gatherTypes(StringJoiner &insert_stmt, StringJoiner &primary_key) {
|
||||
/* Clear old state */
|
||||
repeated_message_stack.clear();
|
||||
repeated_message_stack.push_back(this);
|
||||
for (std::vector<TypeBridge *>::iterator it = types.begin(); it != types.end(); ++it)
|
||||
delete *it;
|
||||
types.clear();
|
||||
|
||||
size_t count = msg_type->field_count();
|
||||
field_count = 0;
|
||||
|
||||
for (unsigned i = 0; i < count; i++) {
|
||||
const FieldDescriptor *field = msg_type->field(i);
|
||||
assert(field != 0);
|
||||
|
||||
TypeBridge *bridge = 0;
|
||||
TypeBridge_message *msg_bridge;
|
||||
bool can_be_repeated = true; // default value
|
||||
|
||||
// For repeated messages
|
||||
TypeBridge_message *top_level_msg;
|
||||
|
||||
const FieldOptions& field_options = field->options();
|
||||
if (field_options.GetExtension(sql_ignore)) {
|
||||
// Field should be ignored
|
||||
continue;
|
||||
}
|
||||
|
||||
switch (field->cpp_type()) {
|
||||
case FieldDescriptor::CPPTYPE_INT32:
|
||||
bridge = new TypeBridge_int32(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_UINT32:
|
||||
bridge = new TypeBridge_uint32(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_INT64:
|
||||
bridge = new TypeBridge_int64(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_UINT64:
|
||||
bridge = new TypeBridge_uint64(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_DOUBLE:
|
||||
bridge = new TypeBridge_double(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_FLOAT:
|
||||
bridge = new TypeBridge_float(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_BOOL:
|
||||
bridge = new TypeBridge_bool(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_ENUM:
|
||||
can_be_repeated = false;
|
||||
bridge = new TypeBridge_enum(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_STRING:
|
||||
can_be_repeated = false;
|
||||
bridge = new TypeBridge_string(field);
|
||||
break;
|
||||
case FieldDescriptor::CPPTYPE_MESSAGE:
|
||||
if (field->is_repeated()) {
|
||||
top_level_msg = this->top_level_msg();
|
||||
/* Here we check wether we are on the repeated path
|
||||
from the root, when we are a repeated entry is ok.
|
||||
*/
|
||||
if (!(this == top_level_msg->repeated_message_stack.back())) {
|
||||
LOG << "Cannot handle repeated inner message in two different paths: " << field->name() << std::endl;
|
||||
exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
if (field->is_optional()) {
|
||||
LOG << "Cannot handle optional inner message: " << field->name() << std::endl;
|
||||
exit(-1);
|
||||
}
|
||||
|
||||
msg_bridge = new TypeBridge_message(field, field->message_type(), this);
|
||||
bridge = msg_bridge;
|
||||
types.push_back(bridge);
|
||||
if (field->is_repeated()) {
|
||||
while (1) {
|
||||
TypeBridge_message *back = top_level_msg->repeated_message_stack.back();
|
||||
if (back == msg_bridge)
|
||||
break;
|
||||
TypeBridge_message *p = msg_bridge;
|
||||
while (back != 0 && p->parent != back && p->parent != 0) {
|
||||
p = p->parent;
|
||||
}
|
||||
top_level_msg->repeated_message_stack.push_back(p);
|
||||
}
|
||||
}
|
||||
|
||||
field_count += msg_bridge->gatherTypes(insert_stmt, primary_key);
|
||||
|
||||
/* Do not the normal field adding process */
|
||||
continue;
|
||||
|
||||
default:
|
||||
std::cerr << "unsupported field: " << field->name() << std::endl;
|
||||
exit(-1);
|
||||
break;
|
||||
}
|
||||
|
||||
if (field->is_repeated() && ! can_be_repeated) {
|
||||
LOG << "Cannot handle repeated field: " << field->name() << std::endl;
|
||||
exit(-1);
|
||||
}
|
||||
if (field->is_repeated()) {
|
||||
bridge = new TypeBridge_repeated(field, bridge);
|
||||
}
|
||||
|
||||
types.push_back(bridge);
|
||||
field_count ++;
|
||||
insert_stmt.push_back(field->name());
|
||||
if (bridge->primary_key)
|
||||
primary_key.push_back(field->name());
|
||||
}
|
||||
|
||||
return field_count;
|
||||
}
|
||||
|
||||
|
||||
void DatabaseProtobufAdapter::create_table(const Descriptor *toplevel_desc) {
|
||||
assert (toplevel_desc != 0);
|
||||
|
||||
std::stringstream create_table_stmt, insert_stmt;
|
||||
StringJoiner insert_join, primary_join, question_marks;
|
||||
|
||||
result_table_name = "result_" + toplevel_desc->name();
|
||||
|
||||
/* Fill our top level dummy type bridge with the fields from the
|
||||
example message */
|
||||
top_level_msg.msg_type = toplevel_desc;
|
||||
int fields = top_level_msg.gatherTypes(insert_join, primary_join);
|
||||
for (int i = 0; i < fields; i++)
|
||||
question_marks.push_back("?");
|
||||
|
||||
create_table_stmt << "CREATE TABLE IF NOT EXISTS " << result_table_name << "(";
|
||||
create_table_stmt << top_level_msg.sql_create_stmt() << ", PRIMARY KEY(" << primary_join.join(", ") << "))";
|
||||
|
||||
insert_stmt << "INSERT INTO " << result_table_name << "(" << insert_join.join(",");
|
||||
insert_stmt << ") VALUES (" << question_marks.join(",") << ")";
|
||||
|
||||
// Create the Table
|
||||
db->query(create_table_stmt.str().c_str());
|
||||
|
||||
|
||||
// Prepare the insert statement
|
||||
stmt = mysql_stmt_init(db->getHandle());
|
||||
if (mysql_stmt_prepare(stmt, insert_stmt.str().c_str(), insert_stmt.str().length())) {
|
||||
LOG << "query '" << insert_stmt.str() << "' failed: " << mysql_error(db->getHandle()) << std::endl;
|
||||
exit(-1);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
int DatabaseProtobufAdapter::field_size_at_pos(const Message *msg, std::vector<int> selector, int pos) {
|
||||
std::vector< TypeBridge_message *>::iterator it;
|
||||
if (top_level_msg.repeated_message_stack.size() <= 1)
|
||||
return 1;
|
||||
|
||||
int i = 1;
|
||||
for (it = top_level_msg.repeated_message_stack.begin() + 1;
|
||||
i != pos && it != top_level_msg.repeated_message_stack.end(); ++it) {
|
||||
|
||||
TypeBridge_message *bridge = *it;
|
||||
const Reflection *ref = msg->GetReflection();
|
||||
if (bridge->desc && bridge->desc->is_repeated()) {
|
||||
msg = &ref->GetRepeatedMessage(*msg, bridge->desc, selector[i]);
|
||||
} else {
|
||||
assert(selector[i] == 0);
|
||||
msg = &ref->GetMessage(*msg, bridge->desc);
|
||||
}
|
||||
i++;
|
||||
}
|
||||
const Reflection *ref = msg->GetReflection();
|
||||
return ref->FieldSize(*msg, top_level_msg.repeated_message_stack[i]->desc);
|
||||
}
|
||||
|
||||
bool DatabaseProtobufAdapter::insert_row(const google::protobuf::Message *msg) {
|
||||
const Reflection *ref = msg->GetReflection();
|
||||
const Descriptor *d = msg->GetDescriptor();
|
||||
assert (d != 0 && ref != 0);
|
||||
|
||||
MYSQL_BIND *bind = new MYSQL_BIND[top_level_msg.field_count];
|
||||
|
||||
/* We determine how many columns should be produced */
|
||||
std::vector<int> selector (top_level_msg.repeated_message_stack.size());
|
||||
|
||||
while (true) {
|
||||
// INSERT WITH SELECTOR
|
||||
top_level_msg.selector = &selector;
|
||||
|
||||
// Use the top_level_msg TypeBridge to bind all parameters
|
||||
// into the MYSQL_BIND structure
|
||||
memset(bind, 0, sizeof(*bind) * (top_level_msg.field_count));
|
||||
top_level_msg.bind(bind, msg);
|
||||
|
||||
// Insert the binded row
|
||||
if (mysql_stmt_bind_param(stmt, bind)) {
|
||||
LOG << "mysql_stmt_bind_param() failed: " << mysql_stmt_error(stmt) << std::endl;
|
||||
delete[] bind;
|
||||
return false;
|
||||
}
|
||||
|
||||
if (mysql_stmt_execute(stmt)) {
|
||||
LOG << "mysql_stmt_execute() failed: " << mysql_stmt_error(stmt) << std::endl;
|
||||
delete[] bind;
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
/* Increment the selector */
|
||||
unsigned i = selector.size() - 1;
|
||||
selector[i] ++;
|
||||
|
||||
while (i > 0 && field_size_at_pos(msg, selector, i) <= selector[i]) {
|
||||
selector[i] = 0;
|
||||
i--;
|
||||
selector[i] ++;
|
||||
}
|
||||
if (i == 0) break;
|
||||
}
|
||||
|
||||
delete[] bind;
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user