message queue implementation by c++ STL

1. 목적

내 기억으로는 c++ STL에는 IPC 기능을 제공하는 Container가 존재하지 않는 것 같다. 결국 POSIX 같은 리눅스/유닉스 기반 IPC를 혼합해서 써야 하는 것인지 의문이다. c++ 공부 차원에서 Thread간 통신을 위해서 Message Queue를 한번 구현해 봤다. 많은 것을 고민하였으나 부족한 점이 많아 보인다. 부족한 점은 잘 잊혀버리니까 정리차원에서 아래에 기술하도록 한다.

2. 테스트 코드

코드의 전체 구성은 다음과 같다. class CMessageQueue와 class CMessageQueuePool로 구성되어 있다. CMessageQueuePool은 각각의 이름으로 생성된 CMessageQueue를 관리하기 위한 목적 이외에는 다른 사용 목적은 없다. CMessageQueuePool은 SingleTon 형식으로 만들려고 했는데 정확히 맞는지 확신은 없다. 일단 상속은 생각하지 않았다. 내 기억으로는 Boost에서는 SingleTon을 상속 받아서 처리하면 매우 쉬운것으로 알고 있다. 사실 CMessageQueuePool은 별개 없고 실제 동작은 CMessageQueue에서 이루어진다.

그럼 CMessageQueue를 살펴보면 std::queue<T>와 std::mutex 맴버변수가 있고 전송과 수신 메세지 맴버함수로 이루어져 있다. std::queue<T>는 실제 주고 받을 메세지가 저장되는 Queue이다. template를 사용한 목적은 메세지의 형식이 목적에 따라 다르기 때문에 template로 처리하였다. std::mutex는 std::queue<T>를 전송과 수신 쓰레드에서 각각 접근하는 공유자원이다. 따라서 mutex로 동기를 맞추기 위해서 추가하였다.

한가지 아쉬운 점은 CMessageQueue는 CMessageQueuePool에서만 객체를 생성할 수 있도록 하고 쉽은데, 몇 가지 이유로 쉽지 않다. CMessageQueue를 CMessageQueuePool 내부에 집어 넣으면 getMessageQueue() 함수처럼 CMessageQueue를 리턴 받아야 하는 함수를 어떻게 처리할 지 의문이다. CMessageQueue를 리턴 받지 않고 모두 CMessageQueuePool 내부에서 처리하도록 하면은 매번 messageQueue 이름으로 CMessageQueue의 객체를 뒤져야 하는 성능상의 단점이 있다. 이 부분은 좀 더 고민을 해 봐야 겠다.

