message queue 구현 - not using boost library

boost library interprocess에 message queue가 구현되어 있으나, 공부하는 차원에서 한번 만들어봤다.

소스가 너무 커서 일단 전체 소스는 zip파일 참조

1. 소스트리

.

├── CMakeLists.txt

├── messageQueue.cpp

├── messageQueue.h

├── remove.sh

├── test.cpp

└── types.h

2. message queue header 파일

#ifndef __MESSAGE_QUEUE_H__ #define __MESSAGE_QUEUE_H__ #include <iostream> #include <cstdio> #include <cstring> #include <string> #include <cerrno> #include <iomanip> #include <thread> #include <vector> #include <time.h> #include <fcntl.h> #include <sys/stat.h> #include <mqueue.h> #include "types.h" /* ------------------------------------------------------------------------------------------------------------------- declare define ------------------------------------------------------------------------------------------------------------------- */ #define PRINT_MSG_QUEUE #ifdef PRINT_MSG_QUEUE //#define PRINT_MSG_QUEUE_INFO(x) printf("%30s = %10d [%#10x]\n", #x, x, x); #else #define PRINT_MSG_QUEUE_INFO(x) #endif // message priority : Linux : 0 ~ sysconf(_SC_MQ_PRIO_MAX) -1, POSIX : 0 ~ 31 #define MSG_PRIO_URGENT 31 #define MSG_PRIO_NORMAL 0 #define MAX_MSG_QUEUE_NAME 32 #define PRE_MSG_QUEUE_NAME "/mq_" typedef struct MSG_Q_STATUS { mqd_t id; char msgQueueName[MAX_MSG_QUEUE_NAME]; std::thread::id senderId; std::thread::id receiverId; UINT32 sendMessageCount; UINT32 receivedMessageCount; } MSG_Q_STATUS_T; /* ------------------------------------------------------------------------------------------------------------------- declare class data ------------------------------------------------------------------------------------------------------------------- */ class messageQueue { public : messageQueue(); ~messageQueue(); static messageQueue* getInstance(); void printMessageQueueStatus(void); RESULT_T findMessageQueueId(SINT8* name, mqd_t *q_id); RESULT_T openMessageQueue(SINT8* name, UINT32 msgLen, UINT32 msgNum); RESULT_T openNonblockMessageQueue(SINT8* name, UINT32 msgLen, UINT32 msgNum); RESULT_T sendMessageQueue(mqd_t q_id, const SINT8* msg, UINT32 msgLen, UINT32 msgPrio); RESULT_T sendTimedMessageQueue(mqd_t q_id, const SINT8* msg, UINT32 msgLen, UINT32 msgPrio); SINT32 receiveMessageQueue(mqd_t q_id, SINT8* msg, UINT32 msgLen, UINT32* msgPrio); SINT32 receiveTimedMessageQueue(mqd_t q_id, SINT8* msg, UINT32 msgLen, UINT32* msgPrio); RESULT_T registerNotifyMessageQueue(mqd_t q_id, void (*callback)(union sigval)); private : std::vector<MSG_Q_STATUS_T> msgQueueStatusArray; }; #endif

3. 테스트

네가 만든 메세지 큐를 사용하는 방식은 아래와 같다.

1) 일단 메세지 큐를 연다. (Non Block 의 경우는 Non block open 함수로 연다)

2) 메세지를 보내는 task를 하나 만들고 메세지를 보낸다.

3) 메세지를 받는 task를 하나 만들고 메세지를 받는다.

main 함수도 하나의 task이므로 모든 메세지가 주고 받을 때까지 기다리다가

끝내려면 'x'를 눌러서 나간다.

thread 생성 후에 join을 하기보다는 detach로 작성한 이유는 메세지를 보내고

받는 쪽에서 처리가 안되면 계속 메세지 큐는 보내는 쪽에서 Blocking 상태임으로

모두 detach 시킨다.

이 경우 main 함수를 끝내면 안된다.

네가 구현한 메세지 큐는 메세지 큐 상태를 출력하는데 보내는 thread와 받는 thread의

ID와 현재 주고 받는 메세지 큐 상태를 check할 수 있다.

#include "messageQueue.h" typedef struct { char name[128]; int age; } testMessage_t; int main(int argc, char* argv[]) { char ch; mqd_t q_id; testMessage_t message = {"heesoon.kim", 41}; std::string msgQName = "test"; messageQueue *pMsgQ = new messageQueue(); if( pMsgQ->openMessageQueue(const_cast<SINT8*>(msgQName.c_str()), sizeof(message), 1) != RESULT_OK ) { std::cout << "message Queue open error " << std::endl; std::exit(EXIT_FAILURE); } if( pMsgQ->findMessageQueueId(const_cast<SINT8*>(msgQName.c_str()), &q_id) != RESULT_OK ) { std::cout << "message Queue find error " << std::endl; std::exit(EXIT_FAILURE); } // task sending message to below task std::thread tr_send = std::thread([&]() { int i = 0; while(i++ < 10) { if( pMsgQ->sendMessageQueue(q_id, (SINT8*)&message, sizeof(message), 1) != RESULT_OK ) { std::cout << "message Queue send error " << std::endl; std::exit(EXIT_FAILURE); } } }); // task receiving message from above task std::thread tr_receive = std::thread([&]() { while(1) { if( pMsgQ->receiveMessageQueue(q_id, (SINT8*)&message, sizeof(message), NULL) == RESULT_NOTOK) { std::cout << "message Queue receive error " << std::endl; } std::cout << "name : " << message.name << ", "; std::cout << "age : " << message.age << " " << std::endl; pMsgQ->printMessageQueueStatus(); } }); tr_send.detach(); tr_receive.detach(); while(1) { std::cout << "In main thread" << std::endl; std::cin.get(ch); if(ch == 'x' || ch == 'X') break; } delete pMsgQ; return 0; }

4. 실행 결과

받는 쪽 thread에서 현재 메세지 큐 상태를 확인하기 위해서 계속 출력하도록 해 봤다

~~~~~~~~~~~~ 중략 ~~~~~~~~~~~~~~~~~~~~~~ name : heesoon.kim, age : 41 --------------------------------------------------                  index : 0                     id : 3           msgQueueName : /mq_test               senderId : 3074534208             receiverId : 3066141504       sendMessageCount : 8   receivedMessageCount : 8 ------------------------------------------------- name : heesoon.kim, age : 41 --------------------------------------------------                  index : 0                     id : 3           msgQueueName : /mq_test               senderId : 3074534208             receiverId : 3066141504       sendMessageCount : 8   receivedMessageCount : 9 ------------------------------------------------- name : heesoon.kim, age : 41 --------------------------------------------------                  index : 0                     id : 3           msgQueueName : /mq_test               senderId : 3074534208             receiverId : 3066141504       sendMessageCount : 10   receivedMessageCount : 10 ------------------------------------------------- x called ~messageQueue

 

 

5. 소스 전체

 

messageQueueTest.zip