18. Message Queue [Interesting topic to study]

안녕하세요. 오늘은 Message Queue에 대해 알아볼까 합니다.

2021.11.21 - [학부공부/OS_운영체제] - 15. IPC(1)

15. IPC(1)

2021.11.24 - [학부공부/OS_운영체제] - 17. IPC(2)

17. IPC(2)

앞선 두개의 포스팅에서 IPC의 메커니즘에 대해 소개해드렸습니다. Message passing model의 case로 Message Queue에 대해 알려 드렸는데요. Message queue 방식은 고정된 size의 queue를 이용하는 기법이라고 설명하였습니다. Shared memory와 pipe의 중간 수준이라고 말씀드렸고요. 오늘은 이 Message queue에 대해 조금 더 알아보도록 하겠습니다. 다른 IPC에 대한 공부를 하실 때도 이러한 study track을 참고하시고 도움되시길 바랍니다.

 

 

18.01. Message Queue

메시지 큐는 IPC에서 메시지 단위의 송수신을 가능하게 해주는 큐입니다. Linux 커널에서 관리하며 기본적으로 모든 프로세스에서 접근이 가능하도록 구현되어 있습니다. 프로세스가 생성한 메시지는 IPC 메시지 큐에 저장되고 다른 프로세스가 읽으면 큐에서 제거되는 방식입니다. 앞선 포스팅에서 개략적인 설명을 드린바 있지만 다시 한번 정리하도록 하겠습니다.

 

18.01.1. msg_msg structure

Message는 고정된 size의 header와 가변적인 size를 가지는 text로 이루어집니다. Header부분에 type을 나타내는 값이 붙을 수 있고 이 값을 이용해 선택적으로 메시지를 읽는 것이 가능합니다. include/linux/msg.h에 struct msg_msg로 정의되어 있습니다.

 /* one msg_msg structure for each message */
struct msg_msg {
    struct list_head m_list; 
    long  m_type;          
    int m_ts;           /* message text size */
	struct msg_msgseg* next;
	void *security;
	/* the actual message follows immediately */
};

 

18.01.2. msg_queue structure

Message Queue 자체는 permission 정보와 큐의 bytes, message 수 등의 정보를 가지고 있고 include/linux/msg.h에  struct msg_queue로 정의되어 있습니다.

/* one msq_queue structure for each present queue on the system */
struct msg_queue {
	struct kern_ipc_perm q_perm;
	time_t q_stime;         /* last msgsnd time */
	time_t q_rtime;         /* last msgrcv time */
	time_t q_ctime;         /* last change time */
	unsigned long q_cbytes;     /* current number of bytes on queue */
	unsigned long q_qnum;       /* number of messages in queue */
	unsigned long q_qbytes;     /* max number of bytes on queue */
	pid_t q_lspid;          /* pid of last msgsnd */
	pid_t q_lrpid;          /* last receive pid */
	
	struct list_head q_messages;
	struct list_head q_receivers;
	struct list_head q_senders;
};

msg_queue는 메시지와 linked list의 구조로 연결되어 있으며 q_messages는 메시지 큐의 가장 앞쪽 메시지와 연결되고 각각의 메시지는 m_list를 통해 연결됩니다.

 

18.01.3. msqid_ds structure

메시지 큐가 생성될 때마다 메시지 큐에 대한 정보를 가진 객체가 생성되는데 마지막으로 송수신한 프로세스 ID, 송수신 시간, 큐의 최대 byte 등 정보가 저장됩니다. include/uapi/linux/msg.h에 struct msqid_ds로 정의되어 있습니다. 

/* Obsolete, used only for backwards compatibility and libc5 compiles */
struct msqid_ds {
	struct ipc_perm msg_perm;
	struct msg *msg_first;      /* first message on queue,unused  */
	struct msg *msg_last;       /* last message in queue,unused */
	__kernel_time_t msg_stime;  /* last msgsnd time */
	__kernel_time_t msg_rtime;  /* last msgrcv time */
	__kernel_time_t msg_ctime;  /* last change time */
	unsigned long  msg_lcbytes; /* Reuse junk fields for 32 bit */
	unsigned long  msg_lqbytes; /* ditto */
	unsigned short msg_cbytes;  /* current number of bytes on queue */
	unsigned short msg_qnum;    /* number of messages in queue */
	unsigned short msg_qbytes;  /* max number of bytes on queue */
	__kernel_ipc_pid_t msg_lspid;   /* pid of last msgsnd */
	__kernel_ipc_pid_t msg_lrpid;   /* last receive pid */
};

참고로 같은 경로에 version마다 다르겠지만 아래와 같은 값들도 상수로 정의되어 있습니다.

