notFull·notEmpty 두 Condition으로 대기실 나누기

앞 글의 SingleConditionBuffer는 ReentrantLock 하나와 Condition 하나(changed)로 만든 유한 큐였습니다. 큐가 가득 차서 기다리는 생산자도, 큐가 비어서 기다리는 소비자도 모두 changed에서 기다립니다. 그래서 상태를 바꾼 스레드는 누가 기다리는지 모르는 채로 signalAll()을 불러 전부 깨워야 했습니다. notify와 잘못된 깨움에서 본 것처럼, 한 명만 깨우는 signal()을 쓰면 하필 같은 역할의 스레드가 깨어나고 정작 진행할 수 있는 반대 역할은 계속 잠들 수 있기 때문입니다.

signalAll()은 안전하지만 낭비가 있습니다. 소비자 여덟 명이 빈 큐 앞에서 기다리는데 생산자가 항목을 하나 넣으면, 여덟 명이 모두 깨어나 차례로 락을 얻고 조건을 확인합니다. 그중 항목을 가져가는 것은 한 명뿐이고 나머지 일곱 명은 다시 잠듭니다. 이 글에서는 생산자와 소비자가 서로 다른 대기실에서 기다리게 해서, 상태를 바꾼 쪽이 반대편 대기실의 스레드 한 명만 깨우도록 큐를 고칩니다. 그다음 신호를 엉뚱한 대기실로 보내면 어떻게 되는지, 큐를 닫을 때는 누구를 깨워야 하는지 확인하고, 깨어나는 횟수가 실제로 얼마나 줄어드는지 세어 봅니다.

조건마다 대기실을 하나씩 만들기

크기가 정해진 큐에는 서로 반대인 조건이 두 개 있습니다. 생산자는 queue.size() < capacity, 즉 빈칸이 있어야 진행할 수 있고, 소비자는 queue.size() > 0, 즉 항목이 있어야 진행할 수 있습니다. 앞 글들에서는 이렇게 스레드가 진행하려면 참이어야 하는 조건을 조건 술어(condition predicate)라고 불렀습니다.

Object의 모니터에는 대기 집합이 하나뿐이지만, ReentrantLock은 newCondition()을 부를 때마다 새 Condition을 만들어 줍니다. 그래서 큐 전체를 락 하나로 보호하면서도 술어마다 대기실을 따로 둘 수 있습니다.

private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();  // 생산자가 빈칸을 기다리는 곳
private final Condition notEmpty = lock.newCondition(); // 소비자가 항목을 기다리는 곳

이름은 기다리는 이유가 아니라 기다림이 끝나는 조건으로 짓는 것이 관례입니다. 생산자는 큐가 '가득 차지 않은(not full)' 상태가 되기를 notFull에서 기다리고, 소비자는 '비어 있지 않은(not empty)' 상태가 되기를 notEmpty에서 기다립니다. Condition API 문서의 예제와 표준 라이브러리의 ArrayBlockingQueue도 같은 이름을 씁니다.

이제 누가 누구를 깨울지는 '어떤 연산이 어떤 술어를 거짓에서 참으로 바꿀 수 있는가'로 정해집니다. put은 항목을 하나 넣으므로 큐를 비어 있지 않은 상태로 만들 수 있습니다. 그래서 put은 항목을 넣은 뒤 notEmpty에 신호를 보냅니다. 반대로 take는 항목을 하나 꺼내 빈칸을 만들므로 notFull에 신호를 보냅니다.

자기가 기다렸던 대기실이 아니라 반대쪽 대기실에 신호를 보낸다는 점이 핵심입니다. 생산자가 깨워야 할 대상은 다른 생산자가 아니라 방금 넣은 항목을 가져갈 소비자이고, 소비자가 깨워야 할 대상은 방금 생긴 빈칸에 항목을 넣을 생산자입니다.

한 명만 깨워도 되는 이유

