[Lecture 10] 멀티쓰레딩 IV - Condition Variables & Monitors

#Condition Variables & Monitors / 조건 변수 & 모니터

조건 변수(mutex와 함께)는 모니터라고 하는 더 큰 추상화의 구성 요소이다.

 

동시성 프로그래밍에서 모니터(Monitor)는 스레드가 상호 배제와 특정 조건이 거짓이 될 때까지 기다리는(차단) 기능을 모두 가질 수 있도록 하는 동기화 구성체이다. 모니터는 또한 다른 스레드의 조건이 충족되었음을 알리는 메커니즘을 가지고 있다. 모니터는 뮤텍스(lock) 개체와 조건 변수로 구성된다.

 

조건 변수는 본질적으로 특정 조건을 기다리는 스레드의 컨테이너이다. 모니터는 스레드가 배타적 액세스 권한을 다시 얻고 작업을 재개하기 전에 특정 조건이 충족될 때까지 대기하기 위해 일시적으로 배타적 액세스를 포기하는 메커니즘을 제공한다. 쉽게 말해 조건 변수는 하나의 스레드가 하나 이상의 다른 스레드에 신호를 보낼 수 있도록 하는 목적으로 스레드 간에 다른 시그널링 방법을 제공한다. 

#Monitors / 모니터

스레드가 상호 작용할 때 일반적으로 상태는 공유되고 이 상태에 대한 업데이트에 대해 서로 신호를 보낼 필요가 있다. 따라서 업데이트와 시그널링은 상호 연결된다

• 모니터는 공유 변수 집합과 해당 변수에 대한 작업을 결합한다
• 캡슐화된 Java 클래스를 생각해 보자. 이런 클래스의 모든 필드는 prive이며 public method을 통해서만 액세스된다
• 모니터는 entry & exit 시 획득 및 해제되는 단일 뮤텍스와 연결되어 있으므로 메소드는 critical section을 형성한다
• 모니터는 하나 이상의 신호 대기열을 사용할 수 있다

#모니터에서 무한 버퍼 구현

monitor buffer {
	// implied: struct lock mlock;
private:
	char buffer[];
	int head, tail;
public:
	produce(item);
	item consume();
}

 

 

buffer::produce(item i)
{ 	// try { mlock.acquire()
	buffer[head++] = i;
	// } finally { mlock.release() }
}
buffer::consume()
{ 	// try { mlock.acquire()
	return buffer[tail++];
	// } finally { mlock.release() }
}

위의 예제는 모니터에 의해 추가될 상호 배제 기능을 보여준다. 각 인스턴스에는 메서드를 입력할 때 획득되고 메서드를 떠나는 모든 경로에 해제되는 암묵적(숨겨진) 잠금이 지정된다.

#모니터에서 유한 버퍼 구현

경계 버퍼가 비어 있거나 가득 차있을 수 있다. 따라서 producer와 consumer는 consumer가 버퍼가 비어 있는 상태에서 비어 있지 않은 상태로 전환되는 시점을 학습하고 producer가 버퍼가 가득 찬 상태에서 채워지지 않은 상태로 전환되는 시점을 학습하도록 조정하고 확인해야 한다.

 

언제나 처럼, 우리는 정확하고 효율적이며 불필요한 지연을 초래하지 않는 방법을 원한다. 이 방법은 condition variable, 즉 조건 변수에 의해 해결될 수 있다.

Producer:

int coin_flip;
sem_t coin_flip_done; // semaphore for thread 1 to signal coin flip
// requires sem_init(&coin_flip_done, 0, 0) to give initial value 0
static void * thread1(void *_)
{
	coin_flip = rand() % 2;
	sem_post(&coin_flip_done); // raise semaphore, increment, 'up'
    
	printf("Thread 1: flipped coin %d\n", coin_flip);
}

Consumer:

static void * thread2(void *_)
{
	// wait until semaphore is raised, then decrement, 'down'
	sem_wait(&coin_flip_done);
	printf("Thread 2: flipped coin %d\n", coin_flip);
}

#Condition Variable / 조건변수

• 위에도 언급한 듯이 조건변수는 3가지 작업을 지원하는 신호 대기열이다. 
  • wait(): 현재 스레드를 대기열에 추가하고 BLOCKED 상태로 들어간다
  • signal(): 대기열에 있는 스레드가 있으면 첫 번째 스레드를 제거하고 READY상태로 만듬
  • broadcast(): 대기열에서 모든 스레드를 제거하고 READY상태로 만듬
  • 대기열에 스레드가 없을 경우 signal()와 broadcast()는 아무런 영향을 미치지 않는다

