LockSupport의 park·unpark

앞 글의 synchronized는 여러 스레드가 같은 상태를 동시에 고치지 못하게 막는 도구였습니다. 동시성 프로그램에는 이와 다른 종류의 요구도 있습니다. 작업 큐에 일이 들어올 때까지, 또는 다른 스레드가 준비를 마칠 때까지 기다려야 하는 경우입니다. 반복문으로 조건을 쉬지 않고 검사하면 아무 일도 하지 않으면서 CPU를 차지하고, Thread.sleep을 섞어 가끔 검사하면 조건이 채워진 뒤에도 남은 잠만큼 늦게 반응합니다.

필요한 것은 조건이 채워질 때까지 CPU를 내놓고 멈춰 있다가, 조건을 채운 스레드가 깨워 주면 바로 다시 실행되는 방법입니다. java.util.concurrent.locks 패키지의 LockSupport가 가장 낮은 수준에서 이 일을 합니다. LockSupport.park()는 지금 실행 중인 스레드를 멈추고, LockSupport.unpark(t)는 멈춰 있는 스레드 t를 깨웁니다. ReentrantLock, CountDownLatch, CyclicBarrier도 스레드를 재우고 깨울 때 내부에서 이 메서드를 씁니다.

애플리케이션 코드에서 park를 직접 쓸 일은 드뭅니다. 하지만 동작 방식을 알면 이런 도구들이 어떻게 기다리는지, 스레드 덤프에 찍힌 WAITING (parking)이 무슨 뜻인지 이해할 수 있습니다. 앞 글에서 배리어를 락 안에 넣어 멈춘 프로그램의 덤프에서도 CyclicBarrier.await가 결국 LockSupport.park를 호출해 스레드를 재우고 있었습니다. 이 글의 실행 결과도 코어가 2개인 Linux 컴퓨터에서 JDK 21로 얻은 것입니다.

허가증 하나로 동작하는 park와 unpark

park와 unpark의 동작은 허가증(permit)이라는 모델로 설명됩니다. LockSupport API 문서에 따르면 park를 쓰는 스레드마다 허가증이 하나 연결되어 있고, 허가증의 상태는 '있음'과 '없음' 두 가지뿐입니다. unpark(t)는 스레드 t의 허가증을 '있음'으로 만듭니다. park()는 허가증이 있으면 그것을 써서 '없음'으로 만들고 바로 돌아오며, 허가증이 없으면 스레드를 멈춥니다. 멈춘 스레드는 다음 세 가지 중 하나가 일어나면 깨어납니다.

• 다른 스레드가 이 스레드를 대상으로 unpark를 호출합니다.
• 다른 스레드가 이 스레드를 인터럽트합니다.
• 아무 이유 없이 돌아옵니다. 이것을 가짜 깨움(spurious wakeup)이라고 합니다.

이 모델에서 꼭 기억할 성질은 세 가지입니다. 첫째, 허가증은 쌓이지 않습니다. Semaphore의 허가와 달리 unpark를 몇 번 호출하든 허가증은 많아야 하나입니다. 둘째, unpark가 park보다 먼저 와도 되며, 이때는 허가증이 미리 '있음'이 되어 있으므로 다음 park가 멈추지 않고 바로 돌아옵니다. 셋째, park는 무엇 때문에 돌아왔는지 알려 주지 않으므로 돌아온 뒤에는 기다리던 조건을 다시 확인해야 합니다.

호출에 따라 일어나는 일을 표로 정리하면 다음과 같습니다.

unpark에 null을 넘기면 아무 일도 일어나지 않고, 아직 시작하지 않은 스레드에 대한 unpark는 효과가 있다는 보장이 없습니다. 그래서 unpark는 대상 스레드를 start()한 뒤에 보내야 합니다.

먼저 보낸 unpark는 기억되지만 쌓이지 않는다

다음 프로그램으로 표의 내용을 확인해 보겠습니다. main은 작업자를 시작한 뒤 unpark를 두 번 보내고, volatile 필드 unparkDone을 true로 바꿔 이제 park를 시작해도 된다고 알립니다. 작업자는 그 신호를 확인한 뒤 park()를 한 번 호출하고, 이어서 최대 0.3초 동안만 기다리는 parkNanos(300_000_000L)를 호출합니다. 두 호출이 각각 0.1초 이상 걸렸는지를 출력하면 실제로 멈췄는지 알 수 있습니다.

import java.util.concurrent.locks.LockSupport;

public final class OneBitPermitOrder {
    private static volatile boolean unparkDone;

    public static void main(String[] args) throws InterruptedException {
        Thread worker =
                new Thread(
                        () -> {
                            while (!unparkDone) {
                                Thread.onSpinWait();
                            }
                            long t0 = System.nanoTime();
                            LockSupport.park();
                            long t1 = System.nanoTime();
                            LockSupport.parkNanos(300_000_000L);
                            long t2 = System.nanoTime();
                            System.out.println("first-park-waited=" + waited(t0, t1));
                            System.out.println("second-park-waited=" + waited(t1, t2));
                        },
                        "permit-worker");
        worker.start();
        LockSupport.unpark(worker);
        LockSupport.unpark(worker);
        unparkDone = true;
        worker.join();
    }

    private static boolean waited(long from, long to) {
        return to - from >= 100_000_000L;
    }
}

작업자는 main이 unpark를 두 번 보낸 뒤에야 park를 시작합니다. 미리 보낸 두 번의 허가가 두 번의 대기를 모두 건너뛰게 하는지 보세요.

first-park-waited=false
second-park-waited=true

100번 실행해 100번 모두 이렇게 나왔습니다. 첫 번째 park()가 바로 돌아온 것은 main이 미리 보낸 unpark가 허가증으로 남아 있었기 때문입니다. 두 번째 parkNanos는 0.3초를 모두 기다렸습니다. unpark를 두 번 보냈지만 허가증은 하나만 저장되었고, 그 하나를 첫 번째 park가 써 버렸기 때문입니다.

parkNanos는 허가증이 없어도 정해진 시간이 지나면 돌아오는 park이며, 다음 글에서 자세히 다룹니다. 두 번째 줄이 언제나 true라고 단정할 수는 없다는 점도 짚어 둡니다. 두 번째 대기는 가짜 깨움으로 일찍 끝날 수도 있기 때문입니다. 100번 실행하는 동안 그런 일은 없었지만, park가 돌아왔다는 사실만으로는 이유를 알 수 없다는 원칙이 여기에도 그대로 적용됩니다.

작업자가 Thread.onSpinWait()를 부르며 반복문으로 신호를 기다리는 부분이 어색해 보일 수 있습니다. CountDownLatch로 기다리게 하는 편이 자연스러워 보이지만, 그러면 이 실험이 깨질 수 있습니다. 래치의 await()도 내부에서 park를 쓰므로, 래치를 기다리는 동안 미리 보낸 허가증을 써 버릴 수 있기 때문입니다. 그래서 이 예제는 park를 쓰지 않고 volatile 필드를 반복해서 읽으며 기다립니다. 래치를 쓰면 실제로 무슨 일이 생기는지는 다음 절에서 확인합니다.

중간에 다른 대기가 끼면 허가증을 잃는다

다음 프로그램은 같은 실험을 CountDownLatch로 만든 것입니다. 작업자는 mayPark 래치가 열리기를 기다린 뒤 park를 두 번 호출하고, 첫 번째 park에서 돌아오면 firstReturned 래치로 main에 알립니다. main은 작업자를 시작한 뒤, 작업자가 명시적 park에 이르기 전에 unpark를 두 번 보내고 mayPark를 엽니다. 그리고 firstReturned를 기다렸다가 작업자가 20밀리초 안에 끝나는지 확인합니다. 허가증이 둘 저장되었다면 두 번째 park도 바로 돌아와 작업자가 끝날 것이므로, 출력에 wrong-second-permit이라는 이름을 붙였습니다.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;