대기실을 나누면 signalAll() 대신 signal()을 써도 안전해집니다. notEmpty에서 기다리는 스레드는 모두 소비자이고, 모두 같은 조건, 즉 '항목이 생기기'를 기다립니다. 항목이 하나 생겼다면 그것을 가져갈 소비자도 한 명이면 충분하고, 그 한 명이 누구든 상관없습니다. 생산자와 소비자가 한 대기실에 섞여 있을 때는 signal()이 생산자를 고를 위험이 있었지만, 이제 notEmpty에는 소비자밖에 없으므로 그런 일이 생기지 않습니다. Condition API 문서도 유한 버퍼 예제에서 바로 이 이유, 즉 항목이나 빈칸이 생길 때 한 스레드만 깨우는 최적화를 쓰기 위해 두 대기실을 둔다고 설명합니다.

이 규칙에는 전제가 있습니다. 한 번의 상태 변경으로 진행할 수 있는 대기자가 한 명뿐이어야 합니다. put은 항목을 정확히 하나 넣고 take는 하나 꺼내므로 이 전제가 맞습니다. 반면 여러 항목을 한꺼번에 넣는 연산이나, 기다리는 모든 스레드의 조건을 한 번에 바꾸는 큐 닫기에서는 한 명만 깨우면 부족합니다. 큐 닫기는 아래에서 직접 구현하고, 여러 항목을 넣는 경우는 연습 문제에서 다룹니다.

대기실을 나눠도 바뀌지 않는 것이 있습니다. 두 Condition은 반드시 큐를 보호하는 바로 그 락에서 만들어야 하고, await()와 signal()은 그 락을 쥔 상태에서만 부를 수 있습니다. 락 없이 부르면 앞 글에서 본 것처럼 IllegalMonitorStateException이 발생합니다. 깨어난 스레드가 조건을 while로 다시 확인해야 하는 것도 그대로입니다. 신호를 받은 소비자가 락을 다시 얻기 전에 다른 소비자가 먼저 들어와 항목을 가져갈 수 있고, 가짜 깨움(spurious wakeup)도 여전히 일어날 수 있기 때문입니다.

신호를 반대 대기실로 보내지 않으면

규칙을 거꾸로 적용하면 어떻게 되는지 보겠습니다. 아래 클래스는 대기실은 제대로 나눴지만, put이 항목을 넣은 뒤 notEmpty가 아니라 notFull에 신호를 보내고, take도 항목을 꺼낸 뒤 notEmpty에 신호를 보냅니다. 각자 자기 역할의 대기실을 깨우는 셈입니다. 용량은 1로 고정했습니다.

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class ReversedConditionSignal {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final Queue<String> queue = new ArrayDeque<>();

    void put(String value) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (queue.size() == 1) notFull.await();
            queue.add(value);
            notFull.signal();
        } finally {
            lock.unlock();
        }
    }

    String take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (queue.isEmpty()) notEmpty.await();
            String value = queue.remove();
            notEmpty.signal();
            return value;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReversedConditionSignal buffer = new ReversedConditionSignal();
        Thread consumer = Thread.ofPlatform().start(() -> {
            try { System.out.println(buffer.take()); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        });
        Thread.sleep(30);
        buffer.put("stored-but-not-signaled");
        consumer.join(150);
        System.out.println("consumerAlive=" + consumer.isAlive());
        consumer.interrupt();
    }
}

main은 소비자 스레드를 먼저 시작합니다. 소비자는 빈 큐에서 take()를 부르므로 notEmpty에 들어가 기다립니다. 30밀리초 뒤 main이 put()으로 항목을 넣지만 신호는 notFull로 갑니다. notFull에는 기다리는 스레드가 없으므로 이 신호는 아무도 깨우지 못하고 사라집니다. 결국 notEmpty의 소비자는 큐에 항목이 있는데도 계속 잠들어 있습니다.

consumerAlive=true

consumer.join(150)은 소비자가 끝나기를 최대 150밀리초 동안 기다립니다. 소비자가 깨어나지 못하니 시간이 다 되어 돌아오고, isAlive()는 true입니다. 큐에 넣은 stored-but-not-signaled는 끝내 출력되지 않습니다. 마지막 줄의 consumer.interrupt()는 기다리던 소비자의 await()에서 InterruptedException을 일으켜 소비자 스레드를 끝냅니다. 소비자는 데몬이 아닌 스레드라서, 이 줄을 지우면 JVM이 소비자를 기다리느라 끝나지 않습니다. 실제로 지우고 실행하면 프로그램을 강제로 종료해야 했습니다.

