(MySQL): первая версия реализации буфера для INSERT-ов одним запросом

extensions/DBServer-MySQL/DBInterface.cc

@@ -47,13 +47,15 @@ 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,0))
+	if( !mysql_real_connect(mysql,host.c_str(), user.c_str(),pswd.c_str(),dbname.c_str(),0,NULL,CLIENT_MULTI_STATEMENTS) )
 	{
 		cout << error() << endl;
 		mysql_close(mysql);
 		connected = false;
 		return false;
 	}
+
+// 	mysql_set_server_option(mysql,MYSQL_OPTION_MULTI_STATEMENTS_ON);
 	connected = true;
 	return true;
 }
@@ -81,18 +83,33 @@ bool DBInterface::insert( 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 )
 		return false;
 
-	if( mysql_query(mysql,q.c_str()) )
+	if( mysql_query(mysql,q) )
 	{
 		queryok=false;
 		return false;
 	}
 
+	lastQ = ( noLastQ ) ? q : "";
 
-	lastQ = q;
 	result = mysql_store_result(mysql); // _use_result - некорректно работает с _num_rows
+
+	// Если при соединении используется CLIENT_MULTI_STATEMENTS
+	// то необходимо вынуть все результаты..
+	while( mysql_more_results(mysql) )
+	{	
+		cerr << "**** store result..." << endl;
+		if( mysql_next_result(mysql) >=0 )
+			mysql_store_result(mysql);
+	}
+
 	if( numRows()==0 )
 	{
 		queryok=false;

extensions/DBServer-MySQL/DBInterface.h

@@ -47,6 +47,8 @@ class DBInterface
 			bool close();
 			
 			bool query(const std::string& q);
+			bool query( const char* q, bool noLastQ=false ); // noLastQ - не запоминать запрос
+
 			const std::string lastQuery();
 			bool insert(const std::string& q);
 

extensions/DBServer-MySQL/DBServer_MySQL.cc

@@ -46,7 +46,8 @@ DBServer_MySQL::DBServer_MySQL(ObjectId id):
 	connect_ok(false),
 	activate(true),
 	qbufSize(200),
-	lastRemove(false)
+	lastRemove(false),
+	qbufByteSize(0)
 {
 	if( getId() == DefaultObjectId )
 	{
@@ -64,6 +65,7 @@ DBServer_MySQL::DBServer_MySQL():
 	connect_ok(false),
 	activate(true),
 	qbufSize(200),
+	qbufByteSize(0),
 	lastRemove(false)
 {
 //	init();
@@ -204,6 +206,23 @@ bool DBServer_MySQL::writeToBase( const string& query )
 {
 	if( unideb.debugging(DBLogInfoLevel) )
 		unideb[DBLogInfoLevel] << myname << "(writeToBase): " << query << endl;
+
+	// Складываем в очередь
+	bool fb = false;
+	{
+		uniset_mutex_lock l(mqbuf,2000);
+		qbuf.push(query);
+		fb = ( qbuf.size() >= qbufSize );
+		qbufByteSize += query.size();
+	}
+
+	if( fb )
+		flushBuffer();
+
+	return true;
+
+#if 0
+
 //	cout << "DBServer_MySQL: " << query << endl;
 	if( !db || !connect_ok )
 	{
@@ -244,32 +263,89 @@ bool DBServer_MySQL::writeToBase( const string& query )
 	}
 	
 	return false;
+#endif
 }
 //--------------------------------------------------------------------------------------------
 void DBServer_MySQL::flushBuffer()
 {
 	uniset_mutex_lock l(mqbuf,400);
 
-	// Сперва пробуем очистить всё что накопилось в очереди до этого...
+	if( qbuf.empty() )
+		return;
+
+	if( unideb.debugging(DBLEVEL) )
+		unideb[DBLEVEL] << myname << "(flushBuffer): qbufSize=" << qbufSize << " qbufByteSize=" << qbufByteSize << endl;
+
+	// buffer  чтобы записать всё одним запросом..
+	char* cbuf = new char[ qbufByteSize + 1 ];
+	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;
+		return;
+	}
+	
+	while( !qbuf.empty() )
+	{
+		std::string q(qbuf.front());
+		
+		int sz = q.size();
+		if( i + sz  > qbufByteSize )
+		{
+			if( unideb.debugging(Debug::CRIT) )
+				unideb[Debug::CRIT] << myname << "(flushBuffer): BUFFER OVERFLOW bytesize=" << qbufByteSize + 1 << " i=" << i << " sz=" << sz<< endl;
+
+			break;
+		}
+		
+		std::memcpy( &(cbuf[i]), q.c_str(), sz );
+		i += sz;
+		qbuf.pop();
+	}
+
+	// последняя точка с запятой, вроде не нужна (по документации и примерам из MySQL)..
+	if( cbuf[i-1] == ';' )
+		cbuf[i-1] = '\0';
+	else
+		cbuf[i] = '\0';
+	
+	if( i < qbufByteSize )
+	{
+		if( unideb.debugging(Debug::CRIT) )
+			unideb[Debug::CRIT] << myname << "(writeToBase): BAD qbufByteSize=" << qbufByteSize << " LOST RECORDS..(i=" << i << ")" << endl;
+
+		qbufByteSize = 0;
 		while( !qbuf.empty() )
+			qbuf.pop();
+	}
+
+	try
+	{
+		if( i > 0 )
 		{
-		db->query( qbuf.front() ); 
+			db->query( (const char*)cbuf, false );
 
 			// Дело в том что на INSERT И UPDATE запросы 
 			// db->query() может возвращать false и надо самому
 			// отдельно проверять действительно ли произошла ошибка
-		// см. DBInterface::query.
+			// см. DBInterface::queryh.
 			string err(db->error());
-		if( err.empty() )
+			if( !err.empty() && unideb.debugging(Debug::CRIT) )
+				unideb[Debug::CRIT] << myname << "(flushBuffer): error: " << err << endl;
+
 			db->freeResult();
-		else if( unideb.debugging(Debug::CRIT) )
+		}
+	}
+	catch( Exception& ex )
 	{
-			unideb[Debug::CRIT] << myname << "(writeToBase): error: " << err <<
-				" lost query: " << qbuf.front() << endl;
+		if( unideb.debugging(Debug::CRIT) )
+			unideb[Debug::CRIT] << myname << "(flushBuffer): " << ex << endl;
 	}
 
-		qbuf.pop();
-	}
+	delete[] cbuf;
+	qbufByteSize = 0;
 }
 //--------------------------------------------------------------------------------------------
 void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
@@ -295,7 +371,7 @@ void DBServer_MySQL::parse( UniSetTypes::SensorMessage *si )
 			<< 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;

extensions/DBServer-MySQL/DBServer_MySQL.h

@@ -169,6 +169,7 @@ class DBServer_MySQL:
 		{
 			PingTimer,        /*!< таймер на переодическую проверку соединения  с сервером БД */
 			ReconnectTimer,   /*!< таймер на повторную попытку соединения с сервером БД (или восстановления связи) */
+			FlushBufferTimer, /*!< таймер на сброс буфера */
 			lastNumberOfTimer
 		};
 
@@ -188,6 +189,7 @@ class DBServer_MySQL:
 
 		void flushBuffer();
 		UniSetTypes::uniset_mutex mqbuf;
+		int qbufByteSize; // размер буфера в байтах
 
 	private:
 		DBTableMap tblMap;

extensions/DBServer-MySQL/start_fg.sh

@@ -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 100
+--dbserver-buffer-size 50