/*
* Scaling factor to compute msgmni:
* the memory dedicated to msg queues (msgmni * msgmnb) should occupy
* at most 1/MSG_MEM_SCALE of the lowmem (see the formula in ipc/msg.c):
* up to 8MB       : msgmni = 16 (MSGMNI)
* 4 GB            : msgmni = 8K
* more than 16 GB : msgmni = 32K (IPCMNI)
*/
#define MSG_MEM_SCALE 32
 
#define MSGMNI    16   /* <= IPCMNI */     /* max # of msg queue identifiers */
#define MSGMAX  8192   /* <= INT_MAX */   /* max size of message (bytes) */
#define MSGMNB 16384   /* <= INT_MAX */   /* default max size of a message queue */

 

 

18.02. Function

각 structure에 대해 알아보았으니 이제 메시지를 drive하는 function들을 소개하겠습니다.

 

18.02.1. msgget function

 int msgget(key_t key, int msgflg);

msgget function은 메시지 큐를 생성하거나 기존 메시지 큐를 참조하는 함수입니다. parameter인 key는 메시지 큐를 구분하기 위한 id라 생각하시면 되고 msgflg는 메시지 큐를 생성할 때 옵션을 주는 것입니다. 

• IPC_CREAT : key에 해당하는 큐가 있으면 큐의 식별자 반환, 없으면 생성.
• IPC_EXCL : 없으면 생성, 존재하면 msgget은 -1을 반환.

 

18.02.2. msgsnd function

메시지 큐에 data를 전송하는 함수입니다.

 int msgsnd(int msqid, struct msgbuf *msgp, size_t msgsz, int msgflg);

parameter인 msqud는 메시지 큐 id이고 msgbuf structure는 데이터 전송시 사용되는 메시지 구조로 첫 4bytes는 반드시 log type이고 1이상의 값을 가집니다.

struct msgbuf {
       long mtype;       /* message type, must be > 0 */
       char mtext[1];    /* message data */
};

mtext는 임의의 데이터로 사용이 가능하고요. msgflg는 메시지 전송의 옵션을 지정합니다.

• 0 : 큐에 공간이 생길 때까지 대기.
• IPC_NOWAIT : 큐에 여유 공간이 없으면 즉시 -1 반환.

msgsnd가 성공적으로 동작한다면 앞서 소개한 msqid_ds 구조체의 field가 변경되겠죠. msq_lspid는 호출된 process id로 변경되고 msg_qnum의 값 1증가, msg_stime은 현재시간으로 update됩니다. 

 

18.02.3. msgrcv function

메시지 큐로부터 데이터를 수신하는 함수입니다.

ssize_t msgrcv(int msqid, struct msgbuf *msgp, size_t msgsz, long msgtype, int msgflg);

parameter인 msqid는 다른 function들과 마찬가지로 메시지 큐 id이고 msgp는 메시지 큐에서 읽은 메시지를 저장하는 공간, msgsz는 메시지 저장 공간의 size입니다. 이때 size란 msgbuf의 text size입니다.

msgflg는 메시지가 없는 경우의 옵션을 지정합니다.

• IPC_NOWAIT : 메시지가 없으면 기다리지 않고 -1 반환
• MSG_COPY : Linux 3.8이상부터 지원
• MSG_EXCEPT
• MSG_NOERROR : 메시지 큐의 data가 준비된 data size보다 크면 초과된 부분을 cut하고 읽어 들일 수 있는 부분만 담아옴. 이 option을 선택하지 않는다면 -1 반환하며 실패

msgtyp은 메시지 큐에 있는 자료 중 어떤 자료를 읽을 지에 대한 option입니다.

0 : 첫번째 자료

양수 : 양수로 지정한 값과 같은 data_type의 자료 중 첫번째 자료

음수 : 절대값과 같거나 제일 작은 data_type의 자료

 

18.02.4. msgctl function

메시지 큐의 현재 상태 정보를 참조, 변경할 수 있는 함수입니다.

int msgctl(int msqid, int cmd, struct msqid_ds *buf)

parameter인 cmd는 메시지 큐에 대한 제어 명령입니다.

• IPC_STAT : buf에 메시지 큐 정보 저장.
• IPC_SET : buf값으로 메시지 큐 정보를 설정. msg_pem과 msg_qbytes만 변경 가능
• IPC_RMID : 메시지 큐를 삭제.(buf가 필요없으므로 0(NULL)으로 pass)

buf는 cmd 명령에 따라 동작하는 메시지 큐 객체 structure입니다.

반응형

 

18.03. Message Queue resource check

IPC 메시지 큐의 자원 수

cat /proc/sys/kernel/msgmni

각 메시지의 크기 (Default : 8192)

cat /proc/sys/kernel/msgmax

큐에 있는 메시지의 총 크기 (Default : 16,384)

cat /proc/sys/kernel/msgmnb

 

 

18.04. Kernel's linked list

Generic한 linked list의 경우 구현하고자 하는 구조체에 구조체를 가리키는 pointer 변수를 삽입하여 구현합니다. 하지만 이런 방식은 여러 구조체에 대해 해당하는 linked list를 동적으로 생성해야 하는 번거로움이 존재합니다.