public final class DoubleUnparkPermitBug {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch mayPark = new CountDownLatch(1), firstReturned = new CountDownLatch(1);
        Thread worker =
                new Thread(
                        () -> {
                            await(mayPark);
                            LockSupport.park();
                            firstReturned.countDown();
                            LockSupport.park();
                        },
                        "double-park");
        worker.start();
        LockSupport.unpark(worker);
        LockSupport.unpark(worker);
        mayPark.countDown();
        firstReturned.await();
        worker.join(20);
        System.out.println("wrong-second-permit=" + !worker.isAlive());
        worker.interrupt();
        worker.join();
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

의도대로라면 첫 번째 park는 미리 보낸 허가증으로 바로 돌아오고, 두 번째 park는 허가증이 없어 멈춥니다. 작업자가 20밀리초 안에 끝나지 않으므로 다음 줄이 출력되고, 마지막의 interrupt가 두 번째 park를 깨워 프로그램이 끝납니다.

wrong-second-permit=false

그런데 160번 실행해 보니 155번은 이렇게 끝났지만 5번은 아무것도 출력하지 않고 멈췄습니다. 멈춘 프로세스의 스레드 덤프에서 두 스레드는 다음 위치에 있었습니다(긴 줄과 필요 없는 줄은 ...로 줄였습니다).

"main" ... waiting on condition
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park([email protected]/Native Method)
	- parking to wait for  <0x00000000a8519810> (a java.util.concurrent.CountDownLatch$Sync)
	...
	at java.util.concurrent.CountDownLatch.await([email protected]/CountDownLatch.java:230)
	at DoubleUnparkPermitBug.main(DoubleUnparkPermitBug.java:20)

"double-park" ... waiting on condition
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park([email protected]/Native Method)
	at java.util.concurrent.locks.LockSupport.park([email protected]/LockSupport.java:371)
	at DoubleUnparkPermitBug.lambda$main$0(DoubleUnparkPermitBug.java:11)
	...

작업자는 11번째 줄, 즉 첫 번째 LockSupport.park()에서 멈춰 있고, main은 20번째 줄의 firstReturned.await()에서 작업자의 신호를 기다립니다. 미리 보낸 허가증이 남아 있었다면 첫 번째 park는 바로 돌아왔어야 합니다. 허가증은 그보다 앞의 mayPark.await() 안에서 사라졌습니다. 작업자가 main의 countDown()보다 먼저 mayPark.await()에 도착하면 래치는 작업자를 내부에서 park로 재우는데, 이 내부 park가 미리 받아 둔 허가증을 써 버립니다.

래치가 아직 닫혀 있으니 작업자는 다시 잠들고, 나중에 main이 래치를 열 때 보낸 unpark도 래치 안의 park가 받아 씁니다. 결국 첫 번째 명시적 park는 허가증 없이 멈추고, 작업자가 신호를 보내지 못하니 main도 영원히 기다립니다. 출력 뒤에 있는 정리용 interrupt는 실행될 기회조차 얻지 못합니다. 160번 가운데 155번이 정상으로 끝난 것은, 대개 작업자가 래치에 도착하기 전에 main이 래치를 먼저 열어 await()가 park 없이 바로 돌아왔기 때문으로 볼 수 있습니다.

도표는 이 프로그램이 갈 수 있는 두 경로를 나란히 보여 줍니다. 래치의 내부 대기가 허가증을 써 버리면 첫 번째 park에서 멈추고, 그렇지 않으면 첫 번째 park가 돌아와 두 번째 park까지 갑니다. 두 번째 경로에서도 출력이 하나로 정해지지는 않습니다. 두 번째 park 역시 가짜 깨움으로 돌아올 수 있으니 wrong-second-permit=true가 나올 가능성이 있고, false도 정확히는 '20밀리초 뒤에 확인했을 때 작업자가 아직 살아 있었다'는 뜻일 뿐 두 번째 park에 들어가 있었다는 직접적인 증거는 아닙니다. 도표의 첫 단계에는 'worker 시작 전'이라고 적혀 있지만, 코드에서 main은 worker.start()를 호출한 다음, 작업자가 명시적 park에 이르기 전에 unpark를 두 번 보냅니다. 앞에서 본 것처럼 아직 시작하지 않은 스레드에 대한 unpark는 효과가 있다는 보장이 없으므로, 순서를 바꿔 start() 전에 보내면 미리 보낸 unpark가 허가증으로 남는다는 이 실험의 전제부터 보장되지 않습니다.

API 문서도 이 위험을 직접 경고합니다. 허가증은 스레드마다 하나뿐이라서, 클래스 로딩처럼 눈에 보이지 않는 경우까지 포함해 중간에 park가 한 번 더 쓰이면 스레드가 응답하지 않게 될 수 있다는 것입니다. 문서는 이를 잃어버린 unpark(lost unpark)라고 부릅니다. 이 프로그램에서 얻을 교훈은 허가증을 '일이 생겼다'는 사실을 담아 두는 저장소로 쓰면 안 된다는 것입니다.

신호를 하나도 잃으면 안 된다면 개수를 세는 도구를 써야 합니다. Semaphore는 허가 개수를, BlockingQueue는 넣은 항목 수를, CountDownLatch는 남은 카운트를 정확히 기억합니다. park를 직접 쓴다면 '일이 생겼다'는 사실을 volatile 필드나 락으로 보호한 큐에 따로 기록하고, park에서 돌아올 때마다 그 기록을 확인해야 합니다. 다음 절의 게이트가 그런 구조입니다.

조건을 반복해서 확인하며 기다리기

park를 안전하게 쓰는 기본 형태는 '조건이 채워지지 않은 동안 park한다'는 반복문입니다. 조건은 다른 스레드가 볼 수 있도록 volatile 필드에 두고, park는 조건이 바뀔 때까지 CPU를 내놓는 수단으로만 씁니다. 다음 ParkedConditionGate는 문이 열릴 때까지 한 스레드를 기다리게 하는 게이트입니다.

import java.util.concurrent.locks.LockSupport;

public final class ParkedConditionGate {
    private volatile boolean open;
    private volatile Thread waiter;

