wait를 while 안에서 부르는 이유

wait와 notify로 조건 기다리기에서 만든 우편함은 while (message == null) wait();처럼 반복문 안에서 기다렸습니다. 그런데 기다리는 이유는 '값이 없다' 하나뿐이고 값이 들어오면 알림이 오니, if로 한 번만 확인해도 될 것 같습니다. 비어 있으면 기다리고, 깨어나면 값이 있을 테니 바로 꺼내는 식입니다.

이 글에서는 그 생각이 어디서 틀리는지 실행으로 확인합니다. 소비자 둘이 값 하나를 두고 다투게 만들면 if로 기다린 코드는 빈 큐에서 값을 꺼내려다 예외를 냅니다. 이어서 스레드가 wait에서 돌아오는 이유를 모두 정리하고, 그 이유가 무엇이든 같은 코드로 처리하는 while 반복이 왜 유일하게 안전한 형태인지 설명합니다. 마지막으로 기다리는 조건에 이름을 붙여, 닫힌 뒤에도 남은 작업을 끝까지 처리하는 작업 큐를 만듭니다. 실행 결과는 모두 JDK 21에서 확인했습니다.

소비자 둘, 값 하나

아래 IfGuardedTake의 take()는 큐가 비어 있는지 if로 한 번 확인하고 기다립니다. main은 소비자 스레드 둘을 시작하고, 둘 다 빈 큐 앞에서 기다리기 시작한 뒤에 값을 딱 하나 넣고 notifyAll()을 부릅니다.

이 순서는 entered 래치가 맞춥니다. 각 소비자는 모니터를 쥔 채로 래치를 하나씩 내리고, wait()에 들어가서야 모니터를 놓습니다. 그래서 래치가 0이 되어 main이 깨어나더라도, 두 번째 소비자가 wait()로 모니터를 내려놓기 전에는 main이 put에 들어갈 수 없습니다. 즉 put이 실행될 때는 두 소비자가 모두 대기 집합에 있습니다.

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;

public final class IfGuardedTake {
    private final Queue<String> values = new ArrayDeque<>();
    private final CountDownLatch entered = new CountDownLatch(2);

    synchronized String take() throws InterruptedException {
        if (values.isEmpty()) {
            entered.countDown();
            wait();
        }
        return values.remove();
    }

    synchronized void put(String value) {
        values.add(value);
        notifyAll();
    }

    public static void main(String[] args) throws InterruptedException {
        IfGuardedTake queue = new IfGuardedTake();
        Runnable consumer = () -> {
            try {
                System.out.println(Thread.currentThread().getName() + "=" + queue.take());
            } catch (Exception e) {
                System.out.println(Thread.currentThread().getName() + " failed=" + e.getClass().getSimpleName());
            }
        };
        Thread c1 = Thread.ofPlatform().name("reader-a").start(consumer);
        Thread c2 = Thread.ofPlatform().name("reader-b").start(consumer);
        queue.entered.await();
        queue.put("only-one");
        c1.join(500);
        c2.join(500);
        c1.interrupt();
        c2.interrupt();
    }
}

소비자는 예외를 잡아 failed=와 예외 이름을 출력하므로, 예외가 나도 프로그램은 정상적으로 끝납니다. 끝부분의 join(500)과 interrupt()는 소비자가 끝나지 않을 때를 대비한 안전장치입니다. 이 버전에서는 두 소비자가 모두 스스로 끝나므로 아무 일도 하지 않습니다. 실행 결과는 다음과 같습니다.

reader-a=only-one
reader-b failed=NoSuchElementException

100번 실행했을 때 어느 소비자가 값을 받는지와 두 줄의 순서는 실행마다 달라졌지만, 값 하나와 NoSuchElementException 하나가 나온다는 점은 100번 모두 같았습니다. 줄 순서가 바뀌는 것은 출력이 take()가 끝나 모니터를 놓은 뒤에 일어나기 때문입니다. 값을 먼저 꺼낸 소비자가 먼저 출력한다는 보장은 없습니다. 값을 reader-a가 받은 경우를 단계별로 따라가 보겠습니다.

