[Linux, OS] Lock, Mutex | Multithread Synchronization | Pipe, FIFO | POSIX Message Queue

2023.11.15 TIL

Code, Image source: The Linux Programming Interface, Michael Kerrisk

Lock, Mutex

Mutex, Lock의 단점

• 속도 저하 (병렬 처리에 방해된다)

Mutex 초기화

• 정적인 방법

  static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;  // lock으로 초기화

• 동적인 방법

  ```c
  #include <pthread.h>
  
  int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexettr_t *attr);
  int pthread_mutex_init(pthread_mutex_t *mutex);
  
  ```

  
  

Mutex 사용

#include <pthread.h>

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

Global variable 문제점 해결

• tlpi-dist/threads/thread_incr_mutex.c
  

상태 변화 알리기

• Thread 간의 signal이라고 생각하면 된다
• 조건 변경 → 다른 thread에게 변경 여부를 알려주고 싶다면?

  1. avail 변경 여부를 계속 확인(polling)

     static int avail = 0;
     
     // Thread a
     // 변경여부 계속 확인
     while (avail > 0)
     	avail--;
     
     // Thread b
     // 변경하기
     avail++;

     → 문제점: while을 계속 돈다… CPU 계속 차지함

  2. Condition Variable 사용

     • tlpi-dist/threads/prod_condvar.c

     static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;  // 초기화
     
     int pthread_cond_signal(pthread_cond_t *cond);  // 1개의 thread(sleeping consumer)만 깨운다
     int pthread_cond_broadcast(pthread_cond_t *cond);  // 모든 thread(sleeping consumer)를 깨운다
     int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);  // sleep...

     → sleeping consumer(waiting thread)는 자고있다가, 필요할때 깨워지면(avail > 0)

     

Deadlock

• 순서를 지키지 않아 둘다 sleep → 아무도 깨워줄수가 없다…
• Thread A

  pthread_mutex_lock(mutex1);  // (1.1) 
  pthread_mutex_lock(mutex2);  // (2.1) block

• Thread B

  pthread_mutex_lock(mutex2);  // (1.2) Thread A sleeps
  pthread_mutex_lock(mutex1);  // (2.2) block

Multithread Synchronization

Producer and Consumer

• Buffer/Queue 사용
• Producer thread와 consumer thread가 queue를 공유
• Producer는 queue에 data를 넣고 다른 일 가능
• Consumer는 data를 기다리고 있을 필요 없이 다른 일 하다가 queue만 확인

C와 C++ code 연동

• C로 구현시 개발 생산성 떨어지지만 기존 코드가 이미 C인경우

  • 특정 모듈을 C++ 사용
  • Process로 빼(Rust/Go로 개발) 통신은 RPC/IPC

• extern “C”

  #ifdef _cplusplus
  	extern "C" {
  #endif
  	// C++ code
  #ifdef _cplusplus
  	}
  #endif

Pipe, FIFO

Process간 통신

• 통신 (Communication)
• 동기화 (Syncronization)
• 시그널 (Signal)

Communication 방법

• Data 전송
  • Byte stream
  • Message
• Shared Memory
  • System V shared memory
  • POSIX shared memory
  • Memory mapping

Pipe

• IPC를 위한 기법
  
• ls | wc -l → ls의 결과물이 input으로 wc로 들어간다
  • ls → stdout (fd 1)
  • pipe로 밀어넣는다
  • stdin (fd 0) → wc

Pipe 생성

#include <unistd.h>

int pipe(int filedes[2]);

• 두개의 File Desciptor를 생성
  • read를 위한 FD
  • write를 위한 FD

Parent and Child Process

• fork() → Parent의 fd 그대로 가져옴
  • Parent가 filedes[1]에 write, child는 filedes[0]에서 read
• tlpi-dist/pipes/simple_pipe.c

popen()

• 성능, 보안 문제
• 빠르게 test program, prototype system 작성시 유용