JDK 21에서 20번 실행했을 때는 모두 위와 같았습니다. 하지만 Thread.sleep(30)이 소비자가 30밀리초 안에 await()에 들어간다는 보장은 아닙니다. 소비자 스레드가 늦게 출발해 put 뒤에 take를 부르면 이미 항목이 있으므로 기다리지 않고 꺼냅니다. 그러면 항목이 출력되고 consumerAlive=false가 나옵니다. 드물게는 가짜 깨움으로 소비자가 깨어나 항목을 가져갈 수도 있습니다. 실행할 때마다 결과가 달라질 수 있다는 점 때문에 이런 버그는 테스트에서 잘 드러나지 않습니다.

이 코드는 락과 공유 상태를 모두 올바르게 다룹니다. 틀린 것은 상태 변경과 신호의 짝 하나뿐입니다. 그래서 코드를 검토할 때는 queue.add 바로 뒤에 notEmpty.signal()이, queue.remove 바로 뒤에 notFull.signal()이 오는지를 따로 확인하는 것이 좋습니다.

닫을 수 있는 유한 큐 완성하기

이제 올바른 방향으로 신호를 보내는 큐를 완성합니다. 실제로 쓰려면 일이 끝났을 때 큐를 닫는 기능도 필요합니다. 닫힌 큐는 새 항목을 받지 않지만, 소비자는 이미 들어와 있던 항목을 끝까지 꺼낼 수 있어야 합니다. 그리고 남은 항목이 없으면 기다리지 않고 '더는 없다'는 결과를 돌려받아야 합니다. 이 결과는 Optional.empty()로 나타냅니다.

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class DualConditionQueue<E> {
    private final Queue<E> queue = new ArrayDeque<>();
    private final int capacity;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private boolean closed;

    public DualConditionQueue(int capacity) {
        if (capacity < 1) throw new IllegalArgumentException("capacity");
        this.capacity = capacity;
    }

    public void put(E value) throws InterruptedException {
        if (value == null) throw new NullPointerException("value");
        lock.lockInterruptibly();
        try {
            while (queue.size() == capacity && !closed) notFull.await();
            if (closed) throw new IllegalStateException("queue closed");
            queue.add(value);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Optional<E> take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (queue.isEmpty() && !closed) notEmpty.await();
            if (queue.isEmpty()) return Optional.empty();
            E value = queue.remove();
            notFull.signal();
            return Optional.of(value);
        } finally {
            lock.unlock();
        }
    }

    public void close() {
        lock.lock();
        try {
            closed = true;
            notFull.signalAll();
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DualConditionQueue<String> jobs = new DualConditionQueue<>(2);
        jobs.put("compile");
        jobs.put("run");
        jobs.close();
        System.out.println(jobs.take().orElseThrow());
        System.out.println(jobs.take().orElseThrow());
        System.out.println("finished=" + jobs.take().isEmpty());
    }
}

put과 take의 while 조건에 !closed가 붙은 이유는 닫힌 큐 앞에서는 기다리면 안 되기 때문입니다. 생산자는 큐가 가득 찼더라도 닫혔다면 반복문을 빠져나와 IllegalStateException을 던집니다. 소비자는 큐가 비었더라도 닫혔다면 반복문을 빠져나오고, 큐가 비어 있으면 Optional.empty()를 돌려줍니다. 큐에 항목이 남아 있으면 닫혔더라도 하나를 꺼내 돌려주므로, 남은 항목을 모두 비운 뒤에야 빈 결과가 나옵니다. 큐가 null을 받지 않는 것은 꺼낸 값과 빈 결과가 헷갈리지 않게 하려는 것입니다.

close()는 signal()이 아니라 signalAll()을, 그것도 두 대기실 모두에 보냅니다. put과 take가 바꾸는 것은 대기자 한 명이 진행할 수 있는지뿐이지만, closed = true는 기다리는 모든 스레드의 조건을 한꺼번에 바꿉니다. 빈 큐에서 기다리던 소비자는 모두 빈 결과를 받고 나가야 하고, 가득 찬 큐에서 기다리던 생산자는 모두 예외를 받고 나가야 합니다. 한 명만 깨우면 나머지는 닫힌 큐 앞에서 영원히 기다립니다.

compile
run
finished=true

main은 용량 2인 큐에 작업 두 개를 넣고 바로 닫습니다. 닫혀도 남은 항목은 꺼낼 수 있으므로 처음 두 번의 take()는 compile과 run을 돌려줍니다. 세 번째 take()는 큐가 닫혔고 비어 있으므로 기다리지 않고 Optional.empty()를 돌려주고, 그래서 isEmpty()가 true입니다.

이 main은 스레드 하나로 실행하므로, 기다리던 스레드를 close()가 깨우는 경로는 지나가지 않습니다. 그 경로는 다음 프로그램으로 확인합니다. 소비자 세 명이 빈 큐에서 기다리게 한 뒤 큐를 닫습니다.

import java.util.ArrayList;
import java.util.List;

public final class CloseWakesConsumers {
    public static void main(String[] args) throws InterruptedException {
        DualConditionQueue<String> jobs = new DualConditionQueue<>(2);
        List<Thread> consumers = new ArrayList<>();
        for (int i = 1; i <= 3; i++) {
            String name = "consumer-" + i;
            consumers.add(Thread.ofPlatform().daemon().name(name).start(() -> {
                try {
                    System.out.println(name + " got " + jobs.take());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        Thread.sleep(100); // 세 소비자가 빈 큐에서 기다리기 시작할 시간을 줍니다
        jobs.close();
        for (Thread consumer : consumers) {
            consumer.join(1_000);
        }
        long stuck = consumers.stream().filter(Thread::isAlive).count();
        System.out.println("stuck=" + stuck);
    }
}

consumer-1 got Optional.empty
consumer-3 got Optional.empty
consumer-2 got Optional.empty
stuck=0

세 소비자가 모두 깨어나 빈 결과를 받고 끝났으므로, 아직 살아 있는 소비자 수인 stuck은 0입니다. 앞의 세 줄은 실행할 때마다 순서가 바뀝니다. 깨어난 소비자들이 락을 다시 얻는 순서가 정해져 있지 않기 때문입니다. Thread.sleep(100)도 세 소비자가 모두 기다리기 시작했다는 보장은 아닙니다. 하지만 늦게 도착한 소비자는 이미 닫힌 큐 앞에서 기다리지 않고 바로 빈 결과를 받으므로 출력은 같습니다.

close()의 notEmpty.signalAll()을 notEmpty.signal()로 바꾸고 다시 실행하면, 소비자 한 명의 줄과 stuck=2만 출력됩니다. 한 명만 깨어나고 두 명은 계속 기다린 것입니다. 소비자를 데몬 스레드로 만들고 join에 시간 제한을 둔 것은 이렇게 소비자가 남는 경우에도 프로그램이 끝나도록 하기 위해서입니다.

깨어나는 횟수는 얼마나 줄어들까

대기실을 나눈 목적은 쓸데없이 깨어나는 스레드를 줄이는 것이었습니다. 실제로 얼마나 줄어드는지 세어 보겠습니다. 아래 프로그램은 같은 큐를 두 가지 방식으로 돌립니다. split이 false이면 notFull과 notEmpty가 같은 Condition 하나를 가리키고 앞 글처럼 signalAll()을 씁니다. true이면 대기실을 나누고 signal()을 씁니다.

생산자 하나가 항목 24,000개를 넣고 소비자 여덟 명이 3,000개씩 꺼내는 동안, await()에서 돌아온 횟수(wakeups)와 돌아왔지만 조건이 그대로라서 다시 기다린 횟수(rewaits)를 셉니다. 두 카운터는 락을 쥔 상태에서만 바꾸므로 따로 동기화할 필요가 없습니다.

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class WakeupCountProbe {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty;
    private final boolean split;
    private final Queue<Integer> queue = new ArrayDeque<>();
    private final int capacity = 4;
    private long wakeups; // await에서 돌아온 횟수
    private long rewaits; // 돌아왔지만 조건이 그대로라서 다시 기다린 횟수

    WakeupCountProbe(boolean split) {
        this.split = split;
        this.notEmpty = split ? lock.newCondition() : notFull; // false면 대기실 하나
    }

    void put(int value) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                notFull.await();
                wakeups++;
                if (queue.size() == capacity) rewaits++;
            }
            queue.add(value);
            if (split) notEmpty.signal(); else notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    int take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
                wakeups++;
                if (queue.isEmpty()) rewaits++;
            }
            int value = queue.remove();
            if (split) notFull.signal(); else notFull.signalAll();
            return value;
        } finally {
            lock.unlock();
        }
    }

    static void run(boolean split) throws InterruptedException {
        WakeupCountProbe buffer = new WakeupCountProbe(split);
        List<Thread> threads = new ArrayList<>();
        threads.add(Thread.ofPlatform().start(() -> {
            try {
                for (int n = 0; n < 24_000; n++) buffer.put(n);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }));
        for (int c = 0; c < 8; c++) {
            threads.add(Thread.ofPlatform().start(() -> {
                try {
                    for (int n = 0; n < 3_000; n++) buffer.take();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        for (Thread thread : threads) thread.join();
        String label = split ? "two conditions + signal  " : "one condition + signalAll";
        System.out.println(label + " wakeups=" + buffer.wakeups + " rewaits=" + buffer.rewaits);
    }

    public static void main(String[] args) throws InterruptedException {
        run(false);
        run(true);
    }
}

one condition + signalAll wakeups=45463 rewaits=32743
two conditions + signal   wakeups=29585 rewaits=16639

숫자는 실행할 때마다 달라집니다. JDK 21에서 열 번 실행했을 때 대기실이 하나인 쪽은 wakeups가 4만 2천에서 4만 7천 사이, rewaits가 2만 9천에서 3만 5천 사이였습니다. 대기실을 나눈 쪽은 wakeups가 2만 3천에서 3만 사이, rewaits가 8천에서 1만 7천 사이였습니다. 매번 나눈 쪽이 덜 깨어났고, 헛되이 깨어난 횟수는 절반 정도이거나 그보다 적었습니다. 두 방식 모두 모든 스레드가 맡은 개수를 끝까지 처리하고 join()에서 돌아왔으므로, 이 실행들에서 멈춘 스레드는 없었습니다.

대기실이 하나일 때 낭비가 큰 이유는 이 구성에서 큐가 대부분 비어 있기 때문입니다. 소비자가 생산자보다 많으니 여러 소비자가 동시에 기다리고 있고, signalAll()은 항목 하나가 들어올 때마다 그들을 모두 깨웁니다. 항목을 가져가는 것은 한 명뿐이니 나머지는 다시 잠듭니다. 대기실을 나누고 signal()을 쓰면 항목 하나에 소비자 한 명만 깨어납니다.

그런데 나눈 쪽도 rewaits가 0은 아닙니다. 신호를 받은 소비자는 곧바로 실행되는 것이 아니라 락을 다시 얻어야 합니다. 그 사이에 방금 다른 항목을 처리하고 돌아온 소비자가 먼저 락을 얻어 항목을 가져가면, 신호를 받은 소비자는 빈 큐를 보고 다시 기다리게 됩니다. 대기실을 나눠도 while 재검사를 없앨 수 없는 이유가 바로 이것입니다.

효과의 크기는 스레드 구성에 따라 다릅니다. 생산자와 소비자를 네 명씩 두고 용량을 8로 늘려 실행해 보면 두 방식의 차이가 거의 없었고, 대기실이 하나인 쪽이 오히려 덜 깨어난 실행도 있었습니다. 같은 역할의 스레드가 동시에 많이 기다릴수록 대기실을 나누는 효과가 커집니다. 그러니 성능 때문에 구조를 바꾼다면 실제와 비슷한 부하에서 이런 횟수와 처리 시간을 직접 재 보고 판단해야 합니다.

더 깊이: 신호를 받는 순서와 락을 다시 얻는 순서

ReentrantLock.newCondition API 문서에 따르면, ReentrantLock이 만든 Condition은 기다리는 스레드에게 FIFO 순서로 신호를 보냅니다. 즉 signal()은 가장 먼저 기다리기 시작한 스레드를 깨웁니다. 하지만 깨어난 스레드가 락을 다시 얻는 순서는 처음 락을 얻으려는 스레드와 같은 규칙을 따릅니다. 기본인 비공정 락에서는 그 순서가 정해져 있지 않고, 공정 락(new ReentrantLock(true))에서는 가장 오래 기다린 스레드가 우선합니다.

그래서 가장 오래 기다린 소비자가 먼저 깨어나더라도 그 소비자가 항목을 먼저 가져간다는 보장은 없습니다. 위 프로그램을 공정 락으로 바꿔 실행해도 대기실을 나눈 쪽의 rewaits는 1만 7천 안팎으로 0이 되지 않았습니다. 신호를 받기 전부터 락을 얻으려고 줄을 서 있던 스레드가 있으면 그 스레드가 먼저 락을 얻기 때문입니다. 요청이 들어온 순서대로 처리하는 것이 요구 사항이라면 Condition의 신호 순서에 기대지 말고, 요청마다 번호를 매기는 식으로 순서를 직접 관리해야 합니다.

직접 만든 큐와 ArrayBlockingQueue

지금까지 만든 큐의 구조는 표준 라이브러리의 ArrayBlockingQueue와 거의 같습니다. OpenJDK 21의 ArrayBlockingQueue는 원소를 담는 배열과 ReentrantLock 하나, 그리고 이름까지 같은 notEmpty와 notFull 두 Condition을 필드로 가지고 있습니다. 그러니 이 글의 큐를 실무에서 그대로 쓸 이유는 거의 없습니다. 표준 구현은 같은 구조를 충분히 검증된 코드로 제공하고, 시간 제한이 있는 대기나 바로 결과를 돌려주는 메서드까지 갖추고 있습니다.

직접 만드는 것이 의미 있는 경우는 표준 큐로 표현하기 어려운 조건이 필요할 때입니다. 위의 close()처럼 큐 자체에 닫힌 상태를 두고 싶거나, 1번 연습 문제처럼 두 큐 가운데 어디서 꺼낼지 고르는 규칙이 필요한 경우입니다. 그때도 이 글의 규칙은 그대로 지켜야 합니다. 술어마다 대기실을 두고, 상태를 바꾼 뒤에는 반대쪽 대기실에 신호를 보내며, 깨어나면 while로 조건을 다시 확인합니다.

정리

• 유한 큐에는 '빈칸이 있다'(생산자)와 '항목이 있다'(소비자)라는 반대 조건이 있습니다. 같은 ReentrantLock에서 newCondition()을 두 번 불러 조건마다 대기실(notFull, notEmpty)을 따로 둡니다.
• 상태를 바꾼 스레드는 자기 대기실이 아니라 반대쪽 대기실에 신호를 보냅니다. put 뒤에는 notEmpty, take 뒤에는 notFull입니다. 방향이 바뀌면 큐에 항목이 있어도 소비자가 깨어나지 못합니다.
• 한 대기실에 같은 조건을 기다리는 스레드만 있고 한 번의 변경으로 한 명만 진행할 수 있다면 signal()로 충분합니다. 큐 닫기처럼 모든 대기자의 조건을 바꾸는 변경에는 두 대기실 모두에 signalAll()이 필요합니다.
• 대기실을 나누면 같은 역할의 스레드가 많이 기다릴 때 깨어나는 횟수가 크게 줄어듭니다. 그래도 신호를 받은 스레드보다 다른 스레드가 먼저 락을 얻을 수 있으므로 while 재검사는 그대로 필요합니다.

연습 문제

1. 용량이 정해진 큐에 높은 우선순위 작업과 일반 작업이 들어옵니다. 소비자는 높은 우선순위 작업을 먼저 꺼내되, 일반 작업이 영원히 밀려나지 않도록 높은 작업을 최대 세 번 연속으로 꺼낸 뒤에는 일반 작업을 하나 꺼내야 합니다. 빈칸을 기다리는 조건과 작업을 기다리는 조건을 나누고, 큐를 닫으면 기다리던 스레드를 모두 깨우세요.
2. 1번 풀이의 선택 규칙은 그대로 두고 용량만 7로 바꿉니다. 일반 작업 n1, n2와 높은 작업 h1부터 h5까지를 이 순서로 모두 넣은 뒤 큐를 닫습니다. 소비자 한 명이 반복해서 take()를 부를 때 일곱 작업이 나오는 순서를 적고, 세 번 제한이 적용되는 곳과 highStreak이 0으로 돌아가는 곳을 표시하세요.
3. DualConditionQueue에 여러 항목을 한 번에 넣는 메서드를 아래처럼 추가했습니다. 항목을 모두 넣은 뒤 신호를 한 번만 보냅니다. 소비자 세 명이 빈 큐에서 기다리는 중에 다른 스레드가 putAll(List.of("a", "b", "c"))를 부르면, 용량이 10일 때와 2일 때 각각 어떻게 되는지 설명하고 고치는 방법을 제시하세요.

public void putAll(List<E> values) throws InterruptedException {
    lock.lockInterruptibly();
    try {
        for (E value : values) {
            while (queue.size() == capacity && !closed) notFull.await();
            if (closed) throw new IllegalStateException("queue closed");
            queue.add(value);
        }
        notEmpty.signal(); // 다 넣은 뒤 한 번만 알린다
    } finally {
        lock.unlock();
    }
}

4. DualConditionQueue의 close()에서 notFull.signalAll()을 지우면 어떤 스레드가 어떤 상황에서 영향을 받을까요? 용량 1인 큐에 항목이 하나 들어 있고 생산자 하나가 put()에서 기다리는 중에 close()를 부른 경우로 설명하세요.

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class FairPrioritySlotSolution<E> {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition hasWork = lock.newCondition();
    private final Queue<E> high = new ArrayDeque<>();
    private final Queue<E> normal = new ArrayDeque<>();
    private final int capacity;
    private int highStreak;
    private boolean closed;

    FairPrioritySlotSolution(int capacity) {
        if (capacity < 1) throw new IllegalArgumentException("capacity");
        this.capacity = capacity;
    }

    void submit(E value, boolean urgent) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (size() == capacity && !closed) notFull.await();
            if (closed) throw new IllegalStateException("closed");
            (urgent ? high : normal).add(value);
            hasWork.signal();
        } finally {
            lock.unlock();
        }
    }

    Optional<E> take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (size() == 0 && !closed) hasWork.await();
            if (size() == 0) return Optional.empty();
            E value;
            if (!high.isEmpty() && (highStreak < 3 || normal.isEmpty())) {
                value = high.remove();
                highStreak++;
            } else {
                value = normal.remove();
                highStreak = 0;
            }
            notFull.signal();
            return Optional.of(value);
        } finally {
            lock.unlock();
        }
    }

    void close() {
        lock.lock();
        try {
            closed = true;
            notFull.signalAll();
            hasWork.signalAll();
        } finally {
            lock.unlock();
        }
    }

    private int size() { return high.size() + normal.size(); }

    public static void main(String[] args) throws InterruptedException {
        FairPrioritySlotSolution<String> queue = new FairPrioritySlotSolution<>(5);
        queue.submit("normal", false);
        queue.submit("h1", true);
        queue.submit("h2", true);
        queue.close();
        System.out.println(queue.take().orElseThrow());
        System.out.println(queue.take().orElseThrow());
        System.out.println(queue.take().orElseThrow());
    }
}

h1
h2
normal

다음 글

BLOCKED·WAITING 상태로 스레드 덤프 읽기에서는 이렇게 기다리는 스레드들이 스레드 덤프에 어떻게 나타나는지, 그리고 멈춘 프로그램에서 누가 무엇을 기다리는지 찾는 방법을 배웁니다. 그다음 글에서는 이 글의 큐를 표준 ArrayBlockingQueue로 바꿉니다.