3단계의 알림은 두 소비자 모두에게 값이 생겼다고 알렸고, 알림을 보낸 순간에는 그 말이 맞았습니다. 하지만 값은 하나뿐이고, 먼저 모니터를 얻은 reader-a가 가져갔습니다. 6단계의 reader-b는 알림을 받았다는 사실만 믿고 큐를 다시 보지 않았기 때문에 빈 큐에서 remove()를 불렀고, ArrayDeque의 remove()는 꺼낼 원소가 없으면 NoSuchElementException을 던집니다.

알림이 전하는 것은 '상태가 바뀌었으니 다시 확인해 보라'는 소식이지, 깨어난 스레드 몫으로 맡겨 둔 값이 아닙니다. '큐에 값이 있다'는 조건은 공유 상태에 대한 그 순간의 사실이고, 다른 스레드가 먼저 움직이면 언제든 다시 거짓이 될 수 있습니다.

if를 while로 바꾸면

고칠 곳은 한 단어입니다. take()의 if를 while로 바꾸면, wait()에서 돌아온 스레드는 다음 줄로 넘어가기 전에 큐가 비었는지 다시 확인합니다.

synchronized String take() throws InterruptedException {
    while (values.isEmpty()) {   // if를 while로 바꾼다
        entered.countDown();
        wait();
    }
    return values.remove();
}

나머지 코드는 그대로 두고 실행하면 다음과 같이 출력됩니다. 40번 실행했을 때 값을 받은 소비자는 실행마다 달라질 수 있었지만, 값 하나와 InterruptedException 하나가 나온다는 점은 언제나 같았습니다.

reader-a=only-one
reader-b failed=InterruptedException

첫 줄은 앞과 같고, 달라진 것은 6단계입니다. 모니터를 얻은 reader-b는 while 조건을 다시 검사해 큐가 비어 있음을 확인하고, 예외를 내는 대신 다시 wait()에 들어갑니다. 반복하면서 래치를 한 번 더 내리지만 래치는 이미 0이므로 아무 일도 일어나지 않습니다. 이제 값을 넣을 스레드가 없으므로 reader-b는 계속 기다리고, main의 join(500)이 시간 제한으로 끝난 뒤 안전장치인 interrupt()가 wait()를 깨워 InterruptedException이 출력됩니다. 둘째 줄이 첫 줄보다 약 0.5초 늦게 나오는 것도 그 때문입니다.

실제 프로그램이라면 기다리던 소비자는 다음 값이 들어올 때 깨어나 그 값을 가져갈 것입니다. 중요한 것은 값을 받지 못한 소비자가 예외로 무너지지 않고, 아직 꺼낼 값이 없다는 사실을 확인한 뒤 다시 기다렸다는 점입니다.

wait에서 돌아오는 네 가지 이유

앞의 실험은 깨어난 스레드보다 다른 스레드가 먼저 움직인 경우였습니다. 스레드가 wait()에서 정상적으로 돌아오는 이유를 모두 모아 보면 다음과 같습니다.

• 다른 스레드가 notify()나 notifyAll()을 불렀습니다. 하지만 앞의 실험처럼, 깨어난 스레드가 모니터를 다시 얻기 전에 다른 스레드가 조건을 다시 거짓으로 만들 수 있습니다.
• 다른 조건을 기다리던 스레드를 깨우려는 notifyAll()이었습니다. 생산자와 소비자가 한 객체에서 기다리면 대기 집합도 하나이므로, 소비자를 깨우려는 알림에 빈자리를 기다리는 생산자도 함께 깨어납니다.
• wait(timeout)처럼 시간 제한을 준 대기에서 시간이 다 됐습니다.
• 아무 이유 없이 깨어났습니다. 이를 가짜 깨움(spurious wakeup)이라고 합니다.

