Skip to content

U
uniset2
Commit 8db0746f authored by Pavel Vainerman's avatar Pavel Vainerman
Browse files
Options

(MySQL): реализовал отдельные insert-буферы для каждой таблицы,

чтобы данные вставлялись одним запросом..
parent 8b84cbd0
Show whitespace changes
Inline Side-by-side
Showing with 164 additions and 72 deletions
extensions/DBServer-MySQL/DBInterface.cc
View file @ 8db0746f
......@@ -47,7 +47,7 @@ DBInterface::~DBInterface()
// -----------------------------------------------------------------------------------------
bool DBInterface::connect( const string& host, const string& user, const string& pswd, const string& dbname)
{
if( !mysql_real_connect(mysql,host.c_str(), user.c_str(),pswd.c_str(),dbname.c_str(),0,NULL,CLIENT_MULTI_STATEMENTS) )
if( !mysql_real_connect(mysql,host.c_str(), user.c_str(),pswd.c_str(),dbname.c_str(),0,NULL,0) ) // CLIENT_MULTI_STATEMENTS) )
{
cout << error() << endl;
mysql_close(mysql);
......@@ -55,7 +55,6 @@ bool DBInterface::connect( const string& host, const string& user, const string&
return false;
}
// mysql_set_server_option(mysql,MYSQL_OPTION_MULTI_STATEMENTS_ON);
connected = true;
return true;
}
......@@ -101,6 +100,7 @@ bool DBInterface::query( const char* q, bool noLastQ )
result = mysql_store_result(mysql); // _use_result - некорректно работает с _num_rows
#if 0
// Если при соединении используется CLIENT_MULTI_STATEMENTS
// то необходимо вынуть все результаты..
while( mysql_more_results(mysql) )
......@@ -109,6 +109,7 @@ bool DBInterface::query( const char* q, bool noLastQ )
if( mysql_next_result(mysql) >=0 )
mysql_store_result(mysql);
}
#endif
if( numRows()==0 )
{
......
extensions/DBServer-MySQL/DBServer_MySQL.cc
View file @ 8db0746f
......@@ -45,9 +45,9 @@ DBServer_MySQL::DBServer_MySQL(ObjectId id):
ReconnectTime(180000),
connect_ok(false),
activate(true),
qbufSize(200),
lastRemove(false),
qbufByteSize(0)
flushBufferTime(120000),
qbufSize(200)
{
if( getId() == DefaultObjectId )
{
......@@ -64,8 +64,6 @@ DBServer_MySQL::DBServer_MySQL():
ReconnectTime(180000),
connect_ok(false),
activate(true),
qbufSize(200),
qbufByteSize(0),
lastRemove(false)
{
// init();
......@@ -202,28 +200,36 @@ void DBServer_MySQL::parse( UniSetTypes::ConfirmMessage* cem )
}
}
//--------------------------------------------------------------------------------------------
bool DBServer_MySQL::writeToBase( const string& query )
bool DBServer_MySQL::insertToBuffer( int key, const string& values )
{
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(writeToBase): " << query << endl;
// Складываем в очередь
bool fb = false;
DBTableMap::iterator it = tblMap.find(key);
if( it == tblMap.end() )
{
uniset_mutex_lock l(mqbuf,2000);
qbuf.push(query);
fb = ( qbuf.size() >= qbufSize );
qbufByteSize += query.size();
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(insertToBuffer): table key=" << key << " not found.." << endl;
return false;
}
if( fb )
flushBuffer();
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(insertToBuffer): " << it->second.tblname << " val: " << values << endl;
return true;
TableInfo* ti = &it->second;
ti->qbuf.push(values);
ti->qbufByteSize += values.size();
#if 0
if( ti->qbuf.size() >= ti->qbufSize )
flushTableBuffer(it);
return true;
}
//--------------------------------------------------------------------------------------------
bool DBServer_MySQL::writeToBase( const string& query )
{
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(writeToBase): " << query << endl;
// cout << "DBServer_MySQL: " << query << endl;
if( !db || !connect_ok )
{
uniset_mutex_lock l(mqbuf,200);
......@@ -237,16 +243,16 @@ bool DBServer_MySQL::writeToBase( const string& query )
qlost = qbuf.front();
qbuf.pop();
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(writeToBase): DB not connected! buffer(" << qbufSize
<< ") overflow! lost query: " << qlost << endl;
}
return false;
}
// На всякий скидываем очередь
flushBuffer();
flushQBuffer();
// А теперь собственно запрос..
db->query(query);
......@@ -263,46 +269,83 @@ bool DBServer_MySQL::writeToBase( const string& query )
}
return false;
#endif
}
//-------------------------------------------------------------------------------------------
void DBServer_MySQL::flushQBuffer()
{
uniset_mutex_lock l(mqbuf);
// Сперва пробуем очистить всё что накопилось в очереди до этого...
while( !qbuf.empty() )
{
db->query( qbuf.front() );
// Дело в том что на INSERT И UPDATE запросы
// db->query() может возвращать false и надо самому
// отдельно проверять действительно ли произошла ошибка
// см. MySQLInterface::query.
string err(db->error());
if( !err.empty() && unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(writeToBase): error: " << err <<
" lost query: " << qbuf.front() << endl;
qbuf.pop();
}
}
//--------------------------------------------------------------------------------------------
void DBServer_MySQL::flushBuffer()
void DBServer_MySQL::flushTableBuffer( DBTableMap::iterator& it )
{
uniset_mutex_lock l(mqbuf,400);
if( it == tblMap.end() )
return;
TableInfo* ti(&it->second);
if( qbuf.empty() )
if( ti->qbuf.empty() )
return;
if( unideb.debugging(DBLEVEL) )
unideb[DBLEVEL] << myname << "(flushBuffer): qbufSize=" << qbufSize << " qbufByteSize=" << qbufByteSize << endl;
unideb[DBLEVEL] << myname << "(flushBuffer): '" << ti->tblname << "' qbufSize=" << ti->qbufSize << " qbufByteSize=" << ti->qbufByteSize << endl;
// buffer чтобы записать всё одним запросом..
char* cbuf = new char[ qbufByteSize + 1 ];
int i= 0;
// "ti->qbuf.size() - 1" - это количество добавляемых запятых..
unsigned int cbufSize = ti->insHeader.size() + ti->qbuf.size() - 1 + ti->qbufByteSize + 1;
char* cbuf = new char[cbufSize];
unsigned int i= 0;
if( !cbuf )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(flushBuffer): Can`t allocate memory for buffer (sz=" << qbufByteSize + 1 << endl;
unideb[Debug::CRIT] << myname << "(flushBuffer): Can`t allocate memory for buffer (sz=" << ti->qbufByteSize + 1 << endl;
return;
}
while( !qbuf.empty() )
std::memcpy( &(cbuf[i]), ti->insHeader.c_str(), ti->insHeader.size() );
i+=ti->insHeader.size();
int n = 1;
while( !ti->qbuf.empty() )
{
std::string q(qbuf.front());
std::string q(ti->qbuf.front());
int sz = q.size();
if( i + sz > qbufByteSize )
unsigned int sz = q.size();
if( i + sz > cbufSize )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(flushBuffer): BUFFER OVERFLOW bytesize=" << qbufByteSize + 1 << " i=" << i << " sz=" << sz<< endl;
unideb[Debug::CRIT] << myname << "(flushBuffer): BUFFER OVERFLOW bytesize=" << cbufSize << " i=" << i << " sz=" << sz<< endl;
break;
}
if( n > 1 ) // если запись не первая.. добавляем ',' в конец предыдущей записи
cbuf[i++] = ',';
std::memcpy( &(cbuf[i]), q.c_str(), sz );
i += sz;
qbuf.pop();
ti->qbuf.pop();
n++;
}
// последняя точка с запятой, вроде не нужна (по документации и примерам из MySQL)..
......@@ -311,14 +354,14 @@ void DBServer_MySQL::flushBuffer()
else
cbuf[i] = '\0';
if( i < qbufByteSize )
if( i < cbufSize )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(writeToBase): BAD qbufByteSize=" << qbufByteSize << " LOST RECORDS..(i=" << i << ")" << endl;
unideb[Debug::CRIT] << myname << "(writeToBase): BAD qbufByteSize=" << ti->qbufByteSize << " LOST RECORDS..(i=" << i << ")" << endl;
qbufByteSize = 0;
while( !qbuf.empty() )
qbuf.pop();
ti->qbufByteSize = 0;
while( !ti->qbuf.empty() )
ti->qbuf.pop();
}
try
......@@ -345,7 +388,7 @@ void DBServer_MySQL::flushBuffer()
}
delete[] cbuf;
qbufByteSize = 0;
ti->qbufByteSize = 0;
}
//--------------------------------------------------------------------------------------------
void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
......@@ -363,25 +406,19 @@ void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
// см. DBTABLE AnalogSensors, DigitalSensors
ostringstream data;
data << "INSERT INTO " << tblName(si->type)
<< "(date, time, time_usec, sensor_id, value, node) VALUES( '"
data << "('"
// Поля таблицы
<< ui.dateToString(si->sm_tv_sec,"-") << "','" // date
<< ui.timeToString(si->sm_tv_sec,":") << "','" // time
<< si->sm_tv_usec << "','" // time_usec
<< si->id << "','" // sensor_id
<< val << "','" // value
<< si->node << "');"; // node
<< si->node << "')"; // node
if( unideb.debugging(DBLEVEL) )
unideb[DBLEVEL] << myname << "(insert_main_history): " << data.str() << endl;
if( !writeToBase(data.str()) )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(insert) sensor msg error: "<< db->error() << endl;
db->freeResult();
}
insertToBuffer(si->type, data.str());
}
catch( Exception& ex )
{
......@@ -427,12 +464,22 @@ void DBServer_MySQL::init_dbserver()
string user(conf->getProp(node,"dbuser"));
string password(conf->getProp(node,"dbpass"));
tblMap[UniSetTypes::Message::SensorInfo] = "main_history";
tblMap[UniSetTypes::Message::Confirm] = "main_history";
PingTime = conf->getIntProp(node,"pingTime");
ReconnectTime = conf->getIntProp(node,"reconnectTime");
qbufSize = conf->getArgPInt("--dbserver-buffer-size",it.getProp("bufferSize"),200);
flushBufferTime = conf->getArgPInt("--dbserver-flush-buffer-time",it.getProp("flushBufferTime"),120000);
qbufSize = conf->getArgPInt("--dbserver-lost-buffer-size",it.getProp("lostBufferSize"),200);
int insbufSize = conf->getArgPInt("--dbserver-insert-buffer-size",it.getProp("insertBufferSize"),100000);
TableInfo ti;
ti.tblname = "main_history";
ti.qbufSize = insbufSize;
ti.qbufByteSize = 0;
ti.insHeader = "INSERT INTO main_history(date, time, time_usec, sensor_id, value, node) VALUES";
tblMap[UniSetTypes::Message::SensorInfo] = ti;
tblMap[UniSetTypes::Message::Confirm] = ti;
if( findArgParam("--dbserver-buffer-last-remove",conf->getArgc(),conf->getArgv()) != -1 )
lastRemove = true;
......@@ -459,6 +506,7 @@ void DBServer_MySQL::init_dbserver()
<< db->error() << endl;
// throw Exception( string(myname+"(init): не смогли создать соединение с БД "+db->error()) );
askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime);
askTimer(DBServer_MySQL::FlushBufferTimer,0);
}
else
{
......@@ -467,10 +515,12 @@ void DBServer_MySQL::init_dbserver()
connect_ok = true;
askTimer(DBServer_MySQL::ReconnectTimer,0);
askTimer(DBServer_MySQL::PingTimer,PingTime);
askTimer(DBServer_MySQL::FlushBufferTimer,flushBufferTime);
// createTables(db);
initDB(db);
initDBTableMap(tblMap);
flushBuffer();
flushQBuffer();
}
}
//--------------------------------------------------------------------------------------------
......@@ -511,6 +561,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
connect_ok = false;
askTimer(DBServer_MySQL::PingTimer,0);
askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime);
askTimer(DBServer_MySQL::FlushBufferTimer,0);
}
else
{
......@@ -532,6 +583,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
connect_ok = true;
askTimer(DBServer_MySQL::ReconnectTimer,0);
askTimer(DBServer_MySQL::PingTimer,PingTime);
askTimer(DBServer_MySQL::FlushBufferTimer,flushBufferTime);
}
connect_ok = false;
if( unideb.debugging(Debug::WARN) )
......@@ -542,6 +594,13 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
}
break;
case FlushBufferTimer:
{
for( DBTableMap::iterator it=tblMap.begin(); it!=tblMap.end(); ++it )
flushTableBuffer(it);
}
break;
default:
if( unideb.debugging(Debug::WARN) )
unideb[Debug::WARN] << myname << "(timerInfo): Unknown TimerID=" << tm->id << endl;
......
extensions/DBServer-MySQL/DBServer_MySQL.h
View file @ 8db0746f
......@@ -37,7 +37,7 @@
- \ref sec_DBS_Conf
- \ref sec_DBS_Tables
- \ref sec_DBS_Buffer
- \ref sec_DBS_InsertBuffer
\section sec_DBS_Comm Общее описание работы DBServer_MySQL
Сервис предназначен для работы с БД MySQL. В его задачи входит
......@@ -73,8 +73,8 @@
\section sec_DBS_Buffer Защита от потери данных
Для того, чтобы на момент отсутствия связи с БД данные по возможности не потерялись,
сделан "кольцевой" буфер. Размер которго можно регулировать параметром "--dbserver-buffer-size"
или параметром \b bufferSize=".." в конфигурационном файле секции "<LocalDBSErver...>".
сделан "кольцевой" буфер. Размер которго можно регулировать параметром "--dbserver-lost-buffer-size"
или параметром \b lostBufferSize=".." в конфигурационном файле секции "<LocalDBServer...>".
Механизм построен на том, что если связь с mysql сервером отсутствует или пропала,
то сообщения помещаются в колевой буфер, который "опустошается" как только она восстановится.
......@@ -82,6 +82,20 @@
более ранние сообщения. Эту логику можно сменить, если указать параметр "--dbserver-buffer-last-remove"
или \b bufferLastRemove="1", то терятся будут сообщения добавляемые в конец.
\section sec_DBS_InsertBuffer Добавление данных (буферизованный INSERT)
Для оптимизации работы с БД, для каждой таблицы сделан отдельный буфер INSERT-запросов (см. TableInfo).
Размер буфера (в количестве сообщений) задаётся параметром "--dbserver-insert-buffer-size"
или параметром \b insertBufferSize=".." в конфигурационном файле секции "<LocalDBServer...>".
Данные буферы (для каждой таблицы!) скидываются в БД в двух случаях:
-# если буфер заполнился (т.е. по заполнению)
-# если истекло заданное время (т.е. принудительно). Время скидывания insert-буферов задаётся параметром "--dbserver-flush-buffer-time" или параметром \b flushBufferTime=".." в конфигурационном файле секции "<LocalDBServer...>".
\note Следует иметь ввиду, что размер буфера (и период принудительного скидывания) потенциально влияют на то, сколько данных
будет ПОТЕРЯНО если произодёт отключение питания или вылет процесса. С другой стороны, постоянная запись в БД без промежуточного клэширования (буферизации) приводит к большой нагрузке на БД (диск и т.п.).
\note Для примерного расчёта размера буфера, можно считать, что размер одной записи в буфере ~ 50 байт.
\note Следует иметь ввиду, что итоговый "INSERT" (когда буфер будет скидываться в БД) не должен по размеру превышать \b max_allowed_packet, которая задаётся в настройках MySQL (см. документацию).
\section sec_DBS_Tables Таблицы MySQL
К основным таблицам относятся следующие:
\code
......@@ -142,7 +156,19 @@ class DBServer_MySQL:
static const Debug::type DBLogInfoLevel = Debug::LEVEL9;
protected:
typedef std::map<int, std::string> DBTableMap;
typedef std::queue<std::string> QueryBuffer;
struct TableInfo
{
QueryBuffer qbuf; // буфер для данных
unsigned int qbufSize; // размер буфера сообщений.
unsigned int qbufByteSize; // размер буфера в байтах
std::string tblname; // название таблицы
std::string insHeader; // заголовок запроса (например: INSERT INTO tblname(date, time, time_usec, sensor_id, value, node) VALUES( '"
};
typedef std::map<int, TableInfo> DBTableMap;
virtual void initDB(DBInterface *db){};
virtual void initDBTableMap(DBTableMap& tblMap){};
......@@ -156,13 +182,18 @@ class DBServer_MySQL:
virtual void parse( UniSetTypes::DBMessage* dbmsg );
virtual void parse( UniSetTypes::ConfirmMessage* cmsg );
/*! запись в БД */
bool writeToBase( const string& query );
/*! поместить в буфер для INSERT запросов к таблице 'key' */
bool insertToBuffer( int key, const string& values );
virtual void init_dbserver();
void createTables( DBInterface* db );
inline const char* tblName(int key)
inline const char* tblName( int key )
{
return tblMap[key].c_str();
return tblMap[key].tblname.c_str();
}
enum Timers
......@@ -173,7 +204,6 @@ class DBServer_MySQL:
lastNumberOfTimer
};
DBInterface *db;
int PingTime;
int ReconnectTime;
......@@ -181,19 +211,21 @@ class DBServer_MySQL:
bool activate;
typedef std::queue<std::string> QueryBuffer;
QueryBuffer qbuf;
unsigned int qbufSize; // размер буфера сообщений.
bool lastRemove;
void flushBuffer();
UniSetTypes::uniset_mutex mqbuf;
int qbufByteSize; // размер буфера в байтах
int flushBufferTime;
/*! сброс insert-буфера для конкретной таблицы */
void flushTableBuffer( DBTableMap::iterator& i );
void flushQBuffer();
private:
DBTableMap tblMap;
DBTableMap tblMap;
UniSetTypes::uniset_mutex mqbuf;
QueryBuffer qbuf; // юуфер для запросов, пока нет связи с БД
unsigned int qbufSize;
};
//------------------------------------------------------------------------------------------
#endif
extensions/DBServer-MySQL/start_fg.sh
View file @ 8db0746f
......@@ -4,5 +4,5 @@ ulimit -Sc 1000000
uniset-start.sh -f ./uniset-mysql-dbserver --confile test.xml --name DBServer1 \
--unideb-add-levels info,crit,warn,level9,system \
--dbserver-buffer-size 50
--dbserver-buffer-size 100
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment