[프로세스와 스레드 다중 처리] POSIX 스레드와 동기화

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

 

⭐ POSIX 스레드

여러가지 일을 동시에 처리하고 싶은 경우에는 멀티 프로세스를 이용한다

프로세스는 자신만의 메모리 공간을 가지고 있고, 새로운 프로세스를 만들 때마다 프로세스를 위한 메모리 공간을 따로 만들어줘야 하기 때문에 생성하는 작업이 복잡하다

동일한 작업을 수행하는 경우에도 공유되는 부분 없이 별도의 영역을 사용하기 때문에 메모리 공간이 낭비된다는 문제가 있다

 

스레드(Thread) 는 프로세스에서 나누어지는 일의 단위로 커널 레벨의 스레드와 유저 레벨의 스레드로 구분된다

일반적으로 애플리케이션에서는 유저 레벨의 스레드를 사용하는데, 유저 레벨의 스레드는 메모리를 다른 스레드와 공유할 수 있기 때문에 생성이 빠르며 메모리의 낭비도 막을 수 있다

하지만 스레드가 메모리를 잘못 건드리면 메모리를 공유하고 잇기 때문에 다른 스레드들에도 영향을 미친다

 

스레드 간에는 전역 변수와 힙 메모리, 파일 디스크립터 등을 공유하고 있으며, 멀티 프로세스와 마찬가지로 몇 가지 일을 동시에 할 수 있지만 스레드 간 콘텍스트 스위칭 (Context Switching) 이 프로세스보다 빠르고, 공유되는 메모리를 이용해서 스레드 간 통신이 프로세스 간 통신보다 간단하다

 

☑️ 스레드의 생성

스레드를 생성하기 위해서는 pthread_create() 함수

스레드로 하고 싶은 일을 하나의 함수로 작성한 후 함수 포인터를 이용해서 등록하기

첫 번째 인자(pthread_t) 는 스레드를 구분하기 위한 ID 로 뒤의 스레드 함수에서 사용하기 위해 필요한 ID

두 번째 인자(attr) 은 스레드를 생성할 때 필요한 속성의 값으로 일반적으로 NULL 사용

리눅스에서 스레드는 합류 가능하고 비실시간 스케줄 정책으로 생성된다

세 번째 인자(start_rountine) 는 스레드로 해야 할 일을 정의해높은 함수

일반적으로 while, for 루프로 원하는 일을 계속해서 실행하도록 한다

네 번째 인자(arg) 는 세 번째 인자에서 사용한 start_routine 함수로 넘겨주는 값이다

 

이 함수의 호출에 성공하면 첫 번째 인자에 pthread_t 형의 현재의 스레드에 대한 ID 를 설정하고 0 반환

프로세스를 생성한 후 자식 프로세스를 대기해야 하는 것처럼 스레드도 종료해야할 때까지 대기해야 한다

서브 스레드가 실행되고 있을 때는 메인 스레드가 종료되지 않도록 해야 하는데, 이때 pthread_join() 함수 사용한다

첫 번째 인자로 pthread_create() 에서 설정된 스레드의 ID

두 번째 인자로 스레드를 동작하는 함수(start_routine) 에서 반환되는 값을 받기 위해 사용한다

 

main() 함수로 시작되는 메인 스레드가 종료되면 서브 스레드들도 함께 종료된다

메인 스레드에서 pthread_create() 를 이용해 생성된 스레드를 분리하고 싶은 경우에 pthread_detach() 함수를 사용할 수 있다

분리된 스레드가 종료될 경우 pthread_join() 함수를 호출하지 않아도 모든 자원을 해제한다

 

스레드를 종료하거나 취소하고 싶은 경우네는 pthread_exit() 함수나 pthread_cancle() 함수 사용

스레드 자신에 대한 ID 를 알고 싶은 경우에 pthread_self() 함수 사용

// 간단한 스레드 (posix 세마포어 함수를 스레드를 이용해서 변경하기)
#include <stdio.h> 		/* printf() 함수를 위한 헤더 파일 */
#include <unistd.h>
#include <fcntl.h> 		/* O_CREAT, O_EXEC 매크로를 위한 헤더 파일 */
#include <pthread.h>
#include <semaphore.h> 		/* sem_open(), sem_destroy(), sem_wait() 등 함수를 위한 헤더 파일 */

sem_t *sem; 			/* 세마포어를 위한 전역 변수 */
static int cnt = 0; 		/* 세마포어에서 사용할 임계 구역 변수 */

void p() 			/* 세마포어의 P 연산 */
{
    sem_post(sem);
}

void v() 			/* 세마포어의 V 연산 */
{
    sem_wait(sem);
}

void *ptheadV(void *arg) 	/* V 연산을 수행하기 위한 함수를 작성한다. */
{
    int i;

    for(i = 0; i < 10; i++) {
        if(cnt >= 7) usleep(100); 	/* 7 이상이면 100밀리초 동안 대기한다. */
        cnt++;
        printf("increase : %d\n", cnt) ;
        fflush(NULL);
        v();
    }

    return NULL;
}