마지막 경우는 이상하게 들리지만 명세가 허용하는 동작입니다. Object.wait API 문서는 스레드가 알림도, 인터럽트도, 시간 만료도 없이 깨어날 수 있다고 밝힙니다. 그러면서 실제로는 드물지만, 애플리케이션은 깨어날 이유가 된 조건을 검사하고 그 조건이 거짓이면 계속 기다리는 방식으로 대비해야 한다고 적고 있습니다. 같은 문서가 권장하는 형태가 바로 wait 호출을 while 반복문으로 감싸는 것입니다. 가짜 깨움은 언제 일어날지 예측할 수도, 일부러 일으킬 수도 없으므로 테스트로 걸러 낼 수 없고, 코드의 모양으로 막는 수밖에 없습니다.

네 가지 이유를 하나씩 구별해 처리할 필요는 없습니다. 어떤 이유로 돌아왔든 할 일은 같기 때문입니다. 모니터를 다시 얻은 지금 조건이 참인지 확인하고, 참이면 진행하고 거짓이면 다시 기다리면 됩니다. while 반복문이 바로 이 일을 합니다. 반면 if는 기다리기 전에 한 번 확인할 뿐이라 네 경우 모두에서 거짓인 조건으로 진행할 수 있습니다. 대기자가 하나뿐이라 처음 두 경우가 생기지 않는 프로그램이라도 가짜 깨움은 막을 수 없으므로, 명세대로라면 if로 기다려도 안전한 경우는 없습니다.

인터럽트는 이 목록에 없습니다. 기다리는 동안 인터럽트되면 wait()는 정상적으로 돌아오지 않고 InterruptedException을 던지므로, 반복문의 조건 검사로 돌아가지 않고 메서드를 빠져나갑니다. 큐를 바꾸기 전에 빠져나가므로 버퍼의 상태도 그대로 남습니다.

while이라는 말 때문에 CPU를 계속 쓰며 도는 바쁜 대기(busy waiting)를 떠올릴 수 있지만 그렇지 않습니다. 반복문의 몸체인 wait()가 모니터를 놓고 잠들기 때문에, 반복은 스레드가 깨어날 때마다 한 번씩만 돌고 깨어나지 않는 동안에는 CPU를 쓰지 않습니다. synchronized 안에서 sleep하면 멈추는 이유의 재시도 반복이 정해진 간격마다 깨어나 조건을 확인한 것과 다른 점입니다.

여러 생산자와 소비자가 함께 쓰는 버퍼

이제 같은 원칙으로 용량이 정해진 버퍼를 만들어 보겠습니다. put은 가득 찬 동안, take는 비어 있는 동안 while로 기다리고, 둘 다 큐를 바꾼 뒤 notifyAll()로 알립니다. 깨어난 스레드가 언제나 조건을 다시 확인하므로 생산자와 소비자가 여럿이어도 안전합니다. put은 기다리기 전에 null부터 거부합니다. 어차피 ArrayDeque는 null을 받지 않으니, 버퍼가 가득 차서 한참 기다린 뒤에야 실패하는 것보다 처음에 바로 알리는 편이 낫기 때문입니다. snapshot()은 실행 결과를 확인하려고 둔 메서드이고, 실제로 값을 주고받는 데는 put과 take만 씁니다.

import java.util.ArrayDeque;
import java.util.List;
import java.util.Objects;
import java.util.Queue;

public final class WhileGuardedBuffer<E> {
    private final Queue<E> queue = new ArrayDeque<>();
    private final int capacity;

    public WhileGuardedBuffer(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException("capacity=" + capacity);
        this.capacity = capacity;
    }

    public synchronized void put(E value) throws InterruptedException {
        Objects.requireNonNull(value, "value");
        while (queue.size() >= capacity) {
            wait();
        }
        queue.add(value);
        notifyAll();
    }