#include <iostream> #include <string> #include <sstream> #include <queue> #include <memory> #include <mutex> #include <map> #include <thread> #include <chrono> #include <pthread.h> enum class DebugLevel4MessageQueuePool { DEBUG_LEVEL_NORMAL, // No Debug Message from class DEBUG_LEVEL_ERROR, // Print Error Message DEBUG_LEVEL_DEBUG, // Print Debug Related Message DEBUG_LEVEL_TRACE, // All Debug Message Trace including class member function DEBUG_LEVEL_MAX }; DebugLevel4MessageQueuePool gDebugLevel = DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE; //const std::string red("\033[0;31m"); //const std::string green("\033[1;32m"); //const std::string yellow("\033[1;33m"); //const std::string cyan("\033[0;36m"); //const std::string magenta("\033[0;35m"); //const std::string reset("\033[0m"); #define DEBUG_PRINT(debugPrint, debugMsg)\ {\ if((debugPrint) <= gDebugLevel)\ {\ switch(debugPrint) \ {\ case DebugLevel4MessageQueuePool::DEBUG_LEVEL_NORMAL:\ std::printf("[%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4MessageQueuePool::DEBUG_LEVEL_ERROR:\ std::printf("\033[0;31m Error in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG:\ std::printf("\033[1;32m Debug in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE:\ std::printf("\033[1;33m Trace in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ default :\ break;\ }\ std::printf("\033[0m");\ }\ } template <typename T> class CMessageQueuePool; template <typename T> class CMessageQueue { friend class CMessageQueuePool<T>; private : std::queue<T> msgQ; std::mutex mtx; std::string nameOfmsgQ; pthread_mutex_t mutexForMsgQCond; pthread_cond_t msgQCond; public : CMessageQueue(); virtual ~CMessageQueue(); virtual void sendMessage(const T& msg); virtual T receiveMessage(void); virtual bool receiveMessage(T& msg); virtual bool isMessageQueueEmpty(void); }; template <typename T> CMessageQueue<T>::CMessageQueue() { msgQCond = PTHREAD_COND_INITIALIZER; mutexForMsgQCond = PTHREAD_MUTEX_INITIALIZER; pthread_cond_init(&msgQCond, NULL); pthread_mutex_init(&mutexForMsgQCond, NULL); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE, ""); } template <typename T> CMessageQueue<T>::~CMessageQueue() { std::queue<T> empty; std::swap(empty, msgQ); pthread_cond_destroy(&msgQCond); pthread_mutex_destroy(&mutexForMsgQCond); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE, ""); } template <typename T> void CMessageQueue<T>::sendMessage(const T& msg) { mtx.lock(); msgQ.push(msg); mtx.unlock(); pthread_mutex_lock(&mutexForMsgQCond); pthread_cond_signal(&msgQCond); pthread_mutex_unlock(&mutexForMsgQCond); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); } template <typename T> T CMessageQueue<T>::receiveMessage(void) { T msg; pthread_mutex_lock(&mutexForMsgQCond); while(isMessageQueueEmpty() == true) { DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_NORMAL, "Message Queue Empty Condition"); pthread_cond_wait(&msgQCond, &mutexForMsgQCond); } pthread_mutex_unlock(&mutexForMsgQCond); mtx.lock(); msg = msgQ.front(); msgQ.pop(); mtx.unlock(); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); return msg; } template <typename T> bool CMessageQueue<T>::receiveMessage(T& msg) { pthread_mutex_lock(&mutexForMsgQCond); while(isMessageQueueEmpty() == true) { DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_NORMAL, "Message Queue Empty Condition"); pthread_cond_wait(&msgQCond, &mutexForMsgQCond); } pthread_mutex_unlock(&mutexForMsgQCond); mtx.lock(); msg = msgQ.front(); msgQ.pop(); mtx.unlock(); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); return true; } template <typename T> bool CMessageQueue<T>::isMessageQueueEmpty(void) { bool isEmpty = false; mtx.lock(); isEmpty = msgQ.empty(); mtx.unlock(); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); return isEmpty; } template <typename T> class CMessageQueuePool { private : std::map<std::string, CMessageQueue<T>*> poolOfmsgQ; //using iterType = typename std::map<std::string, CMessageQueue<T>*>; CMessageQueuePool(){}; static std::shared_ptr<CMessageQueuePool<T>> inst; public : void setCurrentDebugLevel(DebugLevel4MessageQueuePool debugLevel); public : virtual ~CMessageQueuePool(); static std::shared_ptr<CMessageQueuePool<T>> getInstance(void); virtual bool createMessageQueue(std::string nameOfmsgQ); virtual CMessageQueue<T>* getMessageQueue(std::string nameOfmsgQ); }; template <typename T> std::shared_ptr<CMessageQueuePool<T>> CMessageQueuePool<T>::inst = nullptr; template <typename T> std::shared_ptr<CMessageQueuePool<T>> CMessageQueuePool<T>::getInstance(void) { if(inst == nullptr) { inst.reset(new CMessageQueuePool<T>); } DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE, ""); return inst; } template <typename T> CMessageQueuePool<T>::~CMessageQueuePool() { for(auto iter = poolOfmsgQ.begin(); iter != poolOfmsgQ.end(); iter++) { delete iter->second; } DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_TRACE, ""); poolOfmsgQ.clear(); } template <typename T> bool CMessageQueuePool<T>::createMessageQueue(std::string nameOfmsgQ) { if(poolOfmsgQ.find(nameOfmsgQ) != poolOfmsgQ.end()) { DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_ERROR, "Already Message Queue Exist!!"); return false; } CMessageQueue<T> *newMessageQueue = new CMessageQueue<T>; newMessageQueue->nameOfmsgQ = nameOfmsgQ; poolOfmsgQ[nameOfmsgQ] = newMessageQueue; DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); return true; } template <typename T> CMessageQueue<T>* CMessageQueuePool<T>::getMessageQueue(std::string nameOfmsgQ) { for(auto iter = poolOfmsgQ.begin(); iter != poolOfmsgQ.end(); iter++) { if(iter->first.compare(nameOfmsgQ) == 0) { return iter->second; } } DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); return nullptr; } template <typename T> void CMessageQueuePool<T>::setCurrentDebugLevel(DebugLevel4MessageQueuePool debugLevel) { gDebugLevel = debugLevel; DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_DEBUG, ""); } int main() { auto pSharedPtr = CMessageQueuePool<int>::getInstance(); pSharedPtr->createMessageQueue("heesoon.kim"); CMessageQueue<int> *handlerOfmsgQ = pSharedPtr->getMessageQueue("heesoon.kim"); if(handlerOfmsgQ == nullptr) { DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_NORMAL, "No Message handler Exist!!"); return 0; } std::thread sender = std::thread([&](){ for(int i = 0; i < 10; i++) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); handlerOfmsgQ->sendMessage(i); } }); std::thread receiver = std::thread([&](){ while(1) { //std::this_thread::sleep_for(std::chrono::milliseconds(500)); //int data = handlerOfmsgQ->receiveMessage(); int data = 0; handlerOfmsgQ->receiveMessage(data); std::stringstream ss; std::string msg; ss << "received data : " << data << std::endl; msg = ss.str(); DEBUG_PRINT(DebugLevel4MessageQueuePool::DEBUG_LEVEL_NORMAL, msg.c_str()); if(data == 9) break; } }); sender.join(); receiver.join(); return 0; }