    public void awaitOpen() {
        waiter = Thread.currentThread();
        while (!open) {
            LockSupport.park(this);
            if (Thread.interrupted()) throw new IllegalStateException("gate wait interrupted");
        }
    }

    public void open() {
        open = true;
        Thread target = waiter;
        if (target != null) LockSupport.unpark(target);
    }

    public static void main(String[] args) throws InterruptedException {
        ParkedConditionGate gate = new ParkedConditionGate();
        Thread worker =
                new Thread(
                        () -> {
                            gate.awaitOpen();
                            System.out.println("gate-open");
                        },
                        "condition-waiter");
        worker.start();
        while (worker.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        gate.open();
        worker.join();
    }
}

작업자는 awaitOpen()에서 빠져나온 뒤에만 gate-open을 출력합니다. 그러므로 이 한 줄이 나오고 프로그램이 끝났다면, 작업자가 park에서 깨어나 반복문 조건까지 통과했다는 뜻입니다.

gate-open

30번 실행해 모두 이렇게 나왔습니다. main은 작업자의 상태가 WAITING이 될 때까지 기다렸다가 open()을 호출합니다. open()은 open을 true로 쓴 뒤 등록된 작업자를 unpark하고, park에서 깨어난 작업자는 반복문 조건을 다시 검사해 open이 true인 것을 보고 빠져나옵니다.

이 코드에서 중요한 것은 두 메서드 안의 순서입니다. awaitOpen()은 자신을 waiter에 등록하고, 조건을 검사하고, 그다음에 park합니다. open()은 조건을 먼저 쓰고, 그다음에 등록된 스레드를 깨웁니다. 이 순서 덕분에 두 스레드가 어떤 순서로 실행되든 작업자가 영원히 잠드는 일이 없습니다. 두 경우로 나눠 보겠습니다.

• open()이 작업자의 등록보다 먼저 끝나는 경우: open()은 waiter가 아직 null이라 아무도 깨우지 못합니다. 하지만 작업자는 등록한 뒤 while 조건에서 이미 true인 open을 읽으므로 park하지 않고 지나갑니다.
• 작업자가 open을 false로 읽은 직후에 open()이 실행되는 경우: 작업자는 이미 waiter에 등록되어 있으므로 open()이 그 작업자를 unpark합니다. unpark가 작업자의 park보다 먼저 일어나도 허가증이 남아 있으므로, 작업자의 park는 바로 돌아오고 조건을 다시 검사합니다.

두 번째 경우는 wait와 notify를 조건 없이 쓸 때와 대비됩니다. notify는 그 순간 기다리고 있는 스레드만 깨우므로, 알림이 wait보다 먼저 오면 사라지고 대기자는 계속 잠들 수 있습니다. 허가증은 먼저 온 unpark를 기억하므로, 조건 검사와 park 사이의 짧은 틈에 신호가 와도 잃지 않습니다.

main이 작업자의 WAITING 상태를 기다리는 반복문은 두 번째 경우를 재현하려고 넣은 실험용 코드입니다. 게이트가 올바로 동작하는 데는 필요 없고, 실제 코드에서 스레드 상태 조회를 조건 대신 쓰면 안 됩니다. 이 반복문에는 시간 제한도 없어서, 작업자가 어떤 이유로든 WAITING이 되지 않으면 main이 계속 돌게 됩니다.

park에서 돌아왔을 때 인터럽트 확인하기

awaitOpen()은 park에서 돌아올 때마다 Thread.interrupted()로 인터럽트 여부를 확인합니다. park는 인터럽트를 받으면 돌아오지만 InterruptedException을 던지지 않고 인터럽트 플래그도 지우지 않습니다. 그래서 확인하지 않으면 플래그가 켜진 채 반복문이 돌면서 park가 매번 바로 돌아오는 바쁜 반복이 됩니다. 이 게이트는 인터럽트를 받으면 Thread.interrupted()로 플래그를 읽어 지우고 IllegalStateException을 던지는 방식을 택했습니다.

다만 이 정책은 park를 거친 경로에만 적용됩니다. 이미 인터럽트된 스레드가 열린 게이트에서 awaitOpen()을 호출하면 반복문에 들어가지 않으므로 그냥 돌아오고, 플래그도 켜진 채 남습니다. 같은 스레드가 닫힌 게이트에서 호출하면 park가 바로 돌아오고 예외가 납니다. 두 경우를 실행해 보면 앞의 경우는 플래그가 true인 채 정상 반환하고, 뒤의 경우는 플래그가 false로 지워진 채 예외가 났습니다. 진입할 때 인터럽트부터 확인할지, 예외를 던지면서 플래그를 복구할지 같은 정책은 API를 만드는 사람이 정해야 하며, 인터럽트를 다루는 관례는 interrupt 플래그와 InterruptedException 처리에서 다뤘습니다.

한 스레드만 기다릴 수 있는 게이트

이 게이트는 대기자를 waiter 필드 하나에 저장하므로 한 스레드만 기다릴 수 있습니다. 두 스레드가 차례로 awaitOpen()을 호출하면 두 번째 스레드가 waiter를 덮어쓰고, open()은 두 번째 스레드만 깨웁니다. 실제로 두 작업자를 차례로 park시킨 뒤 open()을 호출해 보니, 10번 모두 먼저 기다린 작업자는 1초가 지나도 깨어나지 않았습니다. 여러 대기자를 지원하려면 대기자 목록, 시간 제한, 취소된 대기자 제거까지 직접 만들어야 하고 그만큼 실수할 곳이 늘어납니다. 이런 게이트가 필요하다면 여러 스레드가 함께 기다릴 수 있는 CountDownLatch(1)을 쓰는 편이 낫습니다.

더 깊이: 두 필드가 모두 volatile이어야 하는 이유

작업자는 waiter에 쓰고 open을 읽으며, main은 open에 쓰고 waiter를 읽습니다. 두 필드가 모두 volatile이면 네 접근은 모든 스레드가 동의하는 하나의 순서(JLS의 동기화 순서)로 늘어서는데, 이 순서는 각 스레드의 코드 순서를 따르고 volatile 읽기는 이 순서에서 자기보다 앞선 마지막 쓰기를 봅니다. 작업자의 open 읽기가 main의 open 쓰기보다 뒤에 놓이면 작업자는 true를 보고 park를 건너뜁니다. 앞에 놓이면, 그보다 먼저 일어난 작업자의 waiter 쓰기도 main의 waiter 읽기보다 앞에 있으므로 main이 작업자를 보고 unpark합니다.

둘 중 하나라도 일반 필드라면 이런 전체 순서가 없어서, 작업자는 false를 읽고 main은 null을 읽는 실행도 허용됩니다. 그러면 작업자는 아무도 깨워 주지 않는 park에 들어갑니다. API 문서가 park와 unpark를 언제 호출할지는 volatile이나 원자 변수로 제어해야 믿을 수 있다고 말하는 이유입니다.

blocker로 대기 이유 남기기

ParkedConditionGate는 park()가 아니라 park(this)를 호출합니다. 이때 넘기는 객체를 blocker라고 하며, 스레드가 무엇 때문에 멈춰 있는지 진단 도구에 알려 주는 표시입니다. API 문서는 blocker를 받는 형태를 쓰라고 강하게 권하고, 락을 구현할 때는 보통 this를 넘긴다고 설명합니다. 게이트의 main이 open()을 부르기 전에 3초를 기다리도록 바꾸고 그 사이에 jstack으로 스레드 덤프를 찍으면, 작업자는 다음처럼 보입니다.

"condition-waiter" ... waiting on condition
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park([email protected]/Native Method)
	- parking to wait for  <0x00000000a8519450> (a ParkedConditionGate)
	at java.util.concurrent.locks.LockSupport.park([email protected]/LockSupport.java:221)
	at ParkedConditionGate.awaitOpen(ParkedConditionGate.java:10)
	...

parking to wait for 줄에 찍힌 ParkedConditionGate가 blocker입니다. 덤프를 읽는 사람은 이 스레드가 게이트 객체 때문에 기다린다는 것을 바로 알 수 있습니다. 앞 절의 DoubleUnparkPermitBug 작업자처럼 blocker 없이 park()를 호출하면 이 줄이 나오지 않아, 무엇을 기다리는지 코드를 따라가 봐야 합니다. JDK의 동기화 도구도 blocker를 넘깁니다. 같은 덤프의 main에는 CountDownLatch$Sync가, 앞 글에서 배리어를 기다리던 스레드에는 ConditionObject가 blocker로 찍혀 있었습니다.

blocker는 락이 아닙니다. park는 blocker 객체를 잠그지 않으므로, 다른 스레드가 synchronized (gate)로 같은 객체의 모니터를 얻는 데 아무 영향이 없습니다. 덤프에서도 둘은 다르게 표시됩니다. 모니터를 기다리는 스레드는 BLOCKED (on object monitor) 상태에 waiting to lock 줄로, park한 스레드는 WAITING (parking) 상태에 parking to wait for 줄로 나옵니다. 스레드 덤프를 찍고 읽는 방법은 스레드 덤프와 프로파일러로 장애 원인 찾기에서 다룹니다.

프로그램 안에서 blocker를 읽을 때는 LockSupport.getBlocker(thread)를 씁니다. 다음 프로그램은 작업자가 park해 있는 동안과 끝난 뒤의 blocker를 출력합니다.

import java.util.concurrent.locks.LockSupport;

public final class BlockerPeek {
    public static void main(String[] args) throws InterruptedException {
        Object reason = new Object() {
            @Override
            public String toString() {
                return "order-queue";
            }
        };
        Thread worker = new Thread(() -> LockSupport.park(reason), "parked-worker");
        worker.start();
        while (worker.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        System.out.println("blocker=" + LockSupport.getBlocker(worker));
        LockSupport.unpark(worker);
        worker.join();
        System.out.println("after=" + LockSupport.getBlocker(worker));
    }
}

첫 줄은 작업자가 park해 있는 동안, 둘째 줄은 작업자가 끝난 뒤에 읽은 blocker입니다. reason 객체는 toString()이 order-queue를 돌려주도록 만들었으므로, 이 객체가 blocker로 잡혀 있으면 그 이름이 출력됩니다.

blocker=order-queue
after=null

20번 실행해 모두 이렇게 나왔습니다. 작업자가 park해 있는 동안에는 넘긴 객체가 보이고, 깨어나 끝난 뒤에는 null입니다. API 문서는 이 값이 조회하는 순간의 스냅샷일 뿐이라고 말합니다. 돌려받은 순간 스레드는 이미 깨어났거나 다른 객체로 다시 park했을 수 있으므로, getBlocker의 결과로 동기화를 판단하면 안 되고 진단과 모니터링에만 써야 합니다. blocker는 스레드 덤프 같은 도구에 그대로 드러나므로, 민감한 정보가 담긴 객체 대신 대기 이유를 나타내는 객체를 넘깁니다.

큐와 park로 만든 작업 디스패처

지금까지 본 것을 모아 작업 디스패처를 만들어 보겠습니다. 디스패처는 다른 스레드가 submit으로 넣은 작업을 전용 작업자 스레드 하나가 차례로 처리하는 구조입니다. 허가증은 작업 개수를 기억하지 못하므로, 작업 자체는 큐에 저장하고 park와 unpark는 '큐를 다시 확인하라'는 신호로만 씁니다.

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.LockSupport;

public final class ParkedJobDispatcher {
    private final Queue<String> jobs = new ArrayDeque<>();
    private final Thread worker = new Thread(this::consume, "park-dispatcher");
    private boolean closed;

