[프로세스와 스레드 다중 처리] 프로세스 간 통신 (IPC) with 유닉스 시스템 V/XSI (메시지 큐, 세마포어, 공유 메모리)

<서영진 님의 사물인터넷을 위한 리눅스 프로그래밍 with 라즈베리파이 서적을 참고해서 작성하였습니다 :-)>

 

⭐ 유닉스 시스템 V

AT&T 벨 연구소에서 만들던 유닉스 계열을 통칭함

현재 리눅스는 유닉스 시스템 V 계열의 유닉스에서 제공하는 시스템 호출과 BSD 계열의 유닉스에서 제공하던 시스템 호출을 모두 제공한다

 

시스템 V의 IPC 는 POSIX IPC 가 등장한 이후로 XSI (X System Interface) 로 불리우고 있다

구분	내용
메시지 큐 (Message Queue)	메시지 단위로 I/O 를 수행하는데, 다른 IPC 와 달리 서버와 클라이언트가 같은 시간에 존재할 필요가 없다
세마포어 (Semaphore)	데이터 전송을 목적으로 만들어진 IPC 가 아닌 프로세스 동기화에 필요한 공유 변수와 같은 공유된 리소스 제한을 위한 세마포어의 참조와 조직을 위해 만들어진 IPC
공유 메모리 (Shared Memory)	모든 IPC 는 빈번한 메모리 복사로 인한 성능 문제가 제기되는데, 공유 메모리는 커널과 프로세스 사이에 메모리 복사를 2회 줄임으로써 비약적으로 IPC 성능을 향상시킬 수 있다 공유 메모리에는 세마포어 같은 동기화 메커니즘이 별도로 요구된다

 

☑️ 메시지 큐

유닉스  시스템 V의 프로세스 사이에서 서로 통신을 하기 위해 사용되는 메커니즘

입출력 데이터를 커널 내부에 메시지 리스트 형태로 보관하는데, 메시지 타입은 선택적으로 메시지의 검색이 가능하다

FIFO 와 비슷하지만 메시지는 중간에 저장이 되기 때문에 메시지를 보내는 프로세스와 메시지를 받는 프로세스가 동일한 시점에 주고받을 필요가 없다

 

메시지 큐에서는 msqid_ds 구조체를 사용하며, 호출에 실패하면 -1을 반환한다

메시지 큐를 사용하기 위해서는 현재 메시지를 주고받을 수 있는 채널이 필요한데 msgget() 함수를 이용해서 획득, 호출에 성공하면 채널의 ID 가 반환된다

 

메시지 큐에 대한 채널을 획득한 후에는 데이터를 주고받아야 하는데,

메시지 보내는 msgsnd() 함수 호출에 성공하면 0 반환

메시지 받는 msgrcv() 함수 호출에 성공하면 읽은 데이터의 바이트 수가 반환된다

데이터 주고 받는 msgbuf 구조체

채널을 제어하거나 삭제하는 msgctl() 함수

// 메시지 큐를 이용한 통신 server
// 서버와 클라이언트는 동일한 키 값 가져야 한다 (51234 로 설정함)
#include <stdio.h>
#include <unistd.h>
#include <sys/msg.h>

#define MSQKEY 51234

struct msgbuf {
    long mtype; /* 메시지의 타입 : 0 이상의 정숫값 */
    char mtext[BUFSIZ]; /* 메시지의 내용 : 1바이트 이상의 문자열 */
};

int main(int argc, char **argv)
{
    key_t key;
    int n, msqid;
    struct msgbuf mb;
    key = MSQKEY;

    /* 메시지 큐의 채널을 생성한다. */
    if((msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666)) < 0) {
        perror("msgget()");
        return -1;
    }

    /* 메시지 큐에서 데이터를 가져온다. */
    while((n = msgrcv(msqid, &mb, sizeof(mb), 0, 0)) > 0) {
        switch (mb.mtype) {
            /* 메시지 타입(mtype)이 1이면 화면에 가져온 데이터를 출력한다. */
            case 1:
                write(1, mb.mtext, n);
                break;
            /* 메시지 타입(mtype)이 2이면 메시지 큐의 채널을 삭제한다. */
            case 2:
                if(msgctl(msqid, IPC_RMID, (struct msqid_ds *) 0) < 0) {
                    perror("msgctl()");
                    return -1;
                }
                break;
        }
    }

    return 0;
}

/*
서버는 메시지 큐를 위한 채널을 생성하고 메시지 큐로부터 데이터를 가져와서 mtype 값이 1이면 화면에 가져온 데이터 출력
mtype 값이 2이면 메시지 큐에 대한 채널을 삭제한다

@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ gcc -o msg_server msg_server.c
@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ ./msg_server &
*/

// 메시지 큐를 이용한 통신 client
// 서버와 클라이언트는 동일한 키 값 가져야 한다 (51234 로 설정함)
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/msg.h>

#define MSQKEY 51234

struct msgbuf {
    long mtype;
    char mtext[BUFSIZ];
};