#Condition Variable – Wait Operation

• wait() 작업에는 세 단계가 포함된다
  • 모니터 lock release
  • 대기열에 현재 스레드를 추가하고 BLOCKED 상태로 들어감
  • (신호 또는 브로드캐스트 작동의 결과로 차단이 해제된 경우), 모니터 잠금을 다시 획득한다
• release+ block 단계는 atomic 단계이다. 즉, 스레드가 대기열에 추가되기 전에는 다른 스레드가 잠금을 획득할 수 없다. 이렇게 하면 wake-up 기능이 손실되지 않는다.
• 그러나 unblock 및 reacquisition 단계는 그렇지 않다. 모니터에 다시 들어가는 wake-up 스레드는 다른 곳에 들어가는 스레드에 비해 모니터에 다시 들어갈 때 우선 순위가 부여되지 않는다

#Condition Variable – Rechecking the condition after Wait()

• 스레드가 대기하게 된 조건은 Wait()에서 돌아온 후 다시 점검해야 한다. 이는 다음과 같은 이유로 잠시 동안 반복됨을 의미한다:
  • 다른 스레드가 Wait()에서 스레드가 반환되기 전에 lock을 획득하고 있을 수 있다. 예시:
    • 스레드 A(consumer) 호출 consume(), 큐가 비어 있음, wait() 호출 및 block
    • 스레드 B(producer) 호출 produce()는 항목을 추가하고 신호를 추가한다. 스레드 A가 웨이크업(READY) 상태이지만 잠금을 다시 획득하지 않았다.
    • 스레드 C (consumer) 가 consume()호출
    • 스레드 C는 먼저 모니터 lock을 획득하고, B에서 생산된 아이템을 제거한다.
    • 스레드 A는 모니터 lock을 가져오고 Wait()에서 반환하지만 대기열에 항목이 없다. 지금 올바른 조치는 다시 확인하고 Wait()를 다시 호출하는 것이다

IEEE Std. 1003.1

조건 변수를 사용할 때 각 조건과 연관된 공유 변수를 포함하는 부울 술어가 항상 존재하며, 스레드가 계속 진행되어야 할 경우 참이다. pthread cond timed wait() 또는 thread cond wait() 함수에서 잘못된 웨이크업 현상이 발생할 수 있다. pthread cond timed wait() 또는 pthread cond wait()로부터의 반환은 이 술어의 값에 대한 어떠한 의미도 없기 때문에, 이러한 반환 시 술어를 다시 평가해야 한다.

 

POSIX.1 2008

가짜 웨이크업 허용의 추가적인 이점은 응용 프로그램이 조건 대기 주위에 술어 테스트 루프를 코딩하도록 강제한다는 것이다. 이것은 또한 애플리케이션이 불필요한 조건 브로드캐스트 또는 애플리케이션의 다른 부분에서 코딩될 수 있는 동일한 조건 변수의 신호를 허용하도록 만든다. 따라서 결과 애플리케이션은 더욱 강력하다. 따라서 POSIX.1-2008은 스플리어스 웨이크업 발생 가능성을 명시적으로 문서화한다.

 

#유한 버퍼(C 구현)

#define CAPACITY 10
// internal bounded buffer state
static int buffer[CAPACITY];
static int tail, head;

// protects the monitor: buffer, head, tail
static pthread_mutex_t buffer_lock = PTHREAD_MUTEX_INITIALIZER;

// items available for consumption
static pthread_cond_t items_avail = PTHREAD_COND_INITIALIZER;

// buffer slots available for production
static pthread_cond_t slots_avail = PTHREAD_COND_INITIALIZER;

void produce(int item);

void consume();

 

 

void produce(int item)
{
		pthread_mutex_lock(&buffer_lock); // Enter monitor
		while ((tail + 1 - head) % CAPACITY == 0)
		// Add calling thread to 'slots_avail' queue
		// Leave monitor (release buffer_lock)
		// Block
		// Enter monitor (acquire buffer_lock)
		pthread_cond_wait(&slots_avail, &buffer_lock);
	// invariant: (a) buffer_lock is held
	// (b) slot is available
	printf("thread %p: produces item %d\n", (void *)pthread_self(), item);
	buffer[head] = item; // update buffer state
	head = (head + 1) % CAPACITY;
	pthread_cond_signal(&items_avail); // wake up consumer (if any)
	pthread_mutex_unlock(&buffer_lock); // Leave monitor
}

 

 