void *ptheadP(void *arg) 	/* P 연산을 수행하기 위한 함수를 작성한다. */
{
    int i;

    for(i = 0; i < 10; i++) {
        p(); 			/* 세마포어가 0이 되면 블록된다. */
        cnt--;
        printf("decrease : %d\n", cnt);
        fflush(NULL);
        usleep(100); 		/* 100밀리초 간 기다린다. */
    }

    return NULL;
}

int main(int argc, char **argv)
{
    pthread_t ptV, ptP; 	/* 스레드를 위한 자료형 */
    const char* name = "posix_sem";
    unsigned int value = 7; 	/* 세마포어의 값 */

    /* 세마포어 열기 */
    sem = sem_open(name, O_CREAT, S_IRUSR | S_IWUSR, value);
    pthread_create(&ptV, NULL, ptheadV, NULL); 	/* 스레드 생성 */
    pthread_create(&ptP, NULL, ptheadP, NULL);

    pthread_join(ptV, NULL); 	/* 스레드가 종료될 때까지 대기 */
    pthread_join(ptP, NULL);

    /* 다 쓴 세마포어 닫고 정리 */
    sem_close(sem);
    printf("sem_destroy() : %d\n", sem_destroy(sem));

    /* 세마포어 삭제 */
    sem_unlink(name);

    return 0;
}

/*
메인 스레드 이외의 2개의 스레드가 번갈아가면서 실행된다
첫 번째 스레드를 만들고 값을 증가시키고, 두 번째 스레드를 만들고 나서 값을 감소시킨다
위의 값을 증가시키는 곳에서는 7 이상이면 100밀리초 동안 정지하고,
값을 감소시키는 곳에서도 100밀리초 동안 정지한다

스레드를 사용할 때 잠시 정지하지 않고 계속 실행하게 되면 다른 스레드가 늦어질 수 있으므로
usleep() 함수를 적절히 사용해서 일의 중간에 잠시 정지하도록 하는 것이 좋다

빌드하기 위해 pthread 라이브러리를 링크해야 한다
gcc는 -pthread를 지원하고 있지만, 다른 컴파일러를 위해서 -l 옵션 이용해서 pthread 라이브러리와 링크할 수 있도록 설정한다
빌드 후 실행해보면 값이 증가하고 난 이후에 값이 계쏙해서 감소되는데 감소하는 시점에서 세마포어가 0인 경우에는
값이 증가되기 전까지 semop() 함수에서 대기하고 있는 것을 확인한다
스레드는 실행할 때마다 실행 순서가 바뀔 수 있으므로 결과값이 다르게 나올 수 있다

gani@gani:~/raspi/Process_Signal $ gcc -o thread thread.c -lpthread
gani@gani:~/raspi/Process_Signal $ ./thread 
*/

gcc -o thread thread.c -lpthread
./thread

 

⭐ 동기화

스레드는 프로세스와 다르게 스레드 간에 메모리를 공유하기 때문에 문제가 발생한다

이를 경쟁 조건 (Race Condition)

하나의 스레드에서 전역 변수에 대한 처리가 완료되기 전에 다른 스레드의 접근을 제한해야 하는데, 스레드 내에서 자원에 대해 보호가 필요한 구역을 임계 구역 (Critical Section) 

제한된 자원에 대한 임의적 접근을 수행해야 하는데, 경쟁 조건을 해결하기 위해 시그널, IPC 등을 이용한 프로세스 간의 통신을 이용해서 접근 제어를 할 수 있다

보통 자원의 잠금 (lock) 을 이용하는데 세마포어나 뮤텍스를 이용할 수 있다

한번에 하나의 스레드만 순차적으로 처리하는 것을 동기화 (Serialization)

 

☑️ 뮤텍스 (Mutex)

뮤텍스는 세마포어와 공유되는 메모리에 하나의 스레드가 사용되는 동안 다른 스레드들의 접근을 막아준다

스레드의 동기화 문제를 해결할 수 있다

 

뮤텍스는 스레드들이 임계 영역에 들어갈 때 걸어 잠그고 나갈 때 풀어주는 방식으로 동기화를 수행하는데, 뮤텍스는 세마포어의 일종으로 자원의 수가 하나인 경우라고 볼 수 있다

뮤텍스를 사용하기 위해서는 관련 초기화가 필요한데 pthread_mutex_init() 함수를 사용한다

 

뮤텍스를 생성하고 난 이후에 해당 뮤텍스에 대한 ID 가 pthread_mutex_t 형으로 설정된다

이 값을 이용해서 뮤텍스를 잠그고 해제하면 되는데 잠글 때는 pthread_mutex_lock() 함수를 사용

잠금을 해제할 때는 pthread_mutex_unlock() 함수를 사용

 

pthread_mutex_lock() 함수는 다른 스레드가 뮤텍스를 이미 잠근 상태에서는 그 잠금이 해제될 때까지 계속 대기하고 있는데, 알고리즘이 잘못된 경우 애플리케이션이 무한정 대기하고 있는 문제점이 있다

