Skip to content

U
uniset2
Commit 5e90512f authored by Pavel Vainerman's avatar Pavel Vainerman
Browse files
Options

Merge branch 'mysql-buffer'

parents 60282166 8db0746f
Hide whitespace changes
Inline Side-by-side
Showing with 247 additions and 58 deletions
extensions/DBServer-MySQL/DBInterface.cc
View file @ 5e90512f
...@@ -47,13 +47,14 @@ DBInterface::~DBInterface() ...@@ -47,13 +47,14 @@ DBInterface::~DBInterface()
// ----------------------------------------------------------------------------------------- // -----------------------------------------------------------------------------------------
bool DBInterface::connect( const string& host, const string& user, const string& pswd, const string& dbname) 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,0)) 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; cout << error() << endl;
mysql_close(mysql); mysql_close(mysql);
connected = false; connected = false;
return false; return false;
} }
connected = true; connected = true;
return true; return true;
} }
...@@ -78,28 +79,45 @@ bool DBInterface::insert( const string& q ) ...@@ -78,28 +79,45 @@ bool DBInterface::insert( const string& q )
queryok=true; queryok=true;
return true; return true;
} }
// ----------------------------------------------------------------------------------------- // -----------------------------------------------------------------------------------------
bool DBInterface::query( const string& q ) bool DBInterface::query( const string& q )
{ {
return query( q.c_str(), true );
}
// -----------------------------------------------------------------------------------------
bool DBInterface::query( const char* q, bool noLastQ )
{
if( !mysql ) if( !mysql )
return false; return false;
if( mysql_query(mysql,q.c_str()) ) if( mysql_query(mysql,q) )
{ {
queryok=false; queryok=false;
return false; return false;
} }
lastQ = ( noLastQ ) ? q : "";
lastQ = q;
result = mysql_store_result(mysql); // _use_result - некорректно работает с _num_rows result = mysql_store_result(mysql); // _use_result - некорректно работает с _num_rows
#if 0
// Если при соединении используется CLIENT_MULTI_STATEMENTS
// то необходимо вынуть все результаты..
while( mysql_more_results(mysql) )
{
cerr << "**** store result..." << endl;
if( mysql_next_result(mysql) >=0 )
mysql_store_result(mysql);
}
#endif
if( numRows()==0 ) if( numRows()==0 )
{ {
queryok=false; queryok=false;
return false; return false;
} }
queryok=true; queryok=true;
return true; return true;
} }
// ----------------------------------------------------------------------------------------- // -----------------------------------------------------------------------------------------
......
extensions/DBServer-MySQL/DBInterface.h
View file @ 5e90512f
...@@ -47,6 +47,8 @@ class DBInterface ...@@ -47,6 +47,8 @@ class DBInterface
bool close(); bool close();
bool query(const std::string& q); bool query(const std::string& q);
bool query( const char* q, bool noLastQ=false ); // noLastQ - не запоминать запрос
const std::string lastQuery(); const std::string lastQuery();
bool insert(const std::string& q); bool insert(const std::string& q);
......
extensions/DBServer-MySQL/DBServer_MySQL.cc
View file @ 5e90512f
...@@ -45,8 +45,9 @@ DBServer_MySQL::DBServer_MySQL(ObjectId id): ...@@ -45,8 +45,9 @@ DBServer_MySQL::DBServer_MySQL(ObjectId id):
ReconnectTime(180000), ReconnectTime(180000),
connect_ok(false), connect_ok(false),
activate(true), activate(true),
qbufSize(200), lastRemove(false),
lastRemove(false) flushBufferTime(120000),
qbufSize(200)
{ {
if( getId() == DefaultObjectId ) if( getId() == DefaultObjectId )
{ {
...@@ -63,7 +64,6 @@ DBServer_MySQL::DBServer_MySQL(): ...@@ -63,7 +64,6 @@ DBServer_MySQL::DBServer_MySQL():
ReconnectTime(180000), ReconnectTime(180000),
connect_ok(false), connect_ok(false),
activate(true), activate(true),
qbufSize(200),
lastRemove(false) lastRemove(false)
{ {
// init(); // init();
...@@ -200,11 +200,36 @@ void DBServer_MySQL::parse( UniSetTypes::ConfirmMessage* cem ) ...@@ -200,11 +200,36 @@ void DBServer_MySQL::parse( UniSetTypes::ConfirmMessage* cem )
} }
} }
//-------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------
bool DBServer_MySQL::insertToBuffer( int key, const string& values )
{
// Складываем в очередь
DBTableMap::iterator it = tblMap.find(key);
if( it == tblMap.end() )
{
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(insertToBuffer): table key=" << key << " not found.." << endl;
return false;
}
if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(insertToBuffer): " << it->second.tblname << " val: " << values << endl;
TableInfo* ti = &it->second;
ti->qbuf.push(values);
ti->qbufByteSize += values.size();
if( ti->qbuf.size() >= ti->qbufSize )
flushTableBuffer(it);
return true;
}
//--------------------------------------------------------------------------------------------
bool DBServer_MySQL::writeToBase( const string& query ) bool DBServer_MySQL::writeToBase( const string& query )
{ {
if( unideb.debugging(DBLogInfoLevel) ) if( unideb.debugging(DBLogInfoLevel) )
unideb[DBLogInfoLevel] << myname << "(writeToBase): " << query << endl; unideb[DBLogInfoLevel] << myname << "(writeToBase): " << query << endl;
// cout << "DBServer_MySQL: " << query << endl;
if( !db || !connect_ok ) if( !db || !connect_ok )
{ {
uniset_mutex_lock l(mqbuf,200); uniset_mutex_lock l(mqbuf,200);
...@@ -218,16 +243,16 @@ bool DBServer_MySQL::writeToBase( const string& query ) ...@@ -218,16 +243,16 @@ bool DBServer_MySQL::writeToBase( const string& query )
qlost = qbuf.front(); qlost = qbuf.front();
qbuf.pop(); qbuf.pop();
if( unideb.debugging(Debug::CRIT) ) if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(writeToBase): DB not connected! buffer(" << qbufSize unideb[Debug::CRIT] << myname << "(writeToBase): DB not connected! buffer(" << qbufSize
<< ") overflow! lost query: " << qlost << endl; << ") overflow! lost query: " << qlost << endl;
} }
return false; return false;
} }
// На всякий скидываем очередь // На всякий скидываем очередь
flushBuffer(); flushQBuffer();
// А теперь собственно запрос.. // А теперь собственно запрос..
db->query(query); db->query(query);
...@@ -245,31 +270,125 @@ bool DBServer_MySQL::writeToBase( const string& query ) ...@@ -245,31 +270,125 @@ bool DBServer_MySQL::writeToBase( const string& query )
return false; return false;
} }
//-------------------------------------------------------------------------------------------
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( ti->qbuf.empty() )
return;
if( unideb.debugging(DBLEVEL) )
unideb[DBLEVEL] << myname << "(flushBuffer): '" << ti->tblname << "' qbufSize=" << ti->qbufSize << " qbufByteSize=" << ti->qbufByteSize << endl;
// Сперва пробуем очистить всё что накопилось в очереди до этого... // buffer чтобы записать всё одним запросом..
while( !qbuf.empty() ) // "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 )
{ {
db->query( qbuf.front() ); if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(flushBuffer): Can`t allocate memory for buffer (sz=" << ti->qbufByteSize + 1 << endl;
// Дело в том что на INSERT И UPDATE запросы return;
// db->query() может возвращать false и надо самому }
// отдельно проверять действительно ли произошла ошибка
// см. DBInterface::query. std::memcpy( &(cbuf[i]), ti->insHeader.c_str(), ti->insHeader.size() );
string err(db->error()); i+=ti->insHeader.size();
if( err.empty() )
db->freeResult(); int n = 1;
else if( unideb.debugging(Debug::CRIT) ) while( !ti->qbuf.empty() )
{
std::string q(ti->qbuf.front());
unsigned int sz = q.size();
if( i + sz > cbufSize )
{ {
unideb[Debug::CRIT] << myname << "(writeToBase): error: " << err << if( unideb.debugging(Debug::CRIT) )
" lost query: " << qbuf.front() << endl; unideb[Debug::CRIT] << myname << "(flushBuffer): BUFFER OVERFLOW bytesize=" << cbufSize << " i=" << i << " sz=" << sz<< endl;
break;
} }
qbuf.pop(); if( n > 1 ) // если запись не первая.. добавляем ',' в конец предыдущей записи
cbuf[i++] = ',';
std::memcpy( &(cbuf[i]), q.c_str(), sz );
i += sz;
ti->qbuf.pop();
n++;
}
// последняя точка с запятой, вроде не нужна (по документации и примерам из MySQL)..
if( cbuf[i-1] == ';' )
cbuf[i-1] = '\0';
else
cbuf[i] = '\0';
if( i < cbufSize )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(writeToBase): BAD qbufByteSize=" << ti->qbufByteSize << " LOST RECORDS..(i=" << i << ")" << endl;
ti->qbufByteSize = 0;
while( !ti->qbuf.empty() )
ti->qbuf.pop();
}
try
{
if( i > 0 )
{
db->query( (const char*)cbuf, false );
// Дело в том что на INSERT И UPDATE запросы
// db->query() может возвращать false и надо самому
// отдельно проверять действительно ли произошла ошибка
// см. DBInterface::queryh.
string err(db->error());
if( !err.empty() && unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(flushBuffer): error: " << err << endl;
db->freeResult();
}
}
catch( Exception& ex )
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(flushBuffer): " << ex << endl;
} }
delete[] cbuf;
ti->qbufByteSize = 0;
} }
//-------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------
void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si ) void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
...@@ -287,25 +406,19 @@ void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si ) ...@@ -287,25 +406,19 @@ void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
// см. DBTABLE AnalogSensors, DigitalSensors // см. DBTABLE AnalogSensors, DigitalSensors
ostringstream data; ostringstream data;
data << "INSERT INTO " << tblName(si->type) data << "('"
<< "(date, time, time_usec, sensor_id, value, node) VALUES( '"
// Поля таблицы // Поля таблицы
<< ui.dateToString(si->sm_tv_sec,"-") << "','" // date << ui.dateToString(si->sm_tv_sec,"-") << "','" // date
<< ui.timeToString(si->sm_tv_sec,":") << "','" // time << ui.timeToString(si->sm_tv_sec,":") << "','" // time
<< si->sm_tv_usec << "','" // time_usec << si->sm_tv_usec << "','" // time_usec
<< si->id << "','" // sensor_id << si->id << "','" // sensor_id
<< val << "','" // value << val << "','" // value
<< si->node << "')"; // node << si->node << "')"; // node
if( unideb.debugging(DBLEVEL) ) if( unideb.debugging(DBLEVEL) )
unideb[DBLEVEL] << myname << "(insert_main_history): " << data.str() << endl; unideb[DBLEVEL] << myname << "(insert_main_history): " << data.str() << endl;
if( !writeToBase(data.str()) ) insertToBuffer(si->type, data.str());
{
if( unideb.debugging(Debug::CRIT) )
unideb[Debug::CRIT] << myname << "(insert) sensor msg error: "<< db->error() << endl;
db->freeResult();
}
} }
catch( Exception& ex ) catch( Exception& ex )
{ {
...@@ -351,12 +464,22 @@ void DBServer_MySQL::init_dbserver() ...@@ -351,12 +464,22 @@ void DBServer_MySQL::init_dbserver()
string user(conf->getProp(node,"dbuser")); string user(conf->getProp(node,"dbuser"));
string password(conf->getProp(node,"dbpass")); string password(conf->getProp(node,"dbpass"));
tblMap[UniSetTypes::Message::SensorInfo] = "main_history";
tblMap[UniSetTypes::Message::Confirm] = "main_history";
PingTime = conf->getIntProp(node,"pingTime"); PingTime = conf->getIntProp(node,"pingTime");
ReconnectTime = conf->getIntProp(node,"reconnectTime"); 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 ) if( findArgParam("--dbserver-buffer-last-remove",conf->getArgc(),conf->getArgv()) != -1 )
lastRemove = true; lastRemove = true;
...@@ -383,6 +506,7 @@ void DBServer_MySQL::init_dbserver() ...@@ -383,6 +506,7 @@ void DBServer_MySQL::init_dbserver()
<< db->error() << endl; << db->error() << endl;
// throw Exception( string(myname+"(init): не смогли создать соединение с БД "+db->error()) ); // throw Exception( string(myname+"(init): не смогли создать соединение с БД "+db->error()) );
askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime); askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime);
askTimer(DBServer_MySQL::FlushBufferTimer,0);
} }
else else
{ {
...@@ -391,10 +515,12 @@ void DBServer_MySQL::init_dbserver() ...@@ -391,10 +515,12 @@ void DBServer_MySQL::init_dbserver()
connect_ok = true; connect_ok = true;
askTimer(DBServer_MySQL::ReconnectTimer,0); askTimer(DBServer_MySQL::ReconnectTimer,0);
askTimer(DBServer_MySQL::PingTimer,PingTime); askTimer(DBServer_MySQL::PingTimer,PingTime);
askTimer(DBServer_MySQL::FlushBufferTimer,flushBufferTime);
// createTables(db); // createTables(db);
initDB(db); initDB(db);
initDBTableMap(tblMap); initDBTableMap(tblMap);
flushBuffer(); flushQBuffer();
} }
} }
//-------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------
...@@ -435,6 +561,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm ) ...@@ -435,6 +561,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
connect_ok = false; connect_ok = false;
askTimer(DBServer_MySQL::PingTimer,0); askTimer(DBServer_MySQL::PingTimer,0);
askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime); askTimer(DBServer_MySQL::ReconnectTimer,ReconnectTime);
askTimer(DBServer_MySQL::FlushBufferTimer,0);
} }
else else
{ {
...@@ -456,6 +583,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm ) ...@@ -456,6 +583,7 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
connect_ok = true; connect_ok = true;
askTimer(DBServer_MySQL::ReconnectTimer,0); askTimer(DBServer_MySQL::ReconnectTimer,0);
askTimer(DBServer_MySQL::PingTimer,PingTime); askTimer(DBServer_MySQL::PingTimer,PingTime);
askTimer(DBServer_MySQL::FlushBufferTimer,flushBufferTime);
} }
connect_ok = false; connect_ok = false;
if( unideb.debugging(Debug::WARN) ) if( unideb.debugging(Debug::WARN) )
...@@ -466,6 +594,13 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm ) ...@@ -466,6 +594,13 @@ void DBServer_MySQL::timerInfo( UniSetTypes::TimerMessage* tm )
} }
break; break;
case FlushBufferTimer:
{
for( DBTableMap::iterator it=tblMap.begin(); it!=tblMap.end(); ++it )
flushTableBuffer(it);
}
break;
default: default:
if( unideb.debugging(Debug::WARN) ) if( unideb.debugging(Debug::WARN) )
unideb[Debug::WARN] << myname << "(timerInfo): Unknown TimerID=" << tm->id << endl; unideb[Debug::WARN] << myname << "(timerInfo): Unknown TimerID=" << tm->id << endl;
......
extensions/DBServer-MySQL/DBServer_MySQL.h
View file @ 5e90512f
...@@ -37,7 +37,7 @@ ...@@ -37,7 +37,7 @@
- \ref sec_DBS_Conf - \ref sec_DBS_Conf
- \ref sec_DBS_Tables - \ref sec_DBS_Tables
- \ref sec_DBS_Buffer - \ref sec_DBS_Buffer
- \ref sec_DBS_InsertBuffer
\section sec_DBS_Comm Общее описание работы DBServer_MySQL \section sec_DBS_Comm Общее описание работы DBServer_MySQL
Сервис предназначен для работы с БД MySQL. В его задачи входит Сервис предназначен для работы с БД MySQL. В его задачи входит
...@@ -73,8 +73,8 @@ ...@@ -73,8 +73,8 @@
\section sec_DBS_Buffer Защита от потери данных \section sec_DBS_Buffer Защита от потери данных
Для того, чтобы на момент отсутствия связи с БД данные по возможности не потерялись, Для того, чтобы на момент отсутствия связи с БД данные по возможности не потерялись,
сделан "кольцевой" буфер. Размер которго можно регулировать параметром "--dbserver-buffer-size" сделан "кольцевой" буфер. Размер которго можно регулировать параметром "--dbserver-lost-buffer-size"
или параметром \b bufferSize=".." в конфигурационном файле секции "<LocalDBSErver...>". или параметром \b lostBufferSize=".." в конфигурационном файле секции "<LocalDBServer...>".
Механизм построен на том, что если связь с mysql сервером отсутствует или пропала, Механизм построен на том, что если связь с mysql сервером отсутствует или пропала,
то сообщения помещаются в колевой буфер, который "опустошается" как только она восстановится. то сообщения помещаются в колевой буфер, который "опустошается" как только она восстановится.
...@@ -82,6 +82,20 @@ ...@@ -82,6 +82,20 @@
более ранние сообщения. Эту логику можно сменить, если указать параметр "--dbserver-buffer-last-remove" более ранние сообщения. Эту логику можно сменить, если указать параметр "--dbserver-buffer-last-remove"
или \b bufferLastRemove="1", то терятся будут сообщения добавляемые в конец. или \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 \section sec_DBS_Tables Таблицы MySQL
К основным таблицам относятся следующие: К основным таблицам относятся следующие:
\code \code
...@@ -142,7 +156,19 @@ class DBServer_MySQL: ...@@ -142,7 +156,19 @@ class DBServer_MySQL:
static const Debug::type DBLogInfoLevel = Debug::LEVEL9; static const Debug::type DBLogInfoLevel = Debug::LEVEL9;
protected: 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 initDB(DBInterface *db){};
virtual void initDBTableMap(DBTableMap& tblMap){}; virtual void initDBTableMap(DBTableMap& tblMap){};
...@@ -156,23 +182,28 @@ class DBServer_MySQL: ...@@ -156,23 +182,28 @@ class DBServer_MySQL:
virtual void parse( UniSetTypes::DBMessage* dbmsg ); virtual void parse( UniSetTypes::DBMessage* dbmsg );
virtual void parse( UniSetTypes::ConfirmMessage* cmsg ); virtual void parse( UniSetTypes::ConfirmMessage* cmsg );
/*! запись в БД */
bool writeToBase( const string& query ); bool writeToBase( const string& query );
/*! поместить в буфер для INSERT запросов к таблице 'key' */
bool insertToBuffer( int key, const string& values );
virtual void init_dbserver(); virtual void init_dbserver();
void createTables( DBInterface* db ); 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 enum Timers
{ {
PingTimer, /*!< таймер на переодическую проверку соединения с сервером БД */ PingTimer, /*!< таймер на переодическую проверку соединения с сервером БД */
ReconnectTimer, /*!< таймер на повторную попытку соединения с сервером БД (или восстановления связи) */ ReconnectTimer, /*!< таймер на повторную попытку соединения с сервером БД (или восстановления связи) */
FlushBufferTimer, /*!< таймер на сброс буфера */
lastNumberOfTimer lastNumberOfTimer
}; };
DBInterface *db; DBInterface *db;
int PingTime; int PingTime;
int ReconnectTime; int ReconnectTime;
...@@ -180,18 +211,21 @@ class DBServer_MySQL: ...@@ -180,18 +211,21 @@ class DBServer_MySQL:
bool activate; bool activate;
typedef std::queue<std::string> QueryBuffer;
QueryBuffer qbuf;
unsigned int qbufSize; // размер буфера сообщений.
bool lastRemove; bool lastRemove;
void flushBuffer(); int flushBufferTime;
UniSetTypes::uniset_mutex mqbuf;
/*! сброс insert-буфера для конкретной таблицы */
void flushTableBuffer( DBTableMap::iterator& i );
void flushQBuffer();
private: private:
DBTableMap tblMap;
DBTableMap tblMap;
UniSetTypes::uniset_mutex mqbuf;
QueryBuffer qbuf; // юуфер для запросов, пока нет связи с БД
unsigned int qbufSize;
}; };
//------------------------------------------------------------------------------------------ //------------------------------------------------------------------------------------------
#endif #endif
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