struct list{
	void *data;
	struct list *prev, *next;
};

 

Linux 커널에서는 linked list node를 data안에 넣는 방식으로 구현하는데요.

include/linux/list.h에 관련 macro들이 정의되어 있습니다.

 

한가지 idea를 낸다면 Message Queue 구조를 파악하고 커널 code를 수정한다면 새로운 Message Queue도 구현할 수 있습니다. 예를 들면 priority 기반의 메시지 큐를 구현할 수 있겠죠. 우선순위가 높은 task가 보낸 메시지가 큐의 앞쪽에 배치되도록 변경하게 만들 수 있을 겁니다. 이렇게 기존의 IPC를 발전시키기 위해서는 msgsnd, msgrcv function의 동작을 이해하고 kernel에서 사용하는 struct list_head의 동작을 이해해야 할 것입니다. 물론 priority에 대한 이해도 필요할 것입니다. Linux의 CFS를 예로 들자면 priority가 프로세스의 CPU점유율을 결정할텐데 priority value가 낮은 숫자를 가질수록 더 높은 우선순위를 가지게 됩니다. sys/resource.h에 관련 function들이 있으니 참고해보시면 됩니다.

int getpriority(int which, id_t who);
int setpriority(int which, id_t who, int priority);

위에서 소개한 function에 대해 조금만 더 소개해드리자면 parameter인 which는

• PRIO_PROCESS : 프로세스
• PRIO_PGRP : 프로세스 그룹
• PRIO_USER : 유저

로 설정할 수 있고 who는 which의 ID를 나타냅니다.

 

 

18.05. Example

예제는 다른 site에서 가져왔습니다. reference는 아시겠지만 운영체제 카테고리의 첫 포스팅에 있습니다.

struct real_data{
        short age;
        char name[16];
};
struct message{
        long msg_type;
        struct real_data data;
};

sender.c

#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/msg.h>
#include <stdlib.h>
#include <string.h>
#include "msg_data.h"


void printMsgInfo(int msqid){

        struct msqid_ds m_stat;
        printf("========== messege queue info =============\n");
        if(msgctl(msqid,IPC_STAT,&m_stat)==-1){
                printf("msgctl failed");
                exit(0);
        }
        printf(" message queue info \n");
        printf(" msg_lspid : %d\n",m_stat.msg_lspid);
        printf(" msg_qnum : %d\n",m_stat.msg_qnum);
        printf(" msg_stime : %d\n",m_stat.msg_stime);

        printf("========== messege queue info end =============\n");
}
int main(){
        key_t key=12345;
        int msqid;

        struct message msg;
        msg.msg_type=1;
        msg.data.age=80;
        strcpy(msg.data.name,"REAKWON");

        //msqid를 얻어옴.
        if((msqid=msgget(key,IPC_CREAT|0666))==-1){
                printf("msgget failed\n");
                exit(0);
        }

        //메시지 보내기 전 msqid_ds를 한번 보자.
        printMsgInfo(msqid);

        //메시지를 보낸다.
        if(msgsnd(msqid,&msg,sizeof(struct real_data),0)==-1){
                printf("msgsnd failed\n");
                exit(0);
        }

        printf("message sent\n");
        //메시지 보낸 후  msqid_ds를 한번 보자.
        printMsgInfo(msqid);
}

receiver.c

#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/msg.h>
#include <stdlib.h>
#include <string.h>
#include "msg_data.h"

int main(){
        key_t key=12345;
        int msqid;
        struct message msg;

        //받아오는 쪽의 msqid얻어오고
        if((msqid=msgget(key,IPC_CREAT|0666))==-1){
                printf("msgget failed\n");
                exit(0);
        }
        //메시지를 받는다.
        if(msgrcv(msqid,&msg,sizeof(struct real_data),0,0)==-1){
                printf("msgrcv failed\n");
                exit(0);
        }

        printf("name : %s, age :%d\n",msg.data.name,msg.data.age);

        //이후 메시지 큐를 지운다.
        if(msgctl(msqid,IPC_RMID,NULL)==-1){
                printf("msgctl failed\n");
                exit(0);
        }
}

각각 compile해주시고 실행하시면 됩니다.

ipcs -q

를 통해 queue를 확인하실 수 있습니다. sender와 receiver의 실행횟수를 달리하면서 실행해보시면 메시지 큐에 메시지가 없을때 어떤 동작을 하는지 등을 이해할 수 있습니다.

 

 

오늘은 Message Queue에 대해 알아보았습니다. IPC는 굉장히 재미난 topic인 것 같습니다. 또한 여러가지 idea를 주는 topic이고요. 궁금하신 부분이 있거나 오류가 있는 내용은 댓글남겨주시면 감사하겠습니다. 긴 글 읽어주셔서 감사합니다.