pthread_mutex_trylock() 함수는 다른 스레드가 뮤텍스를 이미 잠금 상태에서는 잠겨있는지만 확인하고, 실행할 수 없으면 블록되지 않고 다음 코드를 실행한다

단순히 뮤텍스가 잠겨있는지 확인할 때 유용하게 사용할 수 있다

 

세마포어에서 자원의 수가 하나인 경우를 바이너리 세마포어

자원의 수가 여러 개인 경우를 카운팅 (counting) 세마포어

 

무조건 대기하는 것이 아니라 특정 시간만 대기하는 경우에는 pthrad_mutex_timedlock() 함수

pthread_mutex_trylock() 함수의 경우는 대기하지 않고 그냥 넘어가고, pthread_mutex_lock() 함수는 계속 대기하는데 일정 시간 만큼 대기하면 대기 사항에 대한 처리를 보다 유연하게 처리한다

뮤텍스의 사용이 다 끝나면 pthread_mutex_destory() 함수를 이용해서 뮤텍스를 해제한다

// 뮤텍스로 동기화 이용하기
#include <stdio.h>
#include <pthread.h>

int g_var = 1;
pthread_mutex_t mid;

void *inc_function(void *);
void *dec_function(void *);

int main(int argc, char **argv)
{
    pthread_t ptInc, ptDec;

    pthread_mutex_init(&mid, NULL); 	/* 뮤텍스 초기화 */

    pthread_create(&ptInc, NULL, inc_function, NULL);

    pthread_create(&ptDec, NULL, dec_function, NULL);
    pthread_join(ptInc, NULL);
    pthread_join(ptDec, NULL);

    pthread_mutex_destroy(&mid); 	/* 뮤텍스 삭제 */

    return 0;
}

void *inc_function(void *arg)
{
    pthread_mutex_lock(&mid); 		/* 임계 구역 설정 */
    printf("Inc : %d < Before\n", g_var);
    g_var++;
    printf("Inc : %d > After\n", g_var);
    pthread_mutex_unlock(&mid); 	/* 임계 구역 해제 */

    return NULL;
}

void *dec_function(void *arg)
{
    pthread_mutex_lock(&mid); 		/* 임계 구역 설정 */
    printf("Dec : %d < Before\n", g_var);
    g_var--;
    printf("Dec : %d > After\n", g_var);
    pthread_mutex_unlock(&mid); 	/* 임계 구역 해제 */

    return NULL;
}

/*
뮤텍스를 2개의 스레드에서 모두 사용할 수 있도록 전역 변수로 선언
pthread_mutex_init() 함수를 통해서 초기화 후 사용이 긑나면
pthread_mutex_destory() 함수를 불러서 해제한다

뮤텍스의 초기화가 완료되면 2개의 스레드 함수에서 임계 구역 (값을 출력한 후 값을 변화시키고 다시 변경된 값을 출력) 옆에
pthread_mutex_lock() 함수를 추가하고 뒷부분에 pthread_mutex_unlock() 함수를 추가하여 보호하도록 한다

증가 부분의 코드가 완료되고 난 이후에 감소 부분이 실행되거나
반대로 감수 부분의 실행이 완료된 이후에 증가 부분이 실행되는 것을 확인한다

gani@gani:~/raspi/Process_Signal $ gcc -o thread_mutex thread_mutex.c -lpthread
gani@gani:~/raspi/Process_Signal $ ./thread_mutex 
*/

gcc -o thread_mutex thread_mutex.c -lpthread
./thread_mutex

 

POSIX 에서 기본적인 뮤텍스 이외에도 읽기, 쓰기, 잠금도 지원한다

데이터의 변경이 필요없는 경우에 읽기 잠금 (Read Lock) 을 사용할 수 있고, 데이터의 쓰기가 필요 없는 경우 쓰기 잠금을 사용할 수 있다

 

☑️ 스레드와 시그널

시그널은 스레드 간에 공유된다

특정 시그널이 발생하면 프로세스의 모든 스레드에 전달되는데, 이때 스레드에 전달되는 시그널을 조정하고 싶으면 pthread_sigmask() 함수를 사용할 수 있다

int pthread_sigmask(int how, const sigset_t *newmask, sigset_t *oldmask);

sigset_t 타입으로 시그널 마스크를 생성한 후 pthread_sigmask() 함수를 이용해서 설정한다

pthread_sigmask() 함수는 현재의 스레드에 첫 번째 인자 (how) 와 두 번째 인자 (newmask) 를 이용해서 시그널 마스크를 설정한다

how 는 SIG_BLOCK, SIG_UNBLOCK, SIG_SETMASK 중 하나를 사용할 수 있는데, SIG_BLOCK 는 현재 설정된 시그널 마스크에 newmask 를 추가하며 SIG_UNBLOCK 는 현재 설정된 시그널 마스크에서 newmask 를 제거하고 SIG_SETMASK 는 newmask 로 현재 시그널 마스크를 설정한다