void consume()
{
        pthread_mutex_lock(&buffer_lock); // Enter monitor
        while (head == tail)
        // Add thread to `items_avail` queue
        // Leave monitor (release buffer_lock)
        // Block
        // Enter monitor (acquire buffer_lock)
        pthread_cond_wait(&items_avail, &buffer_lock);
    // invariant: (a) buffer_lock is held
    // (b) item is available
    int item = buffer[tail];
    tail = (tail + 1) % CAPACITY;
    printf("thread %p: consumes item %d\n", (void *)pthread_self(), item);
    pthread_cond_signal(&slots_avail); // wake up producer (if any)
    pthread_mutex_unlock(&buffer_lock); // Leave monitor
}

#조건 변수가 있는 유한 버퍼

monitor buffer {
	condition items_avail;
	condition slots_avail;
private:
	char buffer[];
	int head, tail;
public:
	produce(item);
	item consume();
}

 

 

buffer::produce(item i)
{ /* wait while buffer is full */
	while ((tail+1-head)%CAPACITY == 0)
		slots_avail.wait();
	buffer[head++] = i;
	items_avail.signal();
}
buffer::consume()
{ /* wait while buffer is empty */
	while (head == tail)
		items_avail.wait();
	item i = buffer[tail++];
	slots_avail.signal();
	return i;
}

위 예제에서 작은 요구 사항: 교착 상태를 피하려면 조건 변수에서 대기하는 스레드가 모니터에서 벗어나야 한다(예: lock release)

#Comparison to Semaphores

Semaphores	Condition Variables
Signal()/Post()는 현재 Wait()에서 차단된 스레드가 없는 경우에도 기억된다 wait()는 호출 스레드를 차단하거나 차단하지 않을 수 있다 wait()/post() 수가 일치하는 경우 사용 공유 상태( shared state)에 안전하게 액세스하기 위한 별도의 조치 필요	wait()에서 스레드가 현재 차단되지 않은 경우 signal()이 영향을 미치지 않는다 wait()는 항상 호출 스레드를 차단한다. 임의/복잡한 상태 변경 조정 시 사용 변경될 수 있는 공유 상태를 보호하는 lock과 함께 사용해야 한다

#자바에서의 모니터 / 조건변수

모든 Java 객체는 정확히 하나의 조건 변수를 가진 모니터로 사용될 수 있다 (java.lang.Object.*의 메소드들인 wait(), notify(), notifyAll()을 통해서).

 

모니터에 들어가는 메소드들은 "synchronized"로 표시.

 

예제:

class buffer {
	private char buffer[];
	private int head, tail;
	public synchronized produce(item i) {
		while (buffer_full())
			this.wait();
		buffer[head++] = i;
		this.notifyAll();
}

public synchronized item consume() {
		while (buffer_empty())
			this.wait();
		i = buffer[tail++];
		this.notifyAll();
		return ;
	}
}

하지만 이러한 방식은 좋은 설계 선택은 아니다. 유연성이 떨어지고 JVM에 막대한 구현 비용이 부과된다.

#java.util.concurrent의 조건 변수

java.util.concurrent 에서는 모니터 패턴을 보다 유연하게 사용할 수 있다. java.util.concurrent에서는 여러 조건 변수가 lockl과 연관될 수 있다. 단 한가지 주의할 점은 모니터 블록에 대한 구문 지원되지 않아 lock()/unlock() 호출시 try/finally 블록에 넣어야 한다.

import java.util.concurrent.locks.*;

class buffer {
	private ReentrantLock monitorlock
		= new ReentrantLock();
	private Condition items_available
		= monitorlock.newCondition();
	private Condition slots_available
		= monitorlock.newCondition();
        
	public /* NO SYNCHRONIZED here */void produce(item i) {
		monitorlock.lock();
		try {
			while (buffer_full())
			slots_available.await();
			buffer[head++] = i;
			items_available.signal();
		} finally {
		  monitorlock.unlock();
		}
	} /* consume analogous */
}

#Thread-Local Storage

per-thread 변수를 제공한고 컴파일러 + 링커 + 런타임의 조합으로 제공되는 효율적인 지원함.

//source code
_Thread_local int x; // C11
__thread int x; // legacy gcc version
x++;

//assembly code
movl %fs:x@tpoff, %eax
addl $1, %eax
movl %eax, %fs:x@tpoff

Thread-Local Storage