3. 테스트 코드 실행

실행을 시켜보면 의도한 되로 정상동작한다. MesasgeQueue 전송 시에는 강제로 MessageQueue Empty 환경을 만들기 위해서 delay를 500ms 주고 던졌다. 수신쪽에서는 MessageQueue가 비어있으면 pthread condition 변수에 의해서 Task를 대기 시킴으로써 불필요한 CPU 점유률과 사용을 제거하였다.

 Trace in [getInstance,198]  
 Trace in [CMessageQueue,86]  
 Debug in [createMessageQueue,227]  
 Debug in [isMessageQueueEmpty,166]  
[receiveMessage,143] Message Queue Empty Condition 
 Debug in [sendMessage,112]  
 Debug in [isMessageQueueEmpty,166]  
 Debug in [receiveMessage,153]  
[operator(),287] received data : 0
 
 Debug in [isMessageQueueEmpty,166]  
[receiveMessage,143] Message Queue Empty Condition 
 Debug in [sendMessage,112]  
 Debug in [isMessageQueueEmpty,166]  
 Debug in [receiveMessage,153]  
[operator(),287] received data : 1
 
 Debug in [isMessageQueueEmpty,166]  
[receiveMessage,143] Message Queue Empty Condition 
 Debug in [sendMessage,112]  
 Debug in [isMessageQueueEmpty,166]  
 Debug in [receiveMessage,153]  
[operator(),287] received data : 2
 
 Debug in [isMessageQueueEmpty,166]  
[receiveMessage,143] Message Queue Empty Condition 
 Debug in [sendMessage,112]  
 Debug in [isMessageQueueEmpty,166]  
 Debug in [receiveMessage,153]  
[operator(),287] received data : 3