int main(int argc, char **argv)
{
    key_t key;
    int rc, msqid;
    char* msg_text = "hello world\n";

    struct msgbuf *mb;
    mb = (struct msgbuf*)malloc(sizeof(struct msgbuf) + strlen(msg_text));

    key = MSQKEY;
    if((msqid = msgget(key, 0666)) < 0) { 		/* 메시지 큐의 채널을 가져온다. */
        perror("msgget()");
        return -1;
    }

    /* mtype을 1로 설정하고 hello world라는 문자열을 보낸다. */
    mb->mtype = 1;
    strcpy(mb->mtext, msg_text);
    rc = msgsnd(msqid, mb, strlen(msg_text)+1, 0); 	/* 메시지 큐로 데이터를 보낸다. */
    if(rc == -1) {
        perror("msgsnd()");
        return -1;
    }

    /* mtype을 2로 설정하고 보낸다. */
    mb->mtype = 2;
    memset(mb->mtext, 0, sizeof(mb->mtext));
    if(msgsnd(msqid, mb, sizeof(mb->mtext), 0) < 0) {
        perror("msgsnd()");
        return -1;
    }

    return 0;
}

/*
클라이언트는 mtype 을 1로 설정하고 문자열을 서버로 보내고
다시 mtype 을 2로 설정하고 서버로 메시지를 보낸다

@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ gcc -o msg_client msg_client.c
@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ ./msg_client 
*/

gcc -o msg_server msg_server.c
./msg_server &

gcc -o msg_client msg_client.c
./msg_client

 

☑️ IPC 관련 명령어

유닉스 시스템 v의 설정된 IPC 채널을 확인하고 싶은 경우에는 ipcs 명렁어 사용한다

-q : 메시지 큐

-s : 세마포어

-m : 공유 메모리

ipcs [-q | -m | -s]

 

채널을 생성하고 싶은 경우

ipcmk [-M size] [-S nsems] [-Q] [-p mode]

 

설정된 채널을 삭제하고 싶은 경우

채널의 ID 나 채널의 키 이용해서 삭제할 수 있음

ipcrm [-q | -m | -s { 채널의_ID }+ ] [-Q | -M | -S] { 채널의_Key }+ ]

 

메시지큐 서버 실행한 후 ipcs 명령어 수행하면 키와 메시지 큐의 ID 가 표시된다

-q 옵션과 메시지 큐의 ID 이용해서 해당 채널 삭제할 수 있다

ipcs

ipcrm -q msqid적으셈

 

☑️ 세마포어 (Semaphore)

두 개의 원자적 함수로 조작되는 정수 변수

다중 프로그래밍 환경에서 공유 자원에 대한 접근을 제한하는 방법

다수의 자원을 놓고 경쟁하는 프로세스들 사이에서 필요한 동기화 기법으로 자원의 수를 기록하는 세마포어 변수와 P (감소), V (증가) 연산을 이용하여 동기화 한다

 

세마포어 자료 구조는 semid_ds 구조체

세마포어를 사용하기 위해 채널 생성 semget() 함수 수행하며, 성공하면 채널의 ID 반환되고 실패하면 -1

세마포어를 증가하거나 감소를 하기 위해 semop() 함수 수행하며, 성공하면 0 반환되고 실패하면 -1

세마포어를 설정할 때 sembuf 구조체 사용

세마포어를 제어하거나 삭제할 때에는 semctl() 함수

// 세마포어를 수행하기 위한 과정
// 세마포어의 P 연산과 V 연산에 대한 함수
// 두 함수를 이용해서 세마포어의 동작 알아보기
#include <stdio.h>
#include <unistd.h>
#include <sys/sem.h>

int cnt = 0;
static int semid;

void p() 				/* 세마포어의 P 연산 */
{
    struct sembuf pbuf;
    pbuf.sem_num = 0;
    pbuf.sem_op = -1;
    pbuf.sem_flg = SEM_UNDO;

    if(semop(semid, &pbuf, 1) == -1) 	/* 세마포어의 감소 연산을 수행한다. */
        perror("p : semop()");
}

void v() 				/* 세마포어의 V 연산 */
{
    struct sembuf vbuf;
    vbuf.sem_num = 0;
    vbuf.sem_op = 1;
    vbuf.sem_flg = SEM_UNDO;

    if(semop(semid, &vbuf, 1) == -1) 	/* 세마포어의 증가 연산을 수행한다. */
        perror("v : semop()");
}

int main(int argc, char **argv)
{
    union semun { 			/* semun 공용체 */
        int val;
        struct semid_ds *buf;
        unsigned short int *arrary;
    } arg;

    /* 세마포어에 대한 채널 얻기 */
    if((semid = semget(IPC_PRIVATE, 1, IPC_CREAT | 0666)) == -1) {
        perror("semget()");
        return -1;
    }

    arg.val = 1; 			/* 세마포어 값을 1로 설정 */
    if(semctl(semid, 0, SETVAL, arg) == -1) {
        perror("semctl() : SETVAL");
        return -1;
    }

    while(1) {
        if(cnt >= 8) {
            cnt--;
            p();
            printf("decrease : %d\n", cnt);
            break;
        } else {
            cnt++;
            v();
            printf("increase : %d\n", cnt);
            usleep(100);
        }
    }

    /* 세마포어에 대한 채널 삭제 */
    if(semctl(semid, 0, IPC_RMID, arg) == -1) {
        perror("semctl() : IPC_RMID");
        return -1;
    }

    return 0;
}