    public synchronized E take() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();
        }
        E value = queue.remove();
        notifyAll();
        return value;
    }

    public synchronized List<E> snapshot() {
        return List.copyOf(queue);
    }

    public static void main(String[] args) throws InterruptedException {
        WhileGuardedBuffer<String> buffer = new WhileGuardedBuffer<>(1);
        Thread first = consumer("first", buffer);
        Thread second = consumer("second", buffer);
        Thread.sleep(30);
        buffer.put("alpha");
        buffer.put("beta");
        first.join();
        second.join();
        System.out.println("remaining=" + buffer.snapshot().size());
    }

    private static Thread consumer(String name, WhileGuardedBuffer<String> buffer) {
        return Thread.ofPlatform().name(name).start(() -> {
            try {
                System.out.println(name + " consumed=" + buffer.take());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
}

main은 용량 1인 버퍼에 소비자 둘을 먼저 시작하고, 30밀리초 쉰 뒤 alpha와 beta를 차례로 넣습니다. sleep(30)은 소비자들이 먼저 기다리기 시작할 시간을 주려는 것일 뿐 반드시 그렇게 된다는 보장은 없지만, 결과는 어느 쪽이든 같습니다. 용량이 1이라서 beta를 넣을 때 alpha가 아직 남아 있으면 main이 빈자리가 생길 때까지 기다리기 때문입니다. 실행 결과는 다음과 같습니다.

first consumed=alpha
second consumed=beta
remaining=0

100번 실행하는 동안 어느 소비자가 어떤 값을 받는지와 앞 두 줄의 순서는 바뀌었습니다. 대부분은 first가 alpha를 받았지만 반대인 실행도 있었습니다. 반면 두 값이 한 번씩 소비되고 마지막 줄이 remaining=0이라는 점은 100번 모두 같았습니다. 그러니 이 프로그램을 검사한다면 'first가 alpha를 받는다'처럼 스케줄에 따라 달라지는 순서가 아니라, 두 값이 정확히 한 번씩 소비되고 버퍼가 비었다는 사실을 확인해야 합니다.

출력만 봐서는 while이 실제로 일했는지 알 수 없습니다. 그래서 깨어난 뒤 조건이 거짓이라 다시 기다린 횟수를 세도록 고친 사본으로 100번 실행해 보았더니, 그런 경우가 한 번 있었습니다. 같은 상황이 if 버전에서 일어나면 어떻게 되는지도 확인했습니다. 두 while을 if로 바꾼 버전은 300번 가운데 295번은 위와 같이 끝났지만, 5번은 소비자 스레드 하나가 NoSuchElementException으로 죽고 마지막 줄이 remaining=1이었습니다. alpha의 알림으로 깨어난 소비자가 모니터를 다시 얻기 전에 다른 소비자가 alpha를 가져갔고, 깨어난 소비자는 beta가 들어오기 전에 모니터를 얻자 다시 확인하지 않고 빈 큐에서 remove()를 부른 것입니다. 뒤늦게 들어온 beta는 가져갈 소비자가 없어 버퍼에 남았습니다.

이런 비율은 컴퓨터와 실행 환경에 따라 달라지지만, 숫자가 보여 주는 것은 if로 쓴 코드가 대부분의 실행에서는 아무 문제 없이 동작한다는 점입니다. 몇 번 실행해서 통과했다고 기다림 코드가 맞다고 할 수 없는 이유입니다. 대기자를 여럿 두어 일부러 경쟁을 만들고 여러 번 실행하는 테스트가 도움이 되지만, 그것도 문제를 드러낼 확률을 높일 뿐입니다. 확실한 방법은 처음부터 while로 쓰는 것입니다.

기다리는 조건을 술어로 적기

기다림 코드를 쓰거나 읽을 때는 각 메서드가 무엇을 기다리는지 참·거짓으로 답하는 식으로 먼저 적어 보면 좋습니다. 이런 식을 조건 술어(condition predicate)라고 합니다. WhileGuardedBuffer라면 put은 '빈자리가 있다', 즉 queue.size() < capacity를 기다리고, take는 '값이 있다', 즉 !queue.isEmpty()를 기다립니다. while의 조건은 이 술어를 뒤집은 것입니다.

알릴 대상도 술어로 정합니다. 상태를 바꾼 뒤에는 그 변경으로 참이 될 수 있는 술어를 기다리는 스레드에게 알려야 합니다. take가 값을 꺼내면 '빈자리가 있다'가 참이 될 수 있으니 put에서 기다리는 쪽에, put이 값을 넣으면 take에서 기다리는 쪽에 알립니다. 한 대기 집합에서는 그쪽만 골라 깨울 수 없으므로 notifyAll()을 씁니다. 반대로 어떤 술어도 참으로 만들 수 없는 변경이라면 알릴 필요가 없습니다.

술어로 생각해 두면 락 도구가 바뀌어도 논리는 그대로입니다. ReentrantLock과 Condition에서 synchronized 대신 명시적 락을 쓰게 되어도, 무엇을 기다리고 상태를 바꾼 뒤 누구에게 알리는지는 같은 술어로 설명됩니다.

술어 검사와 그 뒤의 동작은 한 synchronized 안에 있어야 합니다. 아래처럼 기다리는 부분과 꺼내는 부분을 다른 블록으로 나누면, 두 블록 사이에 모니터가 풀린 틈으로 다른 소비자가 들어와 값을 가져갈 수 있습니다. 반복문으로 확인한 사실이 꺼내는 순간에는 이미 낡은 정보가 되는 것이므로, 결국 if로 기다린 것과 같은 문제가 생깁니다.

public E take() throws InterruptedException {
    synchronized (this) {
        while (queue.isEmpty()) {
            wait();
        }
    }
    // 모니터가 풀린 이 틈에 다른 소비자가 값을 가져갈 수 있다
    synchronized (this) {
        return queue.remove();
    }
}

메서드 이름도 기다리는 방식과 맞아야 합니다. 표준 라이브러리의 BlockingQueue에서 take()는 값이 생길 때까지 기다리고, poll()은 값이 없으면 기다리지 않고 바로 null을 돌려줍니다. 한 메서드가 어떤 때는 기다리고 어떤 때는 바로 돌아온다면, 호출하는 쪽은 그 호출이 얼마나 걸릴지 예측할 수 없습니다. 기다리는지, 얼마나 기다리는지는 메서드 이름과 throws InterruptedException 선언, 시간 제한 매개변수로 드러내는 것이 좋습니다.

더 깊이: 시간 제한이 있는 대기도 while로 감싼다

기다리는 시간에 상한을 두는 경우에도 while이 필요합니다. 아래 poll은 값이 생길 때까지 기다리되, 정해진 시간이 지나면 빈 결과를 돌려줍니다.

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

public final class TimedTakeBuffer<E> {
    private final Queue<E> queue = new ArrayDeque<>();

    public synchronized void put(E value) {
        queue.add(value);
        notifyAll();
    }

    public synchronized Optional<E> poll(long timeout, TimeUnit unit) throws InterruptedException {
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        while (queue.isEmpty()) {
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0) {
                return Optional.empty();
            }
            TimeUnit.NANOSECONDS.timedWait(this, remaining);
        }
        E value = queue.remove();
        notifyAll();
        return Optional.of(value);
    }

    public static void main(String[] args) throws InterruptedException {
        TimedTakeBuffer<String> buffer = new TimedTakeBuffer<>();
        long start = System.nanoTime();
        System.out.println("first=" + buffer.poll(100, TimeUnit.MILLISECONDS));
        long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
        System.out.println("waited at least 100ms=" + (waitedMillis >= 100));
        buffer.put("late");
        System.out.println("second=" + buffer.poll(100, TimeUnit.MILLISECONDS));
    }
}

마감 시각 deadline은 처음에 한 번만 계산하고, 반복할 때마다 남은 시간을 다시 구합니다. 깨어났는데 아직 값이 없으면 남은 시간만큼만 다시 기다리고, 남은 시간이 없으면 빈 결과를 돌려줍니다. 처음 받은 timeout을 매번 그대로 넘기면, 가짜 깨움이나 다른 스레드의 선점 때문에 다시 기다릴 때마다 전체 대기 시간이 늘어납니다. 남은 나노초를 밀리초로 바꿔 wait(long)에 넘기는 것도 위험합니다. 1밀리초보다 적게 남았을 때 값이 0이 되는데, wait(0)은 시간 제한 없이 기다리라는 뜻이기 때문입니다. TimeUnit.timedWait는 남은 시간을 밀리초와 나노초로 나눠 wait(long, int)를 부르고, 나노초가 남아 있으면 시간 제한 대기로 처리되므로 이런 문제가 없습니다.

실행 결과는 다음과 같고, 30번 실행해도 모두 같았습니다.

first=Optional.empty
waited at least 100ms=true
second=Optional[late]

첫 호출은 큐가 비어 있으므로 100밀리초 동안 기다린 뒤 Optional.empty를 돌려줍니다. 둘째 줄은 실제로 100밀리초 이상 기다렸는지를 확인한 결과이며, 정확히 몇 밀리초를 기다렸는지는 실행마다 달라서 출력하지 않았습니다. 셋째 호출은 late가 이미 들어 있으므로 기다리지 않고 바로 꺼냅니다. 실무에서 이런 동작이 필요하다면 BlockingQueue의 poll(timeout, unit)이 같은 일을 해 줍니다.

닫을 수 있는 작업 큐

작업 큐에는 보통 '더 이상 작업이 오지 않는다'는 상태가 필요합니다. 소비자가 '값이 있다'만 기다리면 생산자가 모두 끝난 뒤에도 빈 큐 앞에서 영원히 잠듭니다. 그렇다고 닫혔다는 표시만 보고 바로 끝내면 이미 받아 둔 작업을 처리하지 못합니다. 그래서 소비자가 구별해야 할 술어는 두 가지입니다. 꺼낼 작업이 있다는 canTake와, 닫혔고 남은 작업도 없다는 isDrained입니다.

import java.util.ArrayDeque;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;

public final class CloseableWorkQueue<E> {
    private final Queue<E> jobs = new ArrayDeque<>();
    private boolean closed;

    public synchronized void submit(E job) {
        Objects.requireNonNull(job, "job");
        if (closed) throw new IllegalStateException("queue closed");
        jobs.add(job);
        notifyAll();
    }

    public synchronized Optional<E> takeUntilClosed() throws InterruptedException {
        while (!canTake() && !isDrained()) {
            wait();
        }
        if (isDrained()) return Optional.empty();
        return Optional.of(jobs.remove());
    }

    public synchronized void close() {
        closed = true;
        notifyAll();
    }

    private boolean canTake() {
        return !jobs.isEmpty();
    }

    private boolean isDrained() {
        return closed && jobs.isEmpty();
    }

    public static void main(String[] args) throws InterruptedException {
        CloseableWorkQueue<Integer> queue = new CloseableWorkQueue<>();
        queue.submit(10);
        queue.submit(20);
        queue.close();
        int sum = 0;
        Optional<Integer> next;
        while ((next = queue.takeUntilClosed()).isPresent()) sum += next.orElseThrow();
        System.out.println("sum=" + sum);
    }
}

takeUntilClosed()는 두 술어가 모두 거짓인 동안, 즉 큐가 열려 있는데 비어 있는 동안만 기다립니다. 반복문을 빠져나왔다면 둘 중 하나는 참입니다. 그런데 canTake는 큐가 비어 있지 않다는 뜻이고 isDrained는 비어 있다는 뜻이므로, 둘이 동시에 참일 수는 없습니다. 그래서 isDrained()가 참이면 끝났다는 뜻으로 Optional.empty()를 돌려주고, 아니면 작업을 하나 꺼내 Optional.of로 돌려줍니다. 큐가 닫힌 뒤라도 작업이 남아 있으면 canTake가 참이므로 남은 작업부터 꺼내게 됩니다.

술어를 메서드로 뽑아 이름을 붙인 덕분에, 반복문의 조건이 '꺼낼 것도 없고 끝난 것도 아닌 동안 기다린다'로 읽힙니다. !jobs.isEmpty()와 closed && jobs.isEmpty()를 반복문에 그대로 풀어 썼다면 무엇을 기다리는지 한눈에 알기 어려웠을 것입니다. 일시 중지나 우선순위처럼 조건이 늘어날수록 이름을 붙인 효과가 커집니다.

끝났다는 결과를 Optional.empty()로 돌려주는 것도 의미가 있습니다. 앞 글에서 취소를 null 같은 정상 값으로 바꾸지 말라고 했듯이, 여기서도 정상 결과인 작업, 정상적인 끝인 빈 Optional, 취소인 InterruptedException이 서로 다른 모양으로 돌아옵니다. 또 close()가 notifyAll()을 부른다는 점을 놓치면 안 됩니다. 닫는 것은 isDrained를 참으로 만들 수 있는 상태 변경이므로, 빈 큐 앞에서 기다리던 소비자에게 알려야 합니다.

main은 한 스레드에서 10과 20을 넣고 큐를 닫은 뒤, 빈 결과가 나올 때까지 작업을 꺼내 합계를 구합니다. 출력은 sum=30이고, 스레드가 하나뿐이므로 몇 번을 실행해도 같습니다. 닫은 뒤에도 남아 있던 10과 20을 차례로 꺼내 더했고, 세 번째 호출에서는 큐가 닫혔고 비었으므로 Optional.empty()가 돌아와 반복이 끝났습니다.

sum=30

이 실행에서는 큐가 비었을 때 이미 닫혀 있었으므로 wait()가 한 번도 호출되지 않았습니다. 소비자가 빈 큐 앞에서 기다리는 도중에 큐가 닫히는 경우는 연습 문제 2에서 확인합니다.

코드 리뷰에서 확인할 점

조건을 기다리는 코드를 검토할 때는 다음을 확인합니다.

if로 기다려도 되는 경우를 찾을 필요는 없습니다. 가짜 깨움은 대기자 수와 상관없이 일어날 수 있으므로, 조건을 기다리는 wait는 언제나 while 안에 둡니다. 지금은 대기자가 하나라고 확신해도 코드를 고치는 과정에서 소비자가 늘어날 수 있고, while은 그때도 그대로 맞습니다. Object.wait와 Condition의 API 문서도 모두 반복문 안에서 기다리라고 권하므로, 코드 리뷰에서 wait 앞에 if가 보이면 바로 고쳐야 할 신호로 보면 됩니다.

정리

• wait()에서 돌아왔다는 것은 조건을 다시 확인할 때가 됐다는 뜻일 뿐입니다. 알림을 받은 뒤에도 다른 스레드가 먼저 값을 가져갈 수 있습니다.
• 스레드는 알림, 다른 조건을 위한 notifyAll(), 시간 만료, 가짜 깨움으로 wait()에서 돌아옵니다. while 반복은 이유와 상관없이 조건을 다시 확인하므로 네 경우를 모두 같은 코드로 처리합니다.
• while은 바쁜 대기가 아닙니다. 반복할 때마다 wait()가 모니터를 놓고 잠듭니다. 인터럽트되면 wait()가 InterruptedException을 던지므로 상태를 바꾸지 않고 메서드를 빠져나갑니다.
• 기다리는 조건을 술어로 적고 이름을 붙이면 무엇을 기다리는지, 상태를 바꾼 뒤 누구에게 알릴지가 분명해집니다. 끝나는 상태가 있다면 '꺼낼 것이 있다'와 '닫혔고 비었다'를 따로 둡니다.
• if로 쓴 기다림 코드는 대부분의 실행에서 통과하므로, 몇 번의 실행 결과만으로 맞다고 판단하지 않습니다.

연습 문제

1. 용량이 1인 버퍼를 생산자 스레드 하나와 소비자 스레드 하나만 씁니다. put과 take가 모두 if로 기다리고 상태를 바꾼 뒤 notifyAll()을 부릅니다. 스레드가 둘뿐이니 if로도 충분하다는 주장에 답하세요. 이 구성에서 if를 깨뜨릴 수 있는 것은 무엇인가요?
2. CloseableWorkQueue의 close()에서 notifyAll()을 지우면 어떻게 될까요? 아래 프로그램은 작업자 스레드가 빈 큐 앞에서 기다리기 시작한 뒤 큐를 닫습니다. close()가 notifyAll()을 부를 때와 부르지 않을 때의 출력을 각각 예측하세요. CloseableWorkQueue를 쓰므로 javac -d out src/CloseableWorkQueue.java exercise/CloseWakesWorker.java로 함께 컴파일하고 java -cp out CloseWakesWorker로 실행합니다.

public final class CloseWakesWorker {
    public static void main(String[] args) throws InterruptedException {
        CloseableWorkQueue<Integer> queue = new CloseableWorkQueue<>();
        Thread worker = Thread.ofPlatform().name("worker").start(() -> {
            try {
                System.out.println("result=" + queue.takeUntilClosed());
            } catch (InterruptedException e) {
                System.out.println("worker interrupted");
            }
        });
        while (worker.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        queue.close();
        worker.join(1000);
        System.out.println("alive=" + worker.isAlive());
        worker.interrupt();
    }
}

3. 작업 큐에 pause, resume, close 기능을 넣으세요. 일시 중지된 동안에는 큐에 작업이 있어도 소비자가 꺼내면 안 됩니다. 닫을 때는 남은 작업을 모두 처리한 뒤 빈 결과로 끝내야 하며, 일시 중지 상태가 남아서 종료를 막아서는 안 됩니다. 기다리는 조건은 이름 붙인 메서드로 만드세요.

힌트

1. 알림을 받고 깨어난 스레드의 조건을 다시 거짓으로 만들 수 있는 다른 스레드가 이 구성에 있을까요? wait에서 돌아오는 네 가지 이유를 떠올리세요.
2. close()는 어떤 술어를 참으로 만들 수 있나요? 그 사실을 기다리는 스레드에게 알려 줄 코드는 어디에 있나요?
3. 일시 중지 중에는 작업이 있어도 꺼낼 수 없으므로 '꺼낼 수 있다'는 술어에 paused가 들어가야 합니다.

result=Optional.empty
alive=false

alive=true
worker interrupted

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;

public final class PausableBatchQueueSolution<E> {
    private final Queue<E> queue = new ArrayDeque<>();
    private boolean paused;
    private boolean closed;

    synchronized void submit(E value) {
        if (closed) throw new IllegalStateException("closed");
        if (value == null) throw new NullPointerException("value");
        queue.add(value);
        notifyAll();
    }

    synchronized void pause() {
        if (!closed) paused = true;
    }

    synchronized void resume() {
        paused = false;
        notifyAll();
    }

    synchronized void close() {
        closed = true;
        paused = false;
        notifyAll();
    }

    synchronized Optional<E> take() throws InterruptedException {
        while (!mayConsume() && !drained()) wait();
        if (drained()) return Optional.empty();
        return Optional.of(queue.remove());
    }

    private boolean mayConsume() {
        return !queue.isEmpty() && !paused;
    }

    private boolean drained() {
        return closed && queue.isEmpty();
    }

    public static void main(String[] args) throws InterruptedException {
        PausableBatchQueueSolution<String> jobs = new PausableBatchQueueSolution<>();
        jobs.pause();
        jobs.submit("index");
        Thread worker = Thread.ofPlatform().start(() -> {
            try {
                System.out.println(jobs.take().orElseThrow());
                System.out.println("ended=" + jobs.take().isEmpty());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        jobs.close();
        worker.join();
    }
}

index
ended=true

다음 글

이 글의 버퍼는 상태가 바뀔 때마다 모든 대기자를 깨우는 notifyAll()을 썼습니다. notify와 잘못된 깨움에서는 이를 notify() 하나로 줄이면, 생산자와 소비자가 한 대기 집합을 쓸 때 어떤 문제가 생기는지 살펴보고 notifyAll()의 비용을 따져 봅니다.