Skip to content

U
uniset2
Commit 778fce6f authored by Pavel Vainerman's avatar Pavel Vainerman
Browse files
Options

(UMessageQueue): вариант реализации на atmic-ах, заложил тест для очереди сообщений.

parent 76921f15
Show whitespace changes
Inline Side-by-side
Showing with 222 additions and 350 deletions
include/UMessageQueue.h
View file @ 778fce6f
......@@ -17,44 +17,36 @@
#ifndef UMessageQueue_H_
#define UMessageQueue_H_
//--------------------------------------------------------------------------
#include <queue>
#include <atomic>
#include <vector>
#include <memory>
#include "Mutex.h"
#include "MessageType.h"
//--------------------------------------------------------------------------
typedef std::shared_ptr<UniSetTypes::VoidMessage> VoidMessagePtr;
//--------------------------------------------------------------------------
/*! \class UMessageQueue
* Очередь сообщений.
* \warning Очередь рассчитана на МНОГО ПИСАТЕЛЕЙ и ОДИН(!) читатель. Т.е. чтение должно быть из одного потока!
* Сообщения извлекаются из очереди в порядке приоритета сообщения. При одинаковом приоритете - в порядке поступления в очередь.
*
* Максимальное ограничение на размер очереди сообщений задаётся функцией setMaxSizeOfMessageQueue().
*
* Контроль переполения очереди осуществляется в двух местах push и receiveMessage.
* При переполнении очереди, происходит автоматическая очистка в два этапа.
* Первый: производиться попытка "свёртки" сообщений.
* Из очереди все повторяющиеся
* - SensorMessage
* - TimerMessage
* - SystemMessage
* Если это не помогло, то производиться второй этап "чистки":
* Из очереди удаляется MaxCountRemoveOfMessage сообщений.
* Этот парамер задаётся при помощи setMaxCountRemoveOfMessage(). По умолчанию 1/4 очереди сообщений.
* Чтобы избежать работы с mutex, очередь построена по принципу циклического буфера,
* c использованием atomic-переменных и попыткой реализовать LockFree работу.
* Есть указатель на текущую позицию записи (wp) и есть "догоняющий его" указатель на позицию чтения (rp).
* Если rp догоняет wp - значит новых сообщений нет.
*
* Очистка реализована в функции cleanMsgQueue();
* При этом место под очередь(буффер) резервируется сразу.
* Счётчики сделаны (uint) монотонно растущими.
* Основные идеи:
* - счётчики постоянно увеличиваются
* - каждый пишущий поток пишет в новое место
* - читающий счётчик тоже монотонно растёт
* - реальная позиция для записи или чтения рассчитывается как (pos%size) этим и обеспечивается цикличность.
*
* \warning Т.к. при фильтровании SensorMessage не смотрится значение,
* то при удалении сообщений об изменении аналоговых датчиков очистка может привести
* к некорректной работе фильрующих алгоритмов работающих с "выборкой" последних N значений.
* (потому-что останется одно последнее)
* Максимальное ограничение на размер очереди сообщений задаётся функцией setMaxSizeOfMessageQueue().
*
* ОПТИМИЗАЦИЯ N1:
* Для того, чтобы функции push() и top() реже "сталкавались" на mutex-е очереди сообщений.
* Сделано две очереди сообщений. Одна очередь сообщений наполняется в push() (с блокировкой mutex-а),
* а вторая (без блокировки) обрабатывается в top(). Как только сообщения заканчиваются в
* top() очереди меняются местами (при захваченном mutex).
* Контроль переполения очереди осуществляется в push и в top;
* Если очередь переполняется, то сообщения ТЕРЯЮТСЯ!
* При помощи функции setLostStrategy() можно установить стратегию что терять
* lostNewData - в случае переполнения теряются новые данные (т.е. не будут помещаться в очередь)
* lostOldData - в случае переполнения очереди, старые данные затираются новыми.
*/
class UMessageQueue
{
......@@ -71,8 +63,13 @@ class UMessageQueue
void setMaxSizeOfMessageQueue( size_t s );
size_t getMaxSizeOfMessageQueue();
void setMaxCountRemoveOfMessage( size_t m );
size_t getMaxCountRemoveOfMessage();
enum LostStrategy
{
lostOldData, // default
lostNewData
};
void setLostStrategy( LostStrategy s );
// ---- Статистика ----
/*! максимальное количество которое было в очереди сообщений */
......@@ -87,37 +84,24 @@ class UMessageQueue
return stCountOfQueueFull;
}
// функция определения приоритетного сообщения для обработки
struct VoidMessageCompare:
public std::binary_function<VoidMessagePtr, VoidMessagePtr, bool>
{
bool operator()(const VoidMessagePtr& lhs,
const VoidMessagePtr& rhs) const;
};
typedef std::priority_queue<VoidMessagePtr, std::vector<VoidMessagePtr>, VoidMessageCompare> MQueue;
typedef std::vector<VoidMessagePtr> MQueue;
protected:
/*! Чистка очереди сообщений */
void cleanMsgQueue( MQueue& q );
// заполнить всю очередь указанным сообщением
void mqFill( const VoidMessagePtr& v );
private:
MQueue* wQ = { nullptr }; // указатель на текущую очередь на запись
MQueue* rQ = { nullptr }; // указатель на текущую очередь на чтение
MQueue mq1,mq2;
MQueue mqueue;
std::atomic_uint wpos = { 0 }; // позиция на запись
std::atomic_uint rpos = { 0 }; // позиция на чтение
std::atomic_uint mpos = { 0 }; // текущая позиция последнего элемента (max position) (реально добавленного в очередь)
LostStrategy lostStrategy = { lostOldData };
/*! размер очереди сообщений (при превышении происходит очистка) */
size_t SizeOfMessageQueue = { 2000 };
/*! сколько сообщений удалять при очисте */
size_t MaxCountRemoveOfMessage = { 500 };
/*! замок для блокирования совместного доступа к очереди */
UniSetTypes::uniset_rwmutex qmutex;
// статистическая информация
size_t stMaxQueueMessages = { 0 }; /*!< Максимальное число сообщений хранившихся в очереди */
size_t stCountOfQueueFull = { 0 }; /*!< количество переполнений очереди сообщений */
......
include/UniSetObject.h
View file @ 778fce6f
......@@ -195,16 +195,6 @@ class UniSetObject:
return mqueue.getMaxSizeOfMessageQueue();
}
void setMaxCountRemoveOfMessage( size_t m )
{
mqueue.setMaxCountRemoveOfMessage(m);
}
inline size_t getMaxCountRemoveOfMessage()
{
return mqueue.getMaxCountRemoveOfMessage();
}
inline bool isActive()
{
return active;
......
src/ObjectRepository/UMessageQueue.cc
View file @ 778fce6f
......@@ -25,329 +25,97 @@ using namespace std;
UMessageQueue::UMessageQueue( size_t qsize ):
SizeOfMessageQueue(qsize)
{
wQ = &mq1;
rQ = &mq2;
mqFill(nullptr);
}
//---------------------------------------------------------------------------
void UMessageQueue::push( const UniSetTypes::TransportMessage& tm )
{
// проверяем переполнение, только если стратегия "терять новые данные"
// иначе нет смысла проверять, а можно просто писать новые данные затирая старые
if( lostStrategy == lostNewData && (wpos - rpos) >= SizeOfMessageQueue )
{
// lock
uniset_rwmutex_wrlock mlk(qmutex);
// контроль переполнения
if( !wQ->empty() && wQ->size() > SizeOfMessageQueue )
{
// ucrit << myname << "(push): message queue overflow!" << endl << flush;
cleanMsgQueue(*wQ);
// обновляем статистику
stCountOfQueueFull++;
stMaxQueueMessages = 0;
return;
}
auto v = make_shared<VoidMessage>(tm);
wQ->push(v);
// сперва надо сдвинуть счётчик (чтобы следующий поток уже писал в новое место)
size_t w = wpos.fetch_add(1);
// максимальное число ( для статистики )
if( wQ->size() > stMaxQueueMessages )
stMaxQueueMessages = wQ->size();
// а потом уже добавлять новое сообщение в "зарезервированное" место
mqueue[w%SizeOfMessageQueue] = make_shared<VoidMessage>(tm);
mpos++; // теперь увеличиваем реальное количество элементов в очереди
} // unlock
// ведём статистику
size_t sz = mpos - rpos;
if( sz > stMaxQueueMessages )
stMaxQueueMessages = sz;
}
//---------------------------------------------------------------------------
VoidMessagePtr UMessageQueue::top()
{
// здесь работаем со своей очередью без блокировки
if( !rQ->empty() )
{
auto m = rQ->top(); // получили сообщение
rQ->pop(); // удалили сообщение из очереди
return m;
}
// Если своя очередь пуста
// то смотрим вторую
// если стратегия "потеря старых данных"
// то надо постоянно "подтягивать" rpos к wpos
if( lostStrategy == lostOldData && (wpos - rpos) >= SizeOfMessageQueue )
{
// lock
uniset_rwmutex_wrlock mlk(qmutex);
if( !wQ->empty() )
{
// контроль переполнения
if( wQ->size() > SizeOfMessageQueue )
{
// ucrit << myname << "(receiveMessages): messages queue overflow!" << endl << flush;
cleanMsgQueue(*wQ);
// обновляем статистику по переполнениям
stCountOfQueueFull++;
stMaxQueueMessages = 0;
rpos.store( wpos - SizeOfMessageQueue - 1 );
}
if( !wQ->empty() )
// смотрим "фактическое" количество (mpos)
// т.к. помещение в вектор тоже занимает время
// а при этом wpos у нас уже будет +1
if( rpos < mpos )
{
auto m = wQ->top(); // получили сообщение
wQ->pop(); // удалили сообщение из очереди
// сперва надо сдвинуть счётчик (чтобы следующий поток уже читал новое)
size_t r = rpos.fetch_add(1);
// меняем очереди местами
std::swap(rQ,wQ);
return m;
// т.к. между if и этим местом, может придти другой читающий поток, то
// проверяем здесь ещё раз
if( r < mpos )
return mqueue[r%SizeOfMessageQueue];
}
}
} // unlock queue
return nullptr;
}
//---------------------------------------------------------------------------
size_t UMessageQueue::size()
{
uniset_rwmutex_wrlock mlk(qmutex);
return (wQ->size() + rQ->size());
}
//---------------------------------------------------------------------------
void UMessageQueue::setMaxSizeOfMessageQueue(size_t s)
{
SizeOfMessageQueue = s;
}
size_t UMessageQueue::getMaxSizeOfMessageQueue()
{
// т.к. rpos корректируется только при фактическом вызое top()
// то тут приходиться смотреть если у нас переполнение
// то надо просто вернуть размер очереди
// (т.к. фактически циклический буфер никогда не переполняется)
if( (wpos - rpos) >= SizeOfMessageQueue )
return SizeOfMessageQueue;
}
//---------------------------------------------------------------------------
void UMessageQueue::setMaxCountRemoveOfMessage( size_t m )
{
MaxCountRemoveOfMessage = m;
}
size_t UMessageQueue::getMaxCountRemoveOfMessage()
{
return MaxCountRemoveOfMessage;
return (mpos - rpos);
}
//---------------------------------------------------------------------------
// структура определяющая минимальное количество полей
// по которым можно судить о схожести сообщений
// используется локально и только в функции очистки очереди сообщений
struct MsgInfo
void UMessageQueue::setMaxSizeOfMessageQueue( size_t s )
{
MsgInfo():
type(Message::Unused),
id(DefaultObjectId),
node(DefaultObjectId)
if( s != SizeOfMessageQueue )
{
// struct timezone tz;
tm.tv_sec = 0;
tm.tv_usec = 0;
// gettimeofday(&tm,&tz);
}
int type;
ObjectId id; // от кого
struct timeval tm; // время
ObjectId node; // откуда
inline bool operator < ( const MsgInfo& mi ) const
{
if( type != mi.type )
return type < mi.type;
if( id != mi.id )
return id < mi.id;
if( node != mi.node )
return node < mi.node;
if( tm.tv_sec != mi.tm.tv_sec )
return tm.tv_sec < mi.tm.tv_sec;
return tm.tv_usec < mi.tm.tv_usec;
SizeOfMessageQueue = s;
mqFill(nullptr);
}
};
else
mqFill(nullptr);
}
//---------------------------------------------------------------------------
// структура определяющая минимальное количество полей
// по которым можно судить о схожести сообщений
// используется локально и только в функции очистки очереди сообщений
struct CInfo
size_t UMessageQueue::getMaxSizeOfMessageQueue()
{
CInfo():
sensor_id(DefaultObjectId),
value(0),
time(0),
time_usec(0),
confirm(0)
{
}
explicit CInfo( ConfirmMessage& cm ):
sensor_id(cm.sensor_id),
value(cm.value),
time(cm.time),
time_usec(cm.time_usec),
confirm(cm.confirm)
{}
long sensor_id; /* ID датчика */
double value; /* значение датчика */
time_t time; /* время, когда датчик получил сигнал */
time_t time_usec; /* время в микросекундах */
time_t confirm; /* время, когда произошло квитирование */
inline bool operator < ( const CInfo& mi ) const
{
if( sensor_id != mi.sensor_id )
return sensor_id < mi.sensor_id;
if( value != mi.value )
return value < mi.value;
if( time != mi.time )
return time < mi.time;
return time_usec < mi.time_usec;
}
};
return SizeOfMessageQueue;
}
//---------------------------------------------------------------------------
struct tmpConsumerInfo
{
tmpConsumerInfo() {}
unordered_map<UniSetTypes::KeyType, VoidMessagePtr> smap;
unordered_map<int, VoidMessagePtr> tmap;
unordered_map<int, VoidMessagePtr> sysmap;
std::map<CInfo, VoidMessagePtr> cmap;
list<VoidMessagePtr> lstOther;
};
void UMessageQueue::cleanMsgQueue( UMessageQueue::MQueue& q )
void UMessageQueue::setLostStrategy( UMessageQueue::LostStrategy s )
{
#if 0
ucrit << myname << "(cleanMsgQueue): msg queue cleaning..." << endl << flush;
ucrit << myname << "(cleanMsgQueue): current size of queue: " << q.size() << endl << flush;
#endif
// проходим по всем известным нам типам(базовым)
// ищем все совпадающие сообщения и оставляем только последние...
VoidMessage m;
unordered_map<UniSetTypes::ObjectId, tmpConsumerInfo> consumermap;
while( !q.empty() )
{
auto m = q.top();
q.pop();
switch( m->type )
{
case Message::SensorInfo:
{
SensorMessage sm(m.get());
UniSetTypes::KeyType k(key(sm.id, sm.node));
// т.к. из очереди сообщений сперва вынимаются самые старые, потом свежее и т.п.
// то достаточно просто сохранять последнее сообщение для одинаковых Key
consumermap[sm.consumer].smap[k] = m;
}
break;
case Message::Timer:
{
TimerMessage tm(m.get());
// т.к. из очереди сообщений сперва вынимаются самые старые, потом свежее и т.п.
// то достаточно просто сохранять последнее сообщение для одинаковых TimerId
consumermap[tm.consumer].tmap[tm.id] = m;
}
break;
case Message::SysCommand:
{
SystemMessage sm(m.get());
consumermap[sm.consumer].sysmap[sm.command] = m;
}
break;
case Message::Confirm:
{
ConfirmMessage cm(m.get());
CInfo ci(cm);
// т.к. из очереди сообщений сперва вынимаются самые старые, потом свежее и т.п.
// то достаточно просто сохранять последнее сообщение для одинаковых MsgInfo
consumermap[cm.consumer].cmap[ci] = m;
}
break;
case Message::Unused:
// просто выкидываем (игнорируем)
break;
default:
// сразу помещаем в очередь
consumermap[m->consumer].lstOther.push_front(m);
break;
}
}
// ucrit << myname << "(cleanMsgQueue): ******** cleanup RESULT ********" << endl;
for( auto& c : consumermap )
{
#if 0
ucrit << myname << "(cleanMsgQueue): CONSUMER=" << c.first << endl;
ucrit << myname << "(cleanMsgQueue): after clean SensorMessage: " << c.second.smap.size() << endl;
ucrit << myname << "(cleanMsgQueue): after clean TimerMessage: " << c.second.tmap.size() << endl;
ucrit << myname << "(cleanMsgQueue): after clean SystemMessage: " << c.second.sysmap.size() << endl;
ucrit << myname << "(cleanMsgQueue): after clean ConfirmMessage: " << c.second.cmap.size() << endl;
ucrit << myname << "(cleanMsgQueue): after clean other: " << c.second.lstOther.size() << endl;
#endif
// теперь ОСТАВШИЕСЯ запихиваем обратно в очередь...
for( auto& v : c.second.smap )
q.push(v.second);
for( auto& v : c.second.tmap )
q.push(v.second);
for( auto& v : c.second.sysmap )
q.push(v.second);
for( auto& v : c.second.cmap )
q.push(v.second);
for( auto& v : c.second.lstOther )
q.push(v);
}
// ucrit << myname
// << "(cleanMsgQueue): ******* result size of queue: "
// << q.size()
// << " < " << getMaxSizeOfMessageQueue() << endl;
if( q.size() >= SizeOfMessageQueue )
{
// ucrit << myname << "(cleanMsgQueue): clean failed. size > " << q.size() << endl;
// ucrit << myname << "(cleanMsgQueue): remove " << getMaxCountRemoveOfMessage() << " old messages " << endl;
for( unsigned int i = 0; i < MaxCountRemoveOfMessage; i++ )
{
q.top();
q.pop();
if( q.empty() )
break;
}
// ucrit << myname << "(cleanMsgQueue): result size=" << q.size() << endl;
}
lostStrategy = s;
}
//---------------------------------------------------------------------------
bool UMessageQueue::VoidMessageCompare::operator()(const VoidMessagePtr& lhs, const VoidMessagePtr& rhs) const
void UMessageQueue::mqFill( const VoidMessagePtr& v )
{
if( lhs->priority == rhs->priority )
{
if( lhs->tm.tv_sec == rhs->tm.tv_sec )
return lhs->tm.tv_usec >= rhs->tm.tv_usec;
return lhs->tm.tv_sec >= rhs->tm.tv_sec;
}
return lhs->priority < rhs->priority;
mqueue.reserve(SizeOfMessageQueue);
mqueue.clear();
for( size_t i=0; i<SizeOfMessageQueue; i++ )
mqueue.push_back(v);
}
//---------------------------------------------------------------------------
src/ObjectRepository/UniSetObject.cc
View file @ 778fce6f
......@@ -160,14 +160,7 @@ void UniSetObject::initObject()
if( sz > 0 )
mqueue.setMaxSizeOfMessageQueue(sz);
int maxremove = conf->getArgPInt("--uniset-object-maxcount-remove-message", conf->getField("MaxCountRemoveOfMessage"), sz / 4);
if( maxremove > 0 )
mqueue.setMaxCountRemoveOfMessage(maxremove);
// workingTerminateTimeout = conf->getArgPInt("--uniset-object-working-terminate-timeout",conf->getField("WorkingTerminateTimeout"),2000);
uinfo << myname << "(init): SizeOfMessageQueue=" << mqueue.getMaxSizeOfMessageQueue()
<< " MaxCountRemoveOfMessage=" << mqueue.getMaxCountRemoveOfMessage()
<< endl;
}
// ------------------------------------------------------------------------------------------
......
tests/MQPerfTest/mq-test.cc
View file @ 778fce6f
#include <string>
#include <iostream>
#include <assert.h>
#include <thread>
#include <atomic>
#include "UMessageQueue.h"
......@@ -58,6 +59,17 @@ int main(int argc, const char** argv)
// чтобы не происходило переполнение
mq.setMaxSizeOfMessageQueue(COUNT+1);
// сперва просто проверка что очередь работает.
{
SensorMessage sm(100,2);
TransportMessage tm( std::move(sm.transport_msg()) );
mq.push(tm);
auto msg = mq.top();
assert( msg!=nullptr );
SensorMessage sm2( msg.get() );
assert( sm.id == sm2.id );
}
vector<int> res;
res.reserve(tnum);
......
tests/Makefile.am
View file @ 778fce6f
......@@ -35,7 +35,8 @@ test_conftest.cc \
test_ui.cc \
test_iorfile.cc \
test_messagetype.cc \
test_utypes.cc
test_utypes.cc \
test_mqueue.cc
# threadtst_SOURCES = threadtst.cc
# threadtst_LDADD = $(top_builddir)/lib/libUniSet2.la ${SIGC_LIBS} $(COMCPP_LIBS)
......
tests/test_mqueue.cc 0 → 100644
View file @ 778fce6f
#include <catch.hpp>
// --------------------------------------------------------------------------
#include "UMessageQueue.h"
#include "MessageType.h"
#include "Configuration.h"
// --------------------------------------------------------------------------
using namespace std;
using namespace UniSetTypes;
// --------------------------------------------------------------------------
TEST_CASE( "UMessageQueue: setup", "[mqueue]" )
{
UMessageQueue mq;
// проверка установки размера очереди
mq.setMaxSizeOfMessageQueue(10);
REQUIRE(mq.getMaxSizeOfMessageQueue() == 10 );
}
// --------------------------------------------------------------------------
TEST_CASE( "UMessageQueue: simple push/top", "[mqueue]" )
{
REQUIRE( uniset_conf() != nullptr );
UMessageQueue mq;
SensorMessage sm(100,2);
TransportMessage tm( std::move(sm.transport_msg()) );
mq.push(tm);
auto msg = mq.top();
REQUIRE( msg!=nullptr );
SensorMessage sm2( msg.get() );
REQUIRE( sm.id == sm2.id );
}
// --------------------------------------------------------------------------
TEST_CASE( "UMessageQueue: overflow (lost old data)", "[mqueue]" )
{
REQUIRE( uniset_conf() != nullptr );
UMessageQueue mq;
mq.setMaxSizeOfMessageQueue(1);
mq.setLostStrategy( UMessageQueue::lostOldData );
SensorMessage sm1(100,2);
TransportMessage tm1( std::move(sm1.transport_msg()) );
mq.push(tm1);
REQUIRE( mq.size() == 1 );
SensorMessage sm2(110,50);
TransportMessage tm2( std::move(sm2.transport_msg()) );
mq.push(tm2);
REQUIRE( mq.size() == 1 );
auto msg = mq.top();
REQUIRE( msg!=nullptr );
SensorMessage sm( msg.get() );
REQUIRE( sm.id == sm2.id );
REQUIRE( mq.getCountOfQueueFull() == 1 );
}
// --------------------------------------------------------------------------
TEST_CASE( "UMessageQueue: overflow (lost new data)", "[mqueue]" )
{
REQUIRE( uniset_conf() != nullptr );
UMessageQueue mq;
mq.setMaxSizeOfMessageQueue(1);
mq.setLostStrategy( UMessageQueue::lostNewData );
SensorMessage sm1(100,2);
TransportMessage tm1( std::move(sm1.transport_msg()) );
mq.push(tm1);
REQUIRE( mq.size() == 1 );
SensorMessage sm2(110,50);
TransportMessage tm2( std::move(sm2.transport_msg()) );
mq.push(tm2);
REQUIRE( mq.getCountOfQueueFull() == 1 );
SensorMessage sm3(120,150);
TransportMessage tm3( std::move(sm3.transport_msg()) );
mq.push(tm3);
REQUIRE( mq.size() == 1 );
REQUIRE( mq.getCountOfQueueFull() == 2 );
auto msg = mq.top();
REQUIRE( msg!=nullptr );
SensorMessage sm( msg.get() );
REQUIRE( sm.id == sm1.id );
}
// --------------------------------------------------------------------------
TEST_CASE( "UMessageQueue: many read", "[mqueue]" )
{
REQUIRE( uniset_conf() != nullptr );
UMessageQueue mq;
mq.setMaxSizeOfMessageQueue(1);
mq.setLostStrategy( UMessageQueue::lostNewData );
SensorMessage sm1(100,2);
TransportMessage tm1( std::move(sm1.transport_msg()) );
mq.push(tm1);
REQUIRE( mq.size() == 1 );
auto msg = mq.top();
REQUIRE( msg!=nullptr );
SensorMessage sm( msg.get() );
REQUIRE( sm.id == sm1.id );
for( int i=0; i<5; i++ )
{
auto msg = mq.top();
REQUIRE( msg==nullptr );
}
}
// --------------------------------------------------------------------------
uniset2.files
View file @ 778fce6f
......@@ -457,6 +457,7 @@ tests/test_unixml.cc
tests/test_utypes.cc
tests/test_logserver.cc
tests/test_tcpcheck.cc
tests/test_mqueue.cc
tests/tests-junit.xml
tests/tests.cc
tests/tests_bad_config.xml
......
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