[Unix Programming] Message Passing

Message Passing은 IPC 기법 중 하나로 memory protection을 위해 커널의 도움을 받아 message를 전달한다.

여기서 message는 문자나 byte의 열이라고 생각하면 된다.

Message Passing의 방식은 한 process가 msgsnd를 하게되면 사용자의 주소 공간으로부터 message queue에 message가 저장되고, 다른 process가 msgrcv를 하게되면 message queue에 있는 message를 사용자의 주소 공간으로 가져오게된다. 

 

앞에서 본 것과 같이 Message Passing은 커널에 message queue를 만들어서 통신을 한다. message queue는 msgget 함수를 통해 만들어지며 message의 임시 버퍼로 쓰인다.

 

 

그럼 msgget 함수에 대해 자세히 알아보자.

#include <sys/msg.h>
int msgget(key_t key, int permflags);

먼저 key는 message queue를 식별하는 숫자이다. 이 key값으로 통신 채널을 구분하며 key 값이 다른 message queue를 여러 개 생성할 수 있다. 그 다음 permflags에는 접근권한에 대한 flag들이 들어간다.

msgget 함수의 return 값은 message queue id이며, IPC_EXCL이 설정됐을 때 queue가 이미 존재하면 -1이 반환된다.

 

permflags

- IPC_CREAT: message queue를 생성한다. 기존의 message queue가 있는 경우 덮어쓰지 않는다.

- IPC_EXCL: IPC_CREAT와 동시에 설정된 경우, 한 개의 message queue를 생성한다. 큐가 존재하는 경우엔 실패하며 -1을 return 한다.

 

msgget 함수의 예시를 살펴보자

mqid = msgget((key_t)0100, 0644|IPC_CREAT|IPC_EXCL);

여기서 key값은 0100이고 0644는 접근권한 값이다. (6: owner-r/w, 4: group-r, 4: others-r)

 

 

다음으로 message의 송수신에 필요한 msgsnd와 msgrcv 함수에 대해 알아보자.

#include <sys/msg.h>
int msgsnd(int mqid, const void *message, size_t size, int flags);
int msgrcv(int mqid, void *message, size_t size, long msg_type, int flags);

msgsnd 함수는 mqid가 가리키는 queue에 message를 추가하고, msgrcv는 mqid가 가리키는 queue에서 message를 읽으며, 읽으면 해당 message는 queue에서 제거된다.

 

먼저 mqid는 message queue id를 의미한다. 서로 통신을 하기 위해서 message queue id가 동일해야한다.

그 다음 *message는 송수신할 message의 주소이며 size는 message의 크기를 의미한다.

그리고 msgrcv 함수에서 msg_type이라는 파라미터가 보이는데, 이는 다음에 나오는 message의 구조를 살펴보고 알아보도록 하겠다.

 

flags

- IPC_NOWAIT: non-blocking으로 동작

- MSG_NOERROR: recv시 크기가 크면 절단한다.

 

그 다음 위 함수에서 송수신할 message의 구조에 대해 알아보자.

struct mymsg{
    long mtype;			/* message type */
    char mtext[SOMEVALUE];	/* message text */
};

위에서 본 msgrcv의 msg_type은 message type, 즉 mtype 필드에 따라 어떤 message를 받아들일지 결정한다.

 

msg_type

- 0: queue의 첫 번째 message

- 0보다 큰 값: msg_type과 같은 message

- 0보다 작은 값: 가장 작은 msg_type을 갖는 message

 

 

이제 Message Passing의 예제를 한 번 살펴보자.

먼저 메시지 설비를 위한 q.h라는 헤더파일을 정의한다.

/* q.h -- 메시지 설비 예를 위한 헤더 */

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <errno.h>

#define QKEY (key_t)0105
#define QPERM 0660
#define MAXOBN 50
#define MAXPRIOR 10

struct q_entry {
    long mtype;
    char mtext[MAXOBN+1];
};

 

그 다음은 queue에 객체를 넣는 etest.c 코드이다.

/* etest.c -- queue에 객체 이름을 넣는다. */

#include <stdio.h>
#include <stdlib.h>
#include "q.h"

int enter(char *objname, int priority);
int warn(char *s);
int init_queue(void);

int main(int argc, char **argv) {
    int priority;
    
    if (argc != 3) {
        fprintf(stderr, "usage: %s objname priority\n", argv[0]);
        exit(1);
    }
    if ((priority = atoi(argv[2])) <= 0 || priority > MAXPRIOR) {
        warn("invalid priority");
        exit(2);
    }
    if (enter (argv[1], priority) < 0) {
        warn("enter failure");
        exit(3);
    }
    exit(0);
}

/* enter -- 한 객체를 큐에 넣는다. */
int enter(char *objname, int priority) {
    int len, s_qid;
    struct q_entry s_entry;	/* message를 저장할 구조 */
    
    /* 이름의 길이, 우선순위 수준을 확인한다. */
    if ((len = strlen(objname)) > MAXOBN) {
        warn("name too long");
        return (-1);
    }
    
    if (priority > MAXPRIOR || priority < 0) {
        warn("invalid priority level");
        return (-1);
    }
    
    /* 필요에 따라 message queue를 초기화한다. */
    if ((s_qid = init_queue()) == -1) {
        return (-1);
    }
    
    /* s_entry를 초기화한다. */
    s_entry.mtype = (long)priority;
    strncpy(s_entry.mtext, objname, MAXOBN);
    
    /* message를 보내고, 필요할 경우 기다린다. */
    if (msgsnd(s_qid, &s_entry, len, 0) == -1) {
        perror("msgsnd failed");
        return (-1);
    }
    else {
        return (0);
    }
}

