[리눅스 프로그래밍] - (IPC) POSIX Message Queue 다루기

01. POSIX Message Queue

• Message 기반의 커뮤니케이션을 제공한다.
  • 부분 바이트 read 없음
  • Multiple message read 없음
• 하나의 메세지 큐로 read/write 가능
• 멀티 reader / 멀티 writer 기능 가능
• FILE I/O 기반 동작 → 이를 사용하여 I/O multiplexing 가능
• Message 우선 순위 설정 가능
• Message Notification 기능 존재
• 컴파일 시 -lrt 반드시 추가해야 한다.

02. 사용 API

• 헤더 파일

  #include<fcntl.h>
  #include<sys/stat.h>
  #include<mqueue.h>

• 사용 함수
  • 메세지 큐 관리
    • mqd_t mq_open(const char *name, int oflag) ← MQ 여는 함수
    • mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr); ← MQ 생성하여 여는 함수

      `파라미터
      • name: MQ 이름(반드시 /로 시작해야 함. 예: “/tmp/testmq”) • oflag
      ‐ mask of O_RDONLY, O_WRONLY, O_RDWR, O_CREAT, O_CLOEXEC, O_EXCL, O_NONBLOCK • mode: permission
      • attr: 속성 지정
      반환값
      • 성공시 message queue descriptor 리턴
      • 실패시 -1 리턴`

    • int mq_close(mqd_t mqdes); ← MQ 닫는 함수 ,성공시 0 실패시 -1
    • int mq_unlink(const char *name);← Mq 삭제 함수, 성공시 0 실패시 -1
  • 데이터 송수신
    • int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio); : 데이터 전송 함수

      파라미터
      • mqdes: Message queue descriptor
      • msg_ptr: pointer to send
      • msg_len: length to send
      • msg_prio: priority
      	‐ 0(lowest) ~ 32768(highest)
      반환값
      • 성공: 0 
      • 실패: -1

    • ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio); : 데이터 수신 함수

      파라미터
      • mqdes: Message queue descriptor
      • msg_ptr: pointer to receive
      • msg_len: length to receive
      	‐ 반드시 attr.mq_msgsize 보다 크거나 같아야 함 • msg_prio: priority
      	‐ 0(lowest) ~ 32768(highest)
      반환값
      • 성공: 읽은 바이트 수
      • 실패: -1

    • 기타
      • int mq_getattr(mqd_t mqdes, struct mq_attr *attr); : 속성 값을 가져오는 함수
      • int mq_setattr(mqd_t mqdes, const struct mq_attr *newattr, struct mq_attr *oldattr); : 속성 값을 셋팅하는 함수
        • struct mq_attr

          struct mq_attr {
                         long mq_flags;       /* Flags: 0 or O_NONBLOCK */
                         long mq_maxmsg;      /* Max. # of messages on queue */
                         long mq_msgsize;     /* Max. message size (bytes) */
                         long mq_curmsgs;     /* # of messages currently in queue */
                     };

      • int mq_notify(mqd_t mqdes, const struct sigevent *sevp) : 메세지 notification 설정

        파라미터
        • mqdes: Message queue descriptor
        • sevp: notification 설정
        반환값
        • 성공: 0
        • 실패: -1

        • union sigval

          union sigval { /* Data passed with notification */
          	int sival_int;         /* Integer value */
          	void *sival_ptr;       /* Pointer value */
          };

        • struct sigevent

          struct sigevent {
                     int          sigev_notify; /* Notification method */
                     int          sigev_signo;  /* Notification signal */
                     union sigval sigev_value;  /* Data passed with
                                                   notification */
                     void       (*sigev_notify_function) (union sigval);
                                      /* Function used for thread
                                         notification (SIGEV_THREAD) */
                     void        *sigev_notify_attributes;
                                      /* Attributes for notification thread
                                         (SIGEV_THREAD) */
                     pid_t        sigev_notify_thread_id;
                                      /* ID of thread to signal (SIGEV_THREAD_ID) */
                 };

03. 예제 코드

#include<stdio.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>
#include<errno.h>

#include<malloc.h>

#include<fcntl.h>
#include<sys/stat.h>
#include<mqueue.h>

#define FILE_PATH "/posix_example"

long max_len;

static void print_help(char *program_name)
{
    printf("Usage: %s (s|r)\n", program_name);
}

static void send_data(void)
{
    mqd_t mq_fd;
    struct mq_attr atr;
    char *buf;
    mq_fd = mq_open(FILE_PATH, O_RDWR);
    if(mq_fd == -1){
        perror("mq_open()");
        return ;
    }
    memset((void *)&atr, 0, sizeof(struct mq_attr));
    mq_getattr(mq_fd, &atr);
    printf("* mq_msgsize: %lu\n", atr.mq_msgsize);
    max_len = atr.mq_msgsize;
    buf = (char *)malloc(sizeof(char) * max_len);
    memset((void *)buf, 0, max_len);
    snprintf(buf, max_len, "hello world");
    if(mq_send(mq_fd, buf, max_len, 1) == -1){
        perror("mq_send()");
        free(buf);
        close(mq_fd);
        return ;
    }
    free(buf);
    mq_close(mq_fd);
}

static void recv_data(void)
{
    mqd_t mq_fd;
    struct mq_attr atr;
    char *buf;
    unsigned int priority;
    mq_fd = mq_open(FILE_PATH, O_CREAT | O_RDWR, 0644, NULL);
    if(mq_fd == -1){
        perror("mq_open()");
        return ;
    }
    memset((void *)&atr, 0, sizeof(struct mq_attr));
    mq_getattr(mq_fd, &atr);
    max_len = atr.mq_msgsize;
    buf = (char *)malloc(sizeof(char) * max_len);
    memset((void *)buf, 0, sizeof(buf));
    if(mq_receive(mq_fd, buf, max_len, &priority) == -1){
        perror("mq_receive()");
        free(buf);
        close(mq_fd);
        return ;
    }
    printf("[RECV] data: %s\n", buf);
    free(buf);
    mq_close(mq_fd);
}

int main(int argc, char **argv)
{
    if(argc < 2){
        print_help(argv[0]);
        return -1;
    }

    if(!strcmp((const char*)argv[1], "s"))
    {
        /* Sender */
        send_data();
    }
    else if(!strcmp((const char *)argv[1], "r"))
    {
        /* recv */
        recv_data();
    }
    else {
        print_help(argv[0]);
        return -1;
    }
    return 0;
}