    public void start() {
        worker.start();
    }

    public void submit(String job) {
        synchronized (jobs) {
            if (closed) throw new IllegalStateException("closed");
            jobs.add(job);
        }
        LockSupport.unpark(worker);
    }

    public void close() throws InterruptedException {
        synchronized (jobs) {
            closed = true;
        }
        LockSupport.unpark(worker);
        worker.join();
    }

    private void consume() {
        for (; ; ) {
            String job;
            synchronized (jobs) {
                job = jobs.poll();
                if (job == null && closed) return;
            }
            if (job == null) {
                LockSupport.park(this);
                continue;
            }
            System.out.println("handled=" + job);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ParkedJobDispatcher d = new ParkedJobDispatcher();
        d.start();
        d.submit("index");
        d.submit("publish");
        d.close();
        System.out.println("dispatcher-closed");
    }
}

main은 작업 두 개를 넣고 close()를 부른 뒤 dispatcher-closed를 출력합니다. close()는 작업자를 join()으로 기다리므로, 마지막 줄이 두 handled 줄보다 뒤에 나오는지 보세요.

handled=index
handled=publish
dispatcher-closed

30번 실행해 모두 이 순서로 나왔습니다. jobs 큐와 closed 플래그는 언제나 synchronized (jobs) 안에서만 읽고 씁니다. ArrayDeque는 여러 스레드가 동시에 고치면 안 되는 클래스이고, closed는 큐가 비었는지와 함께 판단해야 하는 값이기 때문입니다. closed가 volatile이 아니어도 되는 것도, 모든 접근이 같은 모니터 안에 있어서 앞 글의 모니터 규칙으로 값이 보이기 때문입니다. consume이 park(this)로 넘기는 blocker는 디스패처 객체이고 상태를 보호하는 락은 jobs이므로, 둘을 혼동하지 않아야 합니다.

submit은 락 안에서 닫혔는지 확인하고 작업을 넣은 뒤, 락 밖에서 작업자를 unpark합니다. 작업자의 consume은 락 안에서 작업을 하나 꺼내 보고, 큐가 비었으면서 닫혔으면 끝냅니다. 꺼낸 작업이 없으면 락을 놓은 뒤 park하고, 깨어나면 반복문 처음으로 돌아가 큐를 다시 확인합니다. 작업이 있으면 락을 놓은 뒤에 처리합니다. park와 작업 처리를 락 밖에서 하는 것은, park가 쥐고 있던 모니터를 놓지 않으므로 락을 쥔 채 잠들거나 오래 일하면 submit과 close가 락을 얻지 못해 멈추기 때문입니다.

신호를 잃지 않는 이유도 따져 봅시다. 작업자가 큐가 빈 것을 확인하고 락을 놓은 직후, park를 호출하기 직전에 submit이 작업을 넣고 unpark를 보냈다고 해 봅시다. 이 unpark는 허가증으로 남으므로 작업자의 park는 바로 돌아오고, 반복문이 큐를 다시 확인해 새 작업을 찾습니다. submit이 연달아 호출되어 unpark가 하나로 합쳐지더라도 작업 수는 큐가 기억하므로, 작업자는 큐가 빌 때까지 하나씩 꺼내 처리합니다. 허가증은 확인할 기회만 주고, 확인할 내용은 큐가 담는 구조입니다.

close는 락 안에서 closed를 true로 바꾼 뒤 작업자를 unpark하고, join()으로 작업자가 끝나기를 기다립니다. 작업자는 closed가 true여도 큐에 남은 작업을 먼저 처리하고, 큐가 비었을 때 끝납니다. 그래서 close 직전에 넣은 publish도 처리되고, dispatcher-closed는 언제나 두 handled 줄 뒤에 나옵니다. 두 작업이 넣은 순서대로 처리되는 것은 main 하나가 차례로 넣었고 작업자 하나가 큐 앞에서부터 꺼내기 때문입니다.

이 디스패처는 원리를 보이기 위한 것이라 빠진 것이 있습니다. 작업자가 외부에서 인터럽트를 받으면 플래그가 켜진 채 남아, 큐가 빈 동안 park가 계속 바로 돌아오는 바쁜 반복에 빠집니다. 또 start() 전에 close()를 호출하면 시작하지 않은 스레드에 대한 join()은 바로 돌아오므로, 남은 작업이 처리되지 않은 채 close가 끝납니다. 이런 경우까지 다룬 완성된 도구가 BlockingQueue이므로, 실제 코드에서는 ArrayBlockingQueue 같은 구현을 쓰는 편이 낫습니다.

park를 직접 써야 할 때

정리하면 LockSupport는 동기화 도구를 만드는 사람을 위한 부품입니다. 기다릴 조건을 락, 세마포어, 래치, 블로킹 큐로 표현할 수 있다면 그 도구를 쓰는 편이 안전합니다. 이 도구들은 여러 대기자, 시간 제한, 인터럽트, 가짜 깨움을 이미 모두 처리해 두었습니다. park를 직접 쓰는 것은 새로운 동기화 도구나 스케줄러, 특수한 저수준 라이브러리처럼 대기자 목록과 취소를 직접 관리해야 할 때로 한정하는 것이 좋습니다. 그런 경우에도 다음 규칙은 지켜야 합니다.

• 기다릴 조건은 volatile 필드나 락으로 보호한 상태에 두고, 허가증을 조건 대신 쓰지 않습니다.
• park는 반복문 안에서 호출하고, 돌아올 때마다 조건과 인터럽트 상태를 다시 확인합니다. park가 돌아온 이유는 unpark, 인터럽트, 가짜 깨움 중 무엇이든 될 수 있고, 시간 제한 park라면 시간 초과일 수도 있습니다.
• 조건을 바꾸는 쪽은 조건을 먼저 쓰고 그다음에 unpark합니다. 기다리는 쪽은 자신을 등록한 다음에 조건을 검사합니다.
• 신호의 개수가 중요하면 허가증 대신 큐나 세마포어로 개수를 셉니다.
• 진단을 위해 park(blocker) 형태를 쓰고, 보통 this를 넘깁니다.

정리

• park()는 허가증이 있으면 그것을 쓰고 바로 돌아오고, 없으면 unpark, 인터럽트, 가짜 깨움 중 하나가 일어날 때까지 멈춥니다. unpark(t)는 t의 허가증을 '있음'으로 만듭니다.
• 허가증은 많아야 하나이고 쌓이지 않습니다. park보다 먼저 온 unpark는 기억되지만, 두 번 보낸 unpark도 허가증 하나일 뿐입니다.
• park와 unpark 사이에 래치 대기 같은 다른 park가 끼면 그쪽에서 허가증을 써 버려 신호를 잃을 수 있습니다.
• 그래서 기다릴 조건은 volatile 필드나 락으로 보호한 상태에 두고, park는 반복문 안에서 조건을 다시 확인하는 방식으로만 씁니다.
• park(blocker)로 넘긴 객체는 스레드 덤프의 parking to wait for 줄에 나타나 대기 이유를 알려 줍니다. blocker는 락이 아닙니다.

연습 문제

1. main이 ready = true를 기록하고 작업자를 unpark한 다음에 작업자가 park 지점에 도착해도 정상적으로 끝나는 프로그램을 작성하세요. ready는 volatile이어야 하고, 작업자는 while 반복문에서 조건을 검사해야 합니다.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;

public final class EarlyUnparkConditionSolution {
    private static volatile boolean ready;

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch enter = new CountDownLatch(1);
        Thread worker =
                new Thread(
                        () -> {
                            await(enter);
                            while (!ready) {
                                LockSupport.park();
                            }
                            System.out.println("observed-ready");
                        },
                        "early-signal-worker");
        worker.start();
        ready = true;
        LockSupport.unpark(worker);
        enter.countDown();
        worker.join();
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

observed-ready

2. 작업 3개를 디스패처에 넣으면서 작업자에게 unpark가 세 번 전달되었습니다. 허가증은 최대 몇 개까지 저장되고, 디스패처는 작업을 몇 개 처리할까요? 작업자의 park가 두 번 돌아왔다면 허가증이 두 개 저장되어 있었다고 말할 수 있을까요?

3. ParkedConditionGate의 open()을 다음처럼 '깨운 뒤 조건 쓰기' 순서로 바꾸면 어떤 문제가 생길 수 있을까요?

public void open() {
    Thread target = waiter;
    if (target != null) LockSupport.unpark(target);
    open = true;
}

다음 글

park와 interrupt, BLOCKED와 WAITING에서는 park가 인터럽트를 받았을 때 플래그가 어떻게 남는지, parkNanos로 기다리는 시간에 상한을 두는 방법, 그리고 모니터를 기다리는 BLOCKED 상태와 park로 기다리는 WAITING 상태의 차이를 살펴봅니다.