int warn(char *s) {
    fprintf(stderr, "warning: %s\n", s);
}

/* init_queue -- queue identifier를 획득한다 */
int init_queue(void) {
    int queue_id;
    
    /* message queue를 생성하거나 개방하려고 시도한다. */
    if ((queue_id = msgget(QKEY, IPC_CREAT|QPERM)) == -1) {
        perror("msgget failed");
    }
    
    return (queue_id);
}

 

그 다음 queue에 있는 message를 가져올 서버 stest.c 코드이다.

/* stest.c -- queue를 위한 단순한 서버 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "q.h"

int serve(void);
int warn(char *s);
int init_queue(void);
int proc_obj(struct q_entry *msg);

int main(void) {
    int pid;
    
    switch (pid = fork()) {
    case 0:		/* 자식 */
        serve();
        break;
    case -1:		/* 실제로는, 서버는 결코 퇴장(exit)하지 않음 */
        warn("fork to start server failed");
        break;
    default:
        printf("server process pid is %d\n", pid);
    }
    exit(pid != -1 ? 0 : 1);
}

/* serve -- queue에서 가장 우선 순위가 높은 객체를 처리한다. */
int serve(void) {
    int mlen, r_qid;
    struct q_entry r_entry;
    
    /* 필요에 따라 message queue를 초기화한다. */
    if ((r_qid = init_queue()) == -1) {
        return (-1);
    }
    
    /* 다음 message를 가져와 처리한다. 필요하면 기다린다. */
    for (;;) {
        if ((mlen = msgrcv(r_qid, &r_entry, MAXOBN, (-1 * MAXPRIOR), MSG_NOERROR)) == -1) {
            perror("msgrcv failed");
            return (-1);
        }
        else {
            /* 우리가 문자열을 가지고 있는지 확인한다. */
            r_entry.mtext[mlen] = '\0';
            
            /* 객체 이름을 처리한다 */
            proc_obj(&r_entry);
        }
    }
}

int warn(char *s) {
    fprintf(stderr, "warning: %s\n", s);
}

/* init_queue -- queue identifier를 획득한다 */
int init_queue(void) {
    int queue_id;
    
    /* message queue를 생성하거나 개방하려고 시도한다. */
    if ((queue_id = msgget(QKEY, IPC_CREAT|QPERM)) == -1) {
        perror("msgget failed");
    }
    
    return (queue_id);
}

int proc_obj(struct q_entry *msg) {
    printf("\npriority: %ld name: %s\n", msg->mtype, msg->mtext);
}

etest.c 를 컴파일한 후

인자에 객체이름과 우선순위를 넣어 큐에 message를 입력할 수 있다.

그 후 stest.c를 실행시키면 해당 queue에 넣은 값이 작은 우선순위를 가진 message부터 출력된다.

 

 

마지막으로 message queue를 제어하는 msgctl 함수에 대해 알아보자.

#include <sys/msg.h>
int msgctl(int mqid, int command, struct msqid_ds *msq_stat);

msgctl은 다음과 같은 목적으로 쓰인다.

- message queue의 상태정보 획득

- message queue에 관련된 제한 변경

- queue 제거

command를 통해 앞에서 말한 함수의 목적을 설정할 수 있으며, msqid_ds 구조체에는 message queue의 상태정보가 들어간다.

 

command

- IPC_STAT: message queue의 상태정보를 msq_stat에 저장

- IPC_SET: message queue에 대한 제어변수들의 값을 지정

- IPC_RMID: message queue의 삭제

 

msqid_ds

- msg_perm: 접근권한

- msg_qnum: queue number

- msg_qbytes: queue에 저장되어 있는 message의 bytes

- lsg_lspid: 마지막으로 송신한 pid

- msg_lrpid: 마지막으로 수신한 pid

- msg_stime: message send time

- msg_rtime: message recv time

- msg_ctime: message create time

 

 

다음은 msgctl의 예시이다.

/* showmsg -- message queue의 자세한 정보를 보인다. */

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

void mqstat_print(key_t, int, struct msqid_ds *);

int main(int argc, char **argv) {
    key_t mkey;
    int msq_id;
    struct msqid_ds msq_status;
    
    if (argc != 2) {
        fprintf(stderr, "usage: showmsg keyval\n");
        exit(1);
    }
    
    /* message queue identifier를 얻는다 */
    mkey = (key_t)atoi(argv[1]);
    if ((msq_id = msgget(mkey, 0)) == -1) {
        perror("msgget failed");
        exit(2);
    }
    
    /* 상태 정보를 얻는다 */
    if (msgctl(msq_id, IPC_STAT, &msq_status) == -1) {
        perror("msgctl failed");
        exit(3);
    }
    
    /* 상태 정보를 프린트한다 */
    mqstat_print(mkey, msq_id, &msq_status);
    exit(0);
}

void mqstat_print(key_t mkey, int mqid, struct msqid_ds *mstat) {
    printf("\nKey %d, msg_qid %d\n\n", mkey, mqid);
    printf("%d message(s) on queue\n\n", mstat->msg_qnum);
    printf("Last send by proc %d at %s\n", mstat->msg_lspid, ctime(&(mstat->msg_stime)));
    printf("List recv by proc %d at %s\n", mstat->msg_lrpid, ctime(&(mstat->msg_rtime)));
}