#include <stdio.h>

FILE *popen(const char *command, const char *mode);  // shell script command, read/write mode
int pclose(FILE *stream);

• tlpi-dist/pipes/popen_glob.c

FIFOs

• mkfifo [-m mode] [pathname]

#include <sys/stat.h>

int mkfifo(const char *pathname, mode_t mode);

$ mkfifo myfifo  # make fifo
$ wc -l < myfifo &  # wc는 input(myfifo의 output)을 read 하기위해 기다림, background로 실행
$ ls -l | tee myfifo | sort -k5n  # tee: 결과가 T 관을 통해 myfifo로도 나가고 sort(stdout)로도 간다

POSIX Message Queue

Inter Process Communication

• Multithread/Multiprocess Programming
• Message size 문제
  • Message queue의 정의된 size만큼 항상 할당 → 낭비 발생
  • Shared memory 사용
    

Message Queue 생성

• Message queue 생성 또는 이미 있는 queue open

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

mqd_t mq_open(const char *name, int ofalg, ... /* mode_t mode, struct mq_attr *attr */ );

• Creation Options
  • O_CREAT → Create queue if it doesn’t already exist
  • O_EXCL → With O_CREAT, create queue exclusively
  • O_RDONLY → Open for reading only
  • O_WRONLY → Open for writing only
  • O_RDWR → Open for reading and writing
  • O_NONBLOCK → Open in nonblocking mode
• tlpi-dist/pmsg/pmsg_create.c

Message Queue 제거

#include <mqueue.h>

int mq_close(mqd_t mqdes);  // close message queue descriptor
int mq_unlink(const char *name);  // delete file

Message Queue 속성

struct mq_attr {
	long mq_flags;  // options
	long mq_maxmsg;  // message 최대 개수
	long mq_msgsize;  // message 최대 size
	long mq_curmsgs;  // 현재 message queue에 있는 message 개수
};

• tlpi-dist/pmsg/pmsg_getattr.c

Message 송신

#include <mqueue.h>

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);  // mq open할때 받았던 자료구조에 값 저장, read buffer, 받을 size, msg priority 

• tlpi-dist/pmsg/pmsg_send.c

Message 수신

#include <mqueue.h>

ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);  // mq descripto, 받을 buffer, 받을 size, priority

• Message가 올때까지 block이 걸림
• 다른 쪽에서 send하는 순간 wakeup, mq_receive 이후 코드 수행
• Message queue로 짜면 Event driven으로 코드를 짤 수 있다 (받는쪽은 Message만 기다리고 있으면 됨)
• tlpi-dist/pmsg/pmsg_receive.c
• mq_msgsize system default : 8K
• numRead = mq_receive(mqd, buffer, attr.mq_mqsize, &prio);
if (numRead == -1) errExit("mq_receive");
  Message가 올때까지 기다림
  

Message 송수신

$ ./pmsg_create -cx /mq
$ ./pmsg_send /mq msg-a 5
$ ./pmsg_send /mq msg-b 0
$ ./pmsg_send /mq msg-c 10

# 높은 priority부터 
$ ./pmsg_receive /mq
Read 5 bytes; priority = 10
msg-c
$ ./pmsg_receive /mq
Read 5 bytes; priority = 5
msg-a
$ ./pmsg_recieve /mq
Read 5 bytes; priority = 0
msg-b

Command Line으로 Message queue 출력

• $ sudo mkdir /dev/mqueue
$ sudo mount -t mqueue none /dev/mqueue (source, target)
  
• $ cd /dev/mqueue
$ cat mq → Message queue 정보 확인 (queue byte size 등)
  $ ./pmsg_send로 보낸 byte만큼 queue에 write 되어 쌓여있음
  
• $ ./pmst_receive /mq 각각 하고나서 $ cat mq 다시 확인해보면 각각 받는 byte만큼 남은 queue size 줄어듦