동기 처리+동시성에서의 문제점
여기서는 배리어 동기, readers-writers lock, 동시성 특유의 문제점과 버그 일부분에 대해 다룬다.
배리어 동기
배리어 동기란 공유 변수를 증가시키다 공유 변수가 어떤 일정한 수에 도달하면 배리어를 벗어나 처리를 수행하는 방식이다.
이러한 배리어 동기를 사용하면 진행 순서를 보장할 수 있는데 예컨대 스레드 A가 스레드 B보다 먼저 넘어가면 안 된다고 가정하자. 이러한 경우에 A가 포함된 모든 1단계 스레드가 배리어에 도달해야만 다음 단계(B 포함)를 실행할 수 있게 되므로 진행 순서를 보장할 수 있다.
스핀락 기반 배리어 동기
다음 코드는 스핀락 기반의 배리어 동기를 보여주는데,
// 공유 변수에 대한 포인터 cnt와 최댓값 max
void barrier(volatile int *cnt, int max) {
__sync_fetch_and_add(cnt, 1); // 공유 변수를 아토믹하게 증가
while (*cnt < max); // cnt가 가리키는 값이 max가 될때까지 대기
}
배리어 동기를 이용하는 코드의 예시는 다음과 같다.
volatile int num = 0; //공유변수
void *worker(void *arg) { // 스레드용 함수
barrier(&num, 10); // 배리어 동기 실행. 모든 스레드가 배리어에 도달하기 전까지 5행 처리 X
// 무언가 처?리
return NULL;
}
int main(int argc, char *argv[]) {
// 스레드 생성
pthread_t th[10];
for (int i = 0; i < 10; i++) {
if (pthread_create(&th[i], NULL, worker, NULL) != 0) {
perror("pthread_create"); return -1;
}
}
// join
return 0;
}
Pthreads를 이용한 배리어 동기
스핀락을 이용한 배리어 동기에서는 대기 중에 루프 처리를 수행해야 하므로 CPU 리소스를 낭비할 여지가 있다. Pthreads의 조건 변수를 이용해서 이를 완화하는 방법에 대해 알아보자.
이것은 다음 코드로 구현될 수 있는데
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
pthread_mutex_t barrier_mut = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t barrier_cond = PTHREAD_COND_INITIALIZER;
void barrier(volatile int *cnt, int max) {
if (pthread_mutex_lock(&barrier_mut) != 0) {
perror("pthread_mutex_lock"); exit(-1);
}
(*cnt)++; // 락 획득 후 공유변수 증가
if (*cnt == max) {
// *cnt와 max가 같으면 스레드 모두 실행
if (pthread_cond_broadcast(&barrier_cond) != 0) {
perror("pthread_cond_broadcast"); exit(-1);
}
} else {
do { // 값이 같지 않으면 대기
if (pthread_cond_wait(&barrier_cond, &barrier_mut) != 0) {
perror("pthread_cond_wait"); exit(-1);
}
} while (*cnt < max);
}
if (pthread_mutex_unlock(&barrier_mut) != 0) {
perror("pthread_mutex_unlock"); exit(-1);
}
}
러스트 배리어 동기
use std::sync::{Arc, Barrier}; // 배리어 동기
use std::thread;
fn main() {
// 스레드 핸들러 저장
let mut v = Vec::new();
// 10 스레드만큼의 배리어 동기를 Arc로 감싸기
let barrier = Arc::new(Barrier::new(10));
// 스레드 실행
for _ in 0..10 {
let b = barrier.clone();
let th = thread::spawn(move || {
b.wait(); // 배리어 동기 대기
println!("finished barrier");
});
v.push(th);
}
for th in v {
th.join().unwrap();
}
}
실행 결과는 다음과 같다.
Compiling barrier v0.1.0 (/home/max/coding/concurrent/barrier)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.17s
Running `target/debug/barrier`
finished barrier
finished barrier
finished barrier
finished barrier
finished barrier
finished barrier
finished barrier
finished barrier
finished barrier
Readers-Writers 락
레이스 컨디션이 발생하는 원인은 쓰기 처리 때문이므로 쓰기만 배타적으로 수행하면 문제가 발생하지 않는다.
뮤텍스와 세마포어에서는 프로세스에 특별한 역할을 설정하지 않았다. 사실 쓰기만 문제가 되는데 읽기까지 제한을 하는 것은 비효율적으로 보일 수 있다. 따라서 쓰기와 읽기를 구분하고, 읽기끼리는 병렬성을 허용하되 쓰기 때는 제한하는 방식을 제안할 수 있는데 그것이 Readers-Writers락이다.
Readers-Writer락(RW 락) 에서는 읽기만 수행하는 프로세스(reader)와 쓰기만 수행하는 프로세스(wrtier)로 분류하고 다음 제약을 만족하도록 한다.
락을 획득 중인 reader는 같은 시각에 다수 존재할 수 있다.
락을 획득 중인 writer는 같은 시각에 1개만 존재할 수 있다.
reader와 writer는 같은 시각에 락 획득 상태가 될 수 없다.
스핀락 기반 RW락
스핀락 기반의 RW락 알고리즘은 다음과 같이 구현될 수 있다. Reader용 락 획득과 반환 함수, Writer용 락 획득과 반환 함수는 별도의 인터페이스로 구현되어 있으니 실제 이용할 때는 공유 리소스의 읽기만 수행할지 쓰기만 수행할지 적절하게 판단해서 이용해야 한다.
// reader용 락 획득 함수
void rwlock_read_acquire(int *rcnt, volatile int *wcnt) {
for (;;) {
while (*wcnt); //writer가 있으면 대기
__sync_fetch_and_add(rcnt, 1);
if (*wcnt == 0) // writer가 없으면 락 획득
break;
__sync_fetch_and_sub(rcnt, 1);
}
}
// reader용 락 반환 함수
void rwlock_read_release(int *rcnt) {
__sync_fetch_and_sub(rcnt, 1);
}
// writer용 락 획득 함수
void rwlock_write_acquire(bool *lock, volatile int *rcnt, int *wcnt) {
__sync_fetch_and_add(wcnt, 1);
while (*rcnt); // reader가 있으면 대기
spinlock_acquire(lock);
}
// writer용 락 반환 함수
void rwlock_write_release(bbol *lock, int *wcnt) {
spinlock_release(lock);
__sync_fetch_and_sub(wcnt, 1);
}
활용은 C에서는
int rcnt = 0;
int wcnt = 0;
bool lock = false;
void reader() {
for (;;) {
rwlock_read_acquire(&rcnt, &wcnt);
//크리티컬 섹션
rwlock_read_release(&rcnt);
}
}
void writer () {
for (;;) {
rwlock_write_acquire(&lock, &rcnt, &wcnt);
// 크리티컬 섹션
rwlock_write_release(&lock, &wcnt);
}
}
러스트에서는
use std::sync::RwLock;
fn main() {
let lock = RwLock::new(10);
{
let v1 = lock.read().unwrap();
let v2 = lock.read().unwrap();
println!("v1: {}, v2: {}", *v1, *v2);
}
{
let mut v = lock.write().unwrap();
*v = 7;
println!("v: {}", *v);
}
}
실행 속도
Read가 대부분인 경우 뮤텍스보다 RW락을 사용하는 것이 실행 속도를 향상시키는 데 도움이 될 것이다. Read에 대해 병렬성을 허용하기 위한 것이 RW락이므로, 당연한 결과이다.
동시성 프로그래밍 특유의 버그와 문제점
동시성 프로그래밍도 특유의 버그와 문제점이 있다. 대표적으로 데드락, 라이브락, 기아(starvation) 등 문제가 발생한다.
데드락
동시성 프로그래밍 특유의 비유에 대해 식사하는 철학자 문제라는 비유가 있다. 원본은 포크지만 젓가락이 더 올바른 비유에 가까울 듯하니 젓가락으로 비유하겠다. 철학자가 원탁에 둘러앉아 있는데 철학자 사이에 젓가락이 하나씩 놓여있고, 젓가락 두 개가 있어야 식사를 할 수 있다.
위의 그림과 같은 상황에서, 철학자들은 다음 알고리즘을 따르는데,
왼쪽 젓가락이 사용 가능할 때까지 기다렸다가 사용 가능해지면 젓가락을 든다.
오른쪽에 대해서도 마찬가지로 사용 가능해지면 젓가락을 든다.
식사를 한다.
젓가락을 다시 내려놓는다.
1로 돌아간다.
만약에 모든 철학자가 동시에 자신의 왼편에 놓인 젓가락을 집어든다고 치자. 모든 철학자가 젓가락 하나만을 가지고 있기 때문에 모두 식사할 수 없는데, 위 알고리즘에 따르면 모든 철학자들이 오른쪽이 사용 가능해질 때까지 대기하고 있지만, 모두가 식사를 하고 젓가락을 내려놓을 수 없으니 모두가 오른쪽을 대기하는 상태로 멈춰버리고, 모두가 식사를 하지 못한다.
이처럼 자원이 비는 것을 기다리며 더 이상 처리가 진행되지 않는 상태를 데드락이라고 한다.
위의 철학자 문제를 코드로 구현해보자면 아래와 같다.
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let c0 = Arc::new(Mutex::new(()));
let c1 = Arc::new(Mutex::new(()));
let c0_p0 = c0.clone();
let c1_p0 = c1.clone();
//철학자0
let p0 = thread::spawn(move || {
for _ in 0..1000 {
let _lock0 = c0_p0.lock().unwrap();
let _lock1 = c1_p0.lock().unwrap();
// 철학자가 밥을 먹는 코드
println!("Philosopher 0 is eating");
}
});
let p1 = thread::spawn(move || {
for _ in 0..1000 {
let _lock1 = c1.lock().unwrap();
let _lock0 = c0.lock().unwrap();
// 철학자가 밥을 먹는 코드
println!("Philosopher 1 is eating");
}
});
p0.join().unwrap();
p1.join().unwrap();
}
이 코드는 마지막까지 실행될 수도 있지만 데드락이 발생해 더 이상 실행되지 않기도 한다.
특히 RW락은 데드락을 주의해야 하는데, 아래 코드처럼 데드락을 발생시킬 수 있다.
use std::sync::{Arc, RwLock};
use std::thread;
fn main() {
let val = Arc::new(RwLock::new(true));
let t = thread::spawn(move || {
let flag = val.read().unwrap(); //read락 획득
if *flag {
*val.write().unwrap() = false; // write락 획득
println!("flag is true");
}
});
t.join().unwrap();
}
read락을 획득한 상태로 write락을 획득하게 되기 때문에 데드락 상태가 된다.
이를 해결하기 위해 다음과 같이 제시될 수 있다.
use std::sync::{Arc, RwLock};
use std::thread;
fn main() {
let val = Arc::new(RwLock::new(true));
let t = thread::spawn(move || {
let flag = *val.read().unwrap(); // 읽기 락 drop
if flag {
*val.write().unwrap() = false; // 쓰기 락 가능
println!("flag is true");
}
});
t.join().unwrap();
}
이 코드는 데드락이 발생하지 않는다.
첫 번째 코드의 작동 원리를 알아보자면 이렇다. val.read()는 RwLockReadGuard를 생성하고 flag가 그것을 소유하게 된다. if *flag는 단순히 참조할 뿐, flag의 수명은 줄이지 않는다. 따라서 읽기 락이 유지된 상태로 쓰기 락을 시도하게 된다.
두 번째 코드에서 val.read()는 RwLockReadGuard를 반환하지만 이 값을 *로 역참조해 복사한다. 이 복사는 소유권의 이동이 아니라 스칼라 타입의 복사(Copy)라서 flag는 read_guard없이도 존재할 수 있다. 그리고 읽기 락은 곧바로 drop되기 때문에 쓰기 락을 안전하게 획득할 수 있다.
라이브락과 starvation
식사하는 철학자 문제를 수정해서 다음과 같이 알고리즘을 변경한다고 치자.
왼쪽 젓가락이 사용 가능할 때까지 기다렸다가 사용 가능해지면 젓가락을 든다.
오른쪽에 대해서도 마찬가지로 사용 가능해지면 젓가락을 들지만, 어느정도 기다려도 들 수 있는 상태가 되지 않으면 왼쪽 젓가락을 내려놓고 단계1로 되돌아간다.
식사를 한다.
젓가락을 다시 내려놓는다.
1로 돌아간다.
이 고쳐진 알고리즘은 잘 작동할 것 같지만 마찬가지로 두 명의 철학자가 왼쪽 젓가락을 동시에 들고 있다가 내려놓고 동시에 드는 상황이 일어나면 이 상황이 반복되어 처리가 진행되지 않는다.
이렇게 리소스를 획득하는 처리는 수행하지만 다음 리소스를 획득하지 못해 이후의 처리를 하지 못하게 되는 상태를 라이브락이라고 한다.
라이브락이 되는 스테이트 머신은 다음과 같이 정의할 수 있다.
스테이트 머신에서 라이브락이 발생할 가능성이 있다 <=> 특정한 리소스를 획득하는 상태에는 도달하지만 그 외의 상태에는 절대 도달하지 못하는 무한 전이 사례가 존재한다.
그리고 굶주림(기아, starvation) 이란 특정 프로세스만 리소스 획득 상태로 전이하지 못하는 상태에 있는 것을 말한다. 이 또한 스테이트 머신을 정의하면
스테이트 머신에서 굶주림이 발생할 가능성이 있다 <=> 어떤 프로세스가 존재하고, 항상 리소스를 요청 가능한 상태에 도달하지만 실제 리소스를 획득하고 진행하는 상태로는 결코 도달하지 못하는 무한 루프를 가진다..
은행원 알고리즘
데드락을 회피하기 위한 알고리즘으로 다익스트라가 고안한 은행원 알고리즘이 있다. 예를 들자면 다음과 같다.
은행원은 2000만원의 자본을 가지고 있고 기업 A와 기업 B는 각각 1500만원, 2000만원의 자금이 필요하다. 은행원은 먼저 기업 A에 1500만원을 대출해주고, A는 이 돈으로 사업을 하고 다시 1500만원을 상환한다. 그 후 은행원은 B에 2000만원을 대출해주고 B는 사업 후 2000만원을 상환한다.
단 여기에 다음과 같은 제약이 존재하는데
기업은 자금을 대출하는 즉시 사용
기업은 필요한 금액을 대출받으면 반드시 전액 상환
기업은 전액 대출받기 전까지 사업할 수 없다.
이자는 존재하지 않는다.
은행은 보유 자금 이상 대출할 수 없다.
이 상황에서 아래 그림과 같은 상황을 가정하자.
이 경우 데드락이 발생한다. 둘 모두 사업할 수 없고 상환할 수도 없다.
위 두 사례에서 알 수 있는 것은 은행원의 자본과 각 기업이 필요로 하는 자본의 금액을 미리 알고 있다면 시뮬레이션을 통해 어떻게 대출을 해주면 데드락이 되는지 예측 가능하다. 따라서 은행원 알고리즘에서는 데드락이 발생하는 상태로 전이하는지 시뮬레이션을 통해 판정함으로써 데드락을 회피한다.
조금 더 부연설명하자면 은행은 한 기업이 요구하는 자금의 전액 정도는 가지고 있어야 한다. 즉 위 그림에서 은행원이 전체 1000만원을 가지고 있다면 성공적으로 대출해줄 수 없을 것이다. 그러나 위에서는 2000만원을 보유하고 있어 순서를 적절히 규정해주면 모두에게 돈을 빌려줄 수 있다.
이를 정리하면 은행원 알고리즘에서 상태는 두 가지가 있다.
안전상태: 시스템 교착을 일으키지 않고 각 프로세스가 요구한 양만큼 자원을 할당해줄 수 있는 안전 순서열이 존재하는 상태 불안전상태: 프로세스가 요구한 양만큼 자원 할당이 불가해 안전순서열이 존재하지 않는 상태
은행원 알고리즘은 결과적으로 자원의 할당 허용 여부를 결정하기 전에 미리 결정된 모든 자원의 최대 가능한 할당량을 시뮬레이션하여 안전 여부를 검사하고, 대기중인 모든 활동의 교착 상태 가능성을 조사하여 안전 상태 여부를 검사한다.
코드로는 다음과 같다.
use std::sync::{Arc, Mutex};
struct Resource<const NRES: usize, const NTH: usize> {
available: [usize; NRES], // 이용 가능한 리소스
allocation: [[usize; NRES]; NTH], // 스레드 i가 확보 중인 리소스
max: [[usize; NRES]; NTH], // 스레드 i가 필요로 하는 리소스의 최댓값
}
impl<const NRES: usize, const NTH: usize> Resource<NRES, NTH> {
fn new(available: [usize; NRES], max: [[usize; NRES]; NTH]) -> Self {
Resource {
available,
allocation: [[0; NRES]; NTH],
max,
}
}
// 현재 상태가 데드록을 발생시키지 않는가 확인
fn is_safe(&self) -> bool {
let mut finish = [false; NTH]; // 스레드 i는 리소스 획득과 반환에 성공했는가?
let mut work = self.available.clone(); // 이용 가능한 리소스의 시뮬레이션값
loop {
// 모든 스레드 i와 리소스 j에 대해,
// finish[i] == false && work[j] >= (self.max[i][j] - self.allocation[i][j])
// 을 만족하는 스레드를 찾는다.
let mut found = false;
let mut num_true = 0;
for (i, alc) in self.allocation.iter().enumerate() {
if finish[i] {
num_true += 1;
continue;
}
// need[j] = self.max[i][j] - self.allocation[i][j] 를 계산하고,
// 모든 리소스 j에 대해, work[j] >= need[j] 인가를 판정한다.
let need = self.max[i].iter().zip(alc).map(|(m, a)| m - a);
let is_avail = work.iter().zip(need).all(|(w, n)| *w >= n);
if is_avail {
// 스레드 i가 리소스 확보 가능
found = true;
finish[i] = true;
for (w, a) in work.iter_mut().zip(alc) {
*w += *a // 스레드 i가 현재 확보하고 있는 리소스를 반환
}
break;
}
}
if num_true == NTH {
// 모든 스레드가 리소스 확보 가능하면 안전함
return true;
}
if !found {
// 스레드가 리소스를 확보할 수 없음
break;
}
}
false
}
// id번 째의 스레드가 resource를 하나 얻음
fn take(&mut self, id: usize, resource: usize) -> bool {
// 스레드 번호, 리소스 번호 검사
if id >= NTH || resource >= NRES || self.available[resource] == 0 {
return false;
}
// 리소스 확보를 시험해 본다
self.allocation[id][resource] += 1;
self.available[resource] -= 1;
if self.is_safe() {
true // 리소스 확보 성공
} else {
// 리소스 확보에 실패했으므로 상태 원복
self.allocation[id][resource] -= 1;
self.available[resource] += 1;
false
}
}
// id번 째의 스레드가 resource를 하나 반환
fn release(&mut self, id: usize, resource: usize) {
// 스레드 번호, 리소스 번호를 검사
if id >= NTH || resource >= NRES || self.allocation[id][resource] == 0 {
return;
}
self.allocation[id][resource] -= 1;
self.available[resource] += 1;
}
}
#[derive(Clone)]
pub struct Banker<const NRES: usize, const NTH: usize> {
resource: Arc<Mutex<Resource<NRES, NTH>>>,
}
impl<const NRES: usize, const NTH: usize> Banker<NRES, NTH> {
pub fn new(available: [usize; NRES], max: [[usize; NRES]; NTH]) -> Self {
Banker {
resource: Arc::new(Mutex::new(Resource::new(available, max))),
}
}
pub fn take(&self, id: usize, resource: usize) -> bool {
let mut r = self.resource.lock().unwrap();
r.take(id, resource)
}
pub fn release(&self, id: usize, resource: usize) {
let mut r = self.resource.lock().unwrap();
r.release(id, resource)
}
}
mod banker;
use banker::Banker;
use std::thread;
const NUM_LOOP: usize = 100000;
fn main() {
// 이용 가능한 젓가락의 수, 철학자가 이용하는 포크 최대 수 설정
let banker = Banker::<2, 2>::new([1, 1], [[1, 1], [1, 1]]);
let banker0 = banker.clone();
let philosopher0 = thread::spawn(move || {
for _ in 0..NUM_LOOP {
// 젓가락 0과 1을 확보
while !banker0.take(0, 0) {}
while !banker0.take(0, 1) {}
println!("0: eating");
// 젓가락 0과 1을 반환
banker0.release(0, 0);
banker0.release(0, 1);
}
});
let philosopher1 = thread::spawn(move || {
for _ in 0..NUM_LOOP {
// 젓가락 1과 0을 확보
while !banker.take(1, 1) {}
while !banker.take(1, 0) {}
println!("1: eating");
// 젓가락 1과 0을 반환
banker.release(1, 1);
banker.release(1, 0);
}
});
philosopher0.join().unwrap();
philosopher1.join().unwrap();
}
재귀락
재귀락이란 락을 획득한 상태에서 프로세스가 그 락을 해제하기 전에 다시 그 락을 획득하는 것을 말한다.
재귀락이 발생하면 어떻게 되는가? 는 알고리즘의 구현에 따라 다르다. 단순한 뮤텍스 구현에 대해 재귀락을 수행하면 데드락 상태가 되겠지만 재귀락을 수행해도 처리를 계속할 수 있는 락이 존재하고 이를 재진입 가능한(reentrant) 락이라고 한다. 다시 정리하자만 재귀락을 수행해도 데드락 상태에 빠지지 않고 처리를 계속할 수 있는 락이다.
C언어에서 재진입 가능한 뮤텍스를 구현하면 다음과 같다.
#include <assert.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
// 재진입 가능한 뮤텍스용 타입 ❶
struct reent_lock {
bool lock; // 록용 공유 변수
int id; // 현재 록을 획득 중인 스레드 ID, 0이 아니면 록 획득중임
int cnt; // 재귀 록 카운트
};
// 재귀 록 획득 함수
void reentlock_acquire(struct reent_lock *lock, int id) {
// 록 획득 중이고 자신이 획득 중인지 판정 ❷
if (lock->lock && lock->id == id) {
// 자신이 획등 중이면 카운트를 증가
lock->cnt++;
} else {
// 어떤 스레드도 혹을 획득하지 않았거나,
// 다른 스레드가 록 획득 중이면 록 획득
spinlock_acquire(&lock->lock);
// 록윽 획득하면 자신의 스레드 ID를 설정하고
// 마운트를 인크리먼트
lock->id = id;
lock->cnt++;
}
}
// 재귀 록 해제 함수
void reentlock_release(struct reent_lock *lock) {
// 카운트를 디크리먼트하고,
// 해당 카운트가 0이 되면 록 해제 ❸
lock->cnt--;
if (lock->cnt == 0) {
lock->id = 0;
spinlock_release(&lock->lock);
}
}
// 활용 예시
struct reent_lock lock_var; // 록용 공유 변수
// n회 재귀적으로 호출해 록을 거는 테스트 함수
void reent_lock_test(int id, int n) {
if (n == 0)
return;
// 재귀 록
reentlock_acquire(&lock_var, id);
reent_lock_test(id, n - 1);
reentlock_release(&lock_var);
}
// 스레드용 함수
void *thread_func(void *arg) {
int id = (int)arg;
assert(id != 0);
for (int i = 0; i < 10000; i++) {
reent_lock_test(id, 10);
}
return NULL;
}
int main(int argc, char *argv[]) {
pthread_t v[NUM_THREADS];
for (int i = 0; i < NUM_THREADS; i++) {
pthread_create(&v[i], NULL, thread_func, (void *)(i + 1));
}
for (int i = 0; i < NUM_THREADS; i++) {
pthread_join(v[i], NULL);
}
return 0;
}
C와는 달리 러스트에서는 재귀락을 수행하는 코드는 의도적으로 작성하지 않으면 거의 일어나지 않는 것으로 보인다. 락용 변수와 리소스가 강하게 결합되어 있기 때문이다.
아래와 같이 재귀락을 수행하는 코드를 짤 수 있기는 하나, 일반적이지 않다.
use std::sync::{Arc, Mutex};
fn main() {
// 뮤텍스를 Arc로 작성하고 클론
let lock0 = Arc::new(Mutex::new(0)); // ❶
// Arc의 클론은 참조 카운터를 증가하기만 한다
let lock1 = lock0.clone(); // ❷
let a = lock0.lock().unwrap();
let b = lock1.lock().unwrap(); // 데드록 ❸
println!("{}", a);
println!("{}", b);
}