/*
세마포어를 위한 채널을 생성하고 세마포어 한 개를 생성한다
코드 내에서 세마포어에 대한 P연산과 V연산을 수행하는데,
V연산으로 세마포어가 증가되고 난 이후에 P연산을 통해 감소될 수 있다

두 연산의 실행 순서를 바꾸면 P연산에서는 사용할 수 있는 세마포어가 생길 때까지 대기한다
8개까지 세마포어가 증가한 후 다시 세마포어가 감소하는 것을 확인할 수 있다

@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ gcc -o sem sem.c
@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ ./sem
*/

gcc -o sem sem.c
./sem

 

☑️ 공유 메모리

프로세스 간에 메모리를 공유하는 방식

메리 복사 횟수를 줄여서 성능 향상을 이끈다

여러 프로세스들이 동시에 같은 메모리의 주소로 접근할 수 있기 때문에 세마포어 같은 동기화 기법이 필요하다

리눅스에서는 공유 메모리를 위한 함수들을 제공한다

공유 메모리를 사용하기 위해서는 공유 메모리에 대한 채널을 생성하거나 기존의 채널을 가져와야 한다

shmget() 함수로 수행에 성공하면 해당 채널에 대한 ID 반환, 에러가 발생하면 -1 반환

공유 메모리의 사용을 위해서는 공유 메모리를 현재 프로세스와 연결하고 메모리의 시작 주소를 가져오는데 shmat() 함수를 사용한다

사용이 끝난 공유 메모리는 현재의 프로세스와의 연결을 해제해야 하는데 shmdt() 함수를 사용한다

연결된 채널을 제어하거나 해제하고 싶은 경우에는 shmctl() 함수이며 성공하면 0, 실패하면 -1을 반환한다

// 공유 메모리
// 2개의 프로세스를 만들어서 부모 프로세스와 자식 프로세스가 같은 메모리 공간 사용하기
#include <stdio.h>
#include <unistd.h>
#include <sys/shm.h>

#define SHM_KEY 0x12345 		/* 공유 메모리를 위한 키 */

int main(int argc, char **argv)
{
    int i, pid, shmid;
    int *cVal;
    void *shmmem = (void *)0;

    if((pid = fork()) == 0) { 		/* 자식 프로세스를 위한 설정 */
        /* 공유 메모리 공간을 가져온다. */
        shmid = shmget((key_t)SHM_KEY, sizeof(int), 0);
        if(shmid == -1) {
            perror("shmget()");
            return -1;
        }

        /* 공유 메모리를 사용하기 위해 프로세스의 메모리에 붙인다. */
        shmmem = shmat(shmid, (void *)0, 0666 | IPC_CREAT);
        if(shmmem == (void *)-1) {
            perror("shmat()");
            return -1;
        }

        cVal = (int *)shmmem;
        *cVal = 1;
        for(i = 0; i < 3; i++) {
            *cVal += 1;
            printf("Child(%d) : %d\n", i, *cVal);
            sleep(1);
        }
    } else if(pid > 0) { 		/* 부모 프로세스로 공유 메모리의 내용을 표시 */
        /* 공유 메모리 공간을 만든다. */
        shmid = shmget((key_t)SHM_KEY, sizeof(int), 0666 | IPC_CREAT);
        if(shmid == -1) {
            perror("shmget()");
            return -1;
        }

        /* 공유 메모리를 사용하기 위해 프로세스의 메모리에 붙인다. */
        shmmem = shmat(shmid, (void *)0, 0);
        if(shmmem == (void *)-1) {
            perror("shmat()");
            return -1;
        }

        cVal = (int *)shmmem;
        for(i = 0; i < 3; i++) {
            sleep(1);
            printf("Parent(%d) : %d\n", i, *cVal);
        }
    }

    shmctl(shmid, IPC_RMID, 0);

    return 0;
}

/*
2개의 프로세스를 생성한 후 shmget() 함수를 이용해서 공유 메모리 대한 채널을 가져와서
shmat() 함수를 이용해 공유 메모리를 현재의 프로세스의 공간에 붙인다

자식 프로세스에서는 변수의 값을 증가시키고 이를 화면에 출력시키고 1초간 슬립 모드
부모 프로세스는 1초간 슬립 모드에 들어간 후 깨어나서 공유 메모리에 대한 내용을 출력한다
자식 프로세스에서 공유 메모리로 설정해놓은 내용을 부모 프로세스에서도 같은 메모리를 공유하여 동일한 값이 출력된다

@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ gcc -o shm shm.c
@gani0325 ➜ /workspaces/IoT_Linux_RaspberryPi/Process_Signal (main) $ ./shm
*/

gcc -o shm shm.c
./shm