Thread Cooperation
앞서 우리는 multithreading과 synchronization에 대해 공부하였다.
critical region에 대한 race condition을 피하는 데에는 synchronization으로 충분하지만, 때로는 여러 개의 thread가 협력하는 동작이 필요할 때도 있다.
Conditions
Lock 객체에서 newCondition() 메소드를 호출하면 생성되는 객체다.
Condition 객체는 thread들이 의사소통하는 것을 용이하게 한다. 한 번 condition 객체는 await(), signal(), signalAll() 메소드를 사용할 수 있다.
아래 예제 코드와 함께 살펴보자.
Account Class
Account
private static class Account {
// Lock 객체 생성
private static Lock lock = new ReentrantLock();
// lock에서 Condition 객체 생성
private static Condition newDeposit = lock.newCondition();
private int balance = 0; // 잔고
public int getBalance() {
return balance;
}
// 출금
public void withdraw(int amount) {
lock.lock(); // 잠금
try {
// 잔액 부족이 발생한 경우
while (balance < amount) {
System.out.println("\t\t\tWait for a deposit");
// 재개 신호(입금)가 들어올 때 까지 기다림(await)
// *주의*: signal() 또는 signalAll()을 호출해주지 않으면
// 이 thread는 영원히 기다리게 된다.
newDeposit.await();
}
balance -= amount;
System.out.println("\t\t\tWithdraw " + amount + "\t\t" + getBalance());
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock(); // 잠금 해제
}
}
public void deposit(int amount) {
lock.lock();
try {
balance += amount;
System.out.println("Deposit " + amount + "\t\t\t\t\t" + getBalance());
// await 해제
newDeposit.signalAll();
} finally {
lock.unlock();
}
}
}새로운 Account 객체다. deposit(입금) 이외에 withdraw(출금) 메소드가 추가 되었다.
우리가 주목해서 볼 내용은 withdraw와 deposit의 상호작용인데, 출금 시에 잔액이 부족할 경우(while-loop 내부) Condition 객체가 await() 메소드를 호출하여 잔액이 충분해질 때 까지 출금을 막아버린다. 이렇게 await() 상태에 들어간 thread는 signal() 또는 signalAll() 메소드를 호출해주지 않으면 영원히 작업을 중지하고 기다린다.
deposit 메소드에 보면 입금 후 Condition 객체에서 signalAll() 메소드를 호출한다.
signal()은 대기 중인 thread 중 하나만(랜덤) 깨우기 때문에 효율적이지만 어떤 thread가 깨어날지 예상할 수 없기 때문에 약간의 위험성을 가진다.
반면 signalAll()은 대기 중인 모든 thread를 깨워서 반드시 깨워야할 thread가 대기상태일 가능성은 없지만 깨어난 thread끼리 다시 lock 경쟁을 벌이기 때문에 오버헤드가 존재한다.
주의할 점으로, Condition 객체는 Lock 객체로 부터 만들어지기 때문에 await(), signal(), signalAll() 등의 메소드를 사용하기 위해서는 반드시 lock을 가져야 한다. 만약 lock을 부여하지 않고 위 메소드들을 호출한다면 IllegalMonitorStateException이 던져질 것이다.
시험에 나오지는 않을 것 같지만, withdraw() 메소드에서 조건 검사를 위해 if문이 아닌 while문을 사용한 이유에 대해 알아보자. 이는 운영체제 레벨에서 발생할 수 있는 문제인데, signal이 없는 경우에도 스케쥴러 이슈로 갑자기 thread가 깨어나는 문제가 생기곤 한다(Spurious Wakeup). 이 때 if문을 사용했다면 곧바로 해당 블럭을 탈출해 출금을 시도해버릴 것이다. 하지만 while 블럭에 넣어둔다면 다시 잔액을 확인하고 대기시킬 것이다. 뿐만 아니라 다른 thread에게 새치기(Interception)를 당한 경우에 발생할 문제도 막을 수 있다. 출금을 하기 전 다른 thread가 먼저 돈을 빼가서 없는 잔고에 출금을 시도해버리는 것이다.
Task Class
DepositTask
public static class DepositTask implements Runnable {
@Override
public void run() {
try {
while (true) {
// 무작위 금액 입금, sleep 1000ms
account.deposit((int) (Math.random() * 10) + 1);
Thread.sleep(1000);
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}WithdrawTask
public static class WithdrawTask implements Runnable {
@Override
public void run() {
while (true) {
// 무작위 금액 출금
account.withdraw((int)(Math.random()*10) + 1);
}
}
}각 Task에서 account의 메소드를 호출한다. DepositTask의 코드에 sleep()이 존재하는 이유는 우리가 지금 잔액 부족 상황을 의도적으로 발생시킬 것이기 때문이다.
Main.java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static Account account = new Account();
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2);
executor.execute(new DepositTask());
executor.execute(new WithdrawTask());
executor.shutdown();
System.out.println("Thread 1\t\tThread 2\t\tBalance");
}
public static class DepositTask implements Runnable {
@Override
public void run() {
try {
while (true) {
account.deposit((int) (Math.random() * 10) + 1);
Thread.sleep(1000);
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
public static class WithdrawTask implements Runnable {
@Override
public void run() {
while (true) {
account.withdraw((int)(Math.random()*10) + 1);
}
}
}
private static class Account {
private static Lock lock = new ReentrantLock();
private static Condition newDeposit = lock.newCondition();
private int balance = 0;
public int getBalance() {
return balance;
}
public void withdraw(int amount) {
lock.lock();
try {
while (balance < amount) {
System.out.println("\t\t\tWait for a deposit");
newDeposit.await();
}
balance -= amount;
System.out.println("\t\t\tWithdraw " + amount + "\t\t\t\t" + getBalance());
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock();
}
}
public void deposit(int amount) {
lock.lock();
try {
balance += amount;
System.out.println("Deposit " + amount + "\t\t\t\t\t\t\t\t" + getBalance());
newDeposit.signalAll();
} finally {
lock.unlock();
}
}
}
}전체 코드 흐름을 살펴보자.
먼저 2개의 thread를 가진 thread pool을 생성하고 이 pool에 DepositTask와 WithdrawTask를 등록한다.
이 프로그램에서 DepositTask는 1000ms의 sleep에 의해 다소 느리게 작동하지만 WithdrawTask는 매우 빠르게 무한 반복한다. 따라서 반드시 잔액 부족의 상황이 발생한다.
Thread 1 Thread 2 Balance
Deposit 4 4
Wait for a deposit
Deposit 8 12
Withdraw 8 4
Wait for a deposit
Deposit 1 5
Wait for a deposit
Deposit 6 11
Withdraw 9 2
Wait for a deposit
Deposit 9 11
Withdraw 5 6
Wait for a deposit
Deposit 5 11
Withdraw 10 1
Wait for a deposit
.
.
.잔액 부족 상황에서는 DepositTask thread가 대기하는 것을 확인할 수 있다.
더 알아볼 점
사실 자바의 모든 객체는 그 자체가 Lock 기능을 가지고 있다.
// Task 1
synchronized (anObject) {
try {
while (!condition) anObject.wait();
} catch(InterruptedException ex) {
ex.printStackTrace();
}
}// Task 2
synchronized (anObject) {
anObject.notify(); // or anObject.notifyAll();
...
}wait(), notify(), notifyAll() 메소드들은 반드시 synchronized 메소드 또는 synchronized 블럭에서만 호출되어야 한다(즉, lock이 필요하다는 뜻). 그렇지 않으면 IllegalMonitorStateException이 던져질 것이다.
wait()이 호출되면 thread를 멈추고 현재 잡고 있는 lock을 반납한 후 대기상태에 들어간다. notify() 또는 notifyAll()이 호출되면 대기 중인 thread를 깨우고 해당 thread의 synchronized 블록의 작업을 끝낸 후 lock을 반납한다.
즉, wait(), notify(), notifyAll() 메소드는 위에서 배운 await(), signal(), signalAll() 메소드의 아날로그 버전인 것이다.
Producer-Consumer Pattern
앞에서 다룬 코드는 일종의 생산자-소비자 패턴으로 이루어져있다. 이제 해당 코드의 좀 더 제너럴한 버전을 함께 볼 것이다.
생산자-소비자 패턴은 다음 세가지 구성요소로 이루어진다.
- 생산자(Producer): 데이터를 생산해 공유자원(버퍼)에 넣는 thread
- 소비자(Consumer): 공유자원에서 데이터를 꺼내 사용하는 thread
- 버퍼(Buffer): 공유 자원
Buffer Class
Buffer
private static class Buffer {
private static final int CAPACITY = 1; // 버퍼 크기
// 버퍼는 queue로 구현됨
private java.util.LinkedList<Integer> queue = new java.util.LinkedList<>();
// lock 생성
private static Lock lock = new ReentrantLock();
// 두 가지 condition 생성
private static Condition notEmpty = lock.newCondition(); // 소비자를 깨움
private static Condition notFull = lock.newCondition(); // 생산자를 깨움
public void write(int value) {
lock.lock(); // 잠금
try {
while (queue.size() == CAPACITY) { // 조건 검사
// 현재 버퍼가 꽉 차있다면
System.out.println("Wait for notFull condition");
notFull.await(); // notFull 시그널 기다림
}
queue.offer(value); // write
notEmpty.signal(); // notEmpty 시그널 전송
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock(); // 잠금 해제
}
}
public int read() {
int value = 0; // 읽은 값 저장용
lock.lock(); // 잠금
try {
while (queue.isEmpty()) { // 조건 검사
// 현재 버퍼가 비어있다면(읽을게 없음)
System.out.println("\t\t\tWait for notEmpty condition");
notEmpty.await(); // notEmpty 시그널 기다림
}
value = queue.remove(); // 읽기 + 버퍼에서 삭제
notFull.signal(); // notFull 시그널 전송
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock(); // 잠금 해제
return value; // 읽은 값 반환
}
}
}빈 버퍼를 읽으려 할 경우에는 notEmpty 시그널을 기다리고 꽉 찬 버퍼에 값을 쓰려 할 경우에는 notFull 시그널을 기다린다.
Task Class
ProducerTask
private static class ProducerTask implements Runnable {
@Override
public void run() {
try {
int i = 1;
while (true) {
System.out.println("Producer writes " + i);
buffer.write(i++);
Thread.sleep((int)(Math.random() * 10000));
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}ConsumerTask
private static class ConsumerTask implements Runnable {
@Override
public void run() {
try {
while (true) {
System.out.println("\t\t\tConsumer reads " + buffer.read());
Thread.sleep((int)(Math.random() * 10000));
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}이번 코드에서도 동일하게 의도적으로 한쪽이 기다리는 상황을 만들기 위해 sleep 시간을 무작위로 맞지않게 설정하였다.
Main.java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static Buffer buffer = new Buffer();
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2);
executor.execute(new ProducerTask());
executor.execute(new ConsumerTask());
executor.shutdown();
}
private static class ProducerTask implements Runnable {
@Override
public void run() {
try {
int i = 1;
while (true) {
System.out.println("Producer writes " + i);
buffer.write(i++);
Thread.sleep((int)(Math.random() * 10000));
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
private static class ConsumerTask implements Runnable {
@Override
public void run() {
try {
while (true) {
System.out.println("\t\t\tConsumer reads " + buffer.read());
Thread.sleep((int)(Math.random() * 10000));
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
private static class Buffer {
private static final int CAPACITY = 1;
private java.util.LinkedList<Integer> queue = new java.util.LinkedList<>();
private static Lock lock = new ReentrantLock();
private static Condition notEmpty = lock.newCondition();
private static Condition notFull = lock.newCondition();
public void write(int value) {
lock.lock();
try {
while (queue.size() == CAPACITY) {
System.out.println("Wait for notFull condition");
notFull.await();
}
queue.offer(value);
notEmpty.signal();
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock();
}
}
public int read() {
int value = 0;
lock.lock();
try {
while (queue.isEmpty()) {
System.out.println("\t\t\tWait for notEmpty condition");
notEmpty.await();
}
value = queue.remove();
notFull.signal();
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
lock.unlock();
return value;
}
}
}
}Account 코드와 동일하게 2개의 thread를 가진 thread pool을 생성하고 이 pool에 ProducerTask와 ConsumerTask를 등록한다.
ProducerTask와 ConsumerTask는 Queue로 구현된 Buffer에 값을 쓰고 읽는 작업을 무한히 반복한다. 우리는 이 과정에서 발생하는 대기 상태를 관찰할 것이다.
아까와의 차이점이 있다면 Buffer에서 Condition 객체가 두 개가 있다는 점이다. 이를 통해 생산자는 소비자를 깨우고, 소비자는 생산자를 깨우는 정확한 협력(Cooperation)이 가능하게 한다.
Wait for notEmpty condition
Producer writes 1
Consumer reads 1
Wait for notEmpty condition
Producer writes 2
Consumer reads 2
Producer writes 3
Consumer reads 3
Producer writes 4
Consumer reads 4
Wait for notEmpty condition
Producer writes 5
Consumer reads 5
Producer writes 6
.
.
.
.위 Producer-Consumer 코드는 Java의 Collection framework에서 제공하는 Blocking Queue의 작동 방식과 동일하다. 즉, Blocking Queue를 버퍼로 사용하면 더 간단하게 위 예제를 작성할 수 있다. 이 내용은 생략하도록 한다.
<참고자료>
탁성우, "플랫폼기반프로그래밍", 부산대학교
