정상 종료와 yield

백그라운드 작업자 종료는 요청, 대기 해제, 진행 중 작업 처리, 최종 회수의 네 단계로 봐야 합니다.

volatile 플래그만 바꾸면 큐에서 무기한 기다리는 스레드를 깨우지 못하고, interrupt만 보내면 남은 항목 처리 정책이 없습니다.

printer 사례로 두 신호를 하나의 생명 주기로 묶습니다.

running 플래그와 take 대기

아래 daemon 작업자는 빈 큐의 take()에서 WAITING입니다.

main이 volatile 플래그를 false로 써도 대기가 풀리지 않아 실제로 wrong-alive-after-flag=true가 출력됩니다.

마지막 interrupt는 예제 프로세스가 스레드를 남기지 않기 위한 정리입니다.

lab/FlagOnlyShutdownBug.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class FlagOnlyShutdownBug {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        Control control = new Control();
        Thread worker =
                new Thread(
                        () -> {
                            try {
                                while (control.running) {
                                    System.out.println(queue.take());
                                }
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        },
                        "flag-only-worker");
        worker.setDaemon(true);
        worker.start();
        while (worker.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        control.running = false;
        worker.join(20);
        System.out.println("wrong-alive-after-flag=" + worker.isAlive());
        worker.interrupt();
        worker.join();
    }

    private static final class Control {
        volatile boolean running = true;
    }
}

running을 false로 써도 빈 큐의 대기는 풀리지 않는다
원문의 지점	작업자 조건	확인 범위
빈 큐의 `take()`	WAITING 관찰 뒤 다음 단계로 진행	항목이나 중단 신호를 기다림
`running=false` 뒤 `join(20)`	플래그 쓰기가 take를 깨우지 않음	`wrong-alive-after-flag=true`
`interrupt()` 뒤 `join()`	대기를 빠져나와 작업자가 종료	정리 완료 · 별도 완료 로그는 없음

빈 큐의 take(): 작업자 조건: WAITING 관찰 뒤 다음 단계로 진행
확인 범위: 항목이나 중단 신호를 기다림
running=false 뒤 join(20): 작업자 조건: 플래그 쓰기가 take를 깨우지 않음
확인 범위: wrong-alive-after-flag=true
interrupt() 뒤 join(): 작업자 조건: 대기를 빠져나와 작업자가 종료
확인 범위: 정리 완료 · 별도 완료 로그는 없음

상태 확인 반복에는 자체 상한이 없습니다. daemon 설정이 이 종료 절차를 대신하지 않으며, 표의 마지막 행은 원문 제어 흐름을 설명합니다.

가시성은 해결됐지만 wake-up이 없습니다.

반대로 interrupt만 써도 현재 항목을 버릴지 완료할지 모호합니다.

종료 API가 큐 acceptance와 작업자 상태를 함께 바꿔야 합니다.

프린터 종료 방식

즉시 취소는 현재 출력을 중단하고 큐를 보존·폐기합니다.

비우기 shutdown은 새 요청을 거부하고 이미 받은 항목을 모두 처리한 뒤 끝납니다.

기한 shutdown은 일정 시간 비우기한 뒤 interrupt로 전환합니다.

요구가 다르므로 메서드 이름과 결과도 달라야 합니다.

다음 구현은 poison 명령을 큐에 넣어 비우기 shutdown을 수행합니다. 문서 출력은 실제 장치 작업 대신 콘솔 로그로 나타냅니다.

큐는 FIFO이므로 종료 표식 앞의 문서가 먼저 처리됩니다.

stop 이후 submit을 거부하고 join으로 작업자 종료를 기다립니다.

src/DrainingPrinter.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class DrainingPrinter {
    private static final String STOP = new String("STOP");
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private boolean accepting = true;
    private final Thread worker = new Thread(this::loop, "draining-printer");

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

    public synchronized void submit(String document) {
        if (!accepting) throw new IllegalStateException("printer stopping");
        queue.add(document);
    }

    public void shutdown() throws InterruptedException {
        synchronized (this) {
            if (accepting) {
                accepting = false;
                queue.add(STOP);
            }
        }
        worker.join();
    }

    private void loop() {
        try {
            for (; ; ) {
                String job = queue.take();
                if (job == STOP) break;
                System.out.println("printed=" + job);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("printer interrupted, pending=" + queue.size());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DrainingPrinter p = new DrainingPrinter();
        p.start();
        p.submit("java");
        p.submit("thread");
        p.shutdown();
        System.out.println("printer-stopped");
    }
}

marker는 외부 입력과 식별 정보가 겹치지 않도록 private 객체가 더 적합하지만 큐 타입을 간단히 보이려고 고유 String 인스턴스를 사용했습니다.

제품 구현에서는 sealed 명령이나 별도 closeable 큐 추상화로 표현합니다.

작업자가 예외로 먼저 종료될 때 shutdown join이 돌아오더라도 남은 항목 처리 실패를 결과로 보고해야 합니다.

즉시 종료의 catch 지점

즉시 취소에서는 interrupt()가 take·sleep을 깨우고 반복문이 반환합니다.

진행 중인 printer 드라이버가 중단 가능하지 않다면 취소가 지연될 수 있습니다.

device API의 시간 제한과 초기화 capability를 확인해야 합니다.

src/ImmediatePrinterStop.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class ImmediatePrinterStop {
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private final Thread worker = new Thread(this::run, "immediate-printer");

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

    void submit(String value) {
        queue.add(value);
    }

    int stopNow() throws InterruptedException {
        worker.interrupt();
        worker.join();
        return queue.size();
    }

    private void run() {
        try {
            while (true) {
                String value = queue.take();
                Thread.sleep(5);
                System.out.println("printed=" + value);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ImmediatePrinterStop p = new ImmediatePrinterStop();
        p.start();
        p.submit("one");
        p.submit("two");
        int pending = p.stopNow();
        System.out.println("pending-after-stop=" + pending);
    }
}

비우기 종료와 즉시 취소는 같은 작업을 보장하지 않는다
비교 지점	`DrainingPrinter`	`ImmediatePrinterStop`
새 입력	shutdown에서 `accepting=false` · 이후 submit 거부	submit에 수용 차단 검사 없음
종료 신호	앞선 문서 뒤에 고유 `STOP`을 넣음	작업자에 `interrupt()` 요청
이미 받은 값	정상 경로에서 java·thread 순서로 로그	one·two의 앞부분만 로그에 남을 수 있음
호출자 회수	무기한 `join()`으로 종료를 기다림	무기한 `join()`으로 종료를 기다림
큐의 잔여 항목	정상 비우기 뒤 큐가 빔	0·1·2 가능 · 꺼낸 뒤 중단된 값은 제외

새 입력: DrainingPrinter: shutdown에서 accepting=false · 이후 submit 거부
ImmediatePrinterStop: submit에 수용 차단 검사 없음
종료 신호: DrainingPrinter: 앞선 문서 뒤에 고유 STOP을 넣음
ImmediatePrinterStop: 작업자에 interrupt() 요청
이미 받은 값: DrainingPrinter: 정상 경로에서 java·thread 순서로 로그
ImmediatePrinterStop: one·two의 앞부분만 로그에 남을 수 있음
호출자 회수: DrainingPrinter: 무기한 join()으로 종료를 기다림
ImmediatePrinterStop: 무기한 join()으로 종료를 기다림
큐의 잔여 항목: DrainingPrinter: 정상 비우기 뒤 큐가 빔
ImmediatePrinterStop: 0·1·2 가능 · 꺼낸 뒤 중단된 값은 제외

두 예제의 출력은 실제 장치 작업이 아닌 콘솔 로그입니다. 즉시 취소의 pending은 모든 미완료 작업 수가 아닙니다. 이미 take한 항목이 sleep 중 중단되면 로그와 큐 개수 모두에 남지 않을 수 있습니다.

미처리 항목은 호출자가 영속성 큐로 넘기도록 반환 타입을 확장할 수 있습니다.

Empty·sleep·yield의 차이

반복문 본문이 비어 있는 while(condition){}는 바쁜 대기 spin입니다.

실행 기회를 받는 동안 CPU를 사용하며 조건 가시성이 보장되지 않으면 끝나지 않을 수 있습니다.

매우 짧은 락-free 대기에서만 Thread.onSpinWait()와 함께 제한적으로 사용하고 일반 큐 대기는 블로킹 기본형을 씁니다.

Thread.sleep(duration)은 현재 스레드에 지정한 시간 동안의 대기를 요청합니다. 타이머와 스케줄러의 영향을 받고, interrupt로 대기가 일찍 끝날 수도 있습니다.

정확한 재개 시각과 다른 스레드 완료는 보장하지 않습니다.

중단 가능하므로 shutdown 통로가 됩니다.

Thread.yield()는 current 실행 가능 스레드가 스케줄러에 양보 의사를 알리는 힌트입니다.

무시될 수 있고 어떤 스레드가 다음인지 정하지 않습니다.

정확성, 공정성, producer-소비자 조정을 yield에 의존시키지 않습니다.

yield와 큐 프로토콜

작업을 여러 개 연속 처리한 뒤 yield()를 호출하면 다른 실행 가능 스레드가 기회를 얻을 수도 있습니다.

그러나 큐의 take()와 락이 이미 명확한 대기·소유권 규칙을 제공하며 yield()는 성능 실험 대상일 뿐입니다.

처리량이 떨어지면 제거할 수 있어야 합니다.

app/BatchedPrinterWorker.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class BatchedPrinterWorker {
    private sealed interface Command permits Print, Stop {}

    private record Print(String value) implements Command {}

    private record Stop() implements Command {}

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Command> queue = new LinkedBlockingQueue<>();
        Thread worker = new Thread(() -> consume(queue), "batch-printer");
        worker.start();
        queue.add(new Print("A"));
        queue.add(new Print("B"));
        queue.add(new Print("C"));
        queue.add(new Stop());
        worker.join();
        System.out.println("batch-complete");
    }

    private static void consume(BlockingQueue<Command> queue) {
        try {
            int batch = 0;
            for (; ; ) {
                Command command = queue.take();
                if (command instanceof Stop) return;
                Print print = (Print) command;
                System.out.println("printed=" + print.value());
                if (++batch % 2 == 0) Thread.yield();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

출력 A, B, C 순서는 FIFO 큐 때문에 고정되지 yield 때문이 아닙니다.

Stop도 같은 channel을 타므로 앞선 출력이 비우기됩니다.

shutdown API의 검증 목록

새 submit이 어느 시점부터 거부되는지, 큐의 대기 항목을 비울지, 진행 중인 항목이 원자적인지, interrupt가 블로킹 호출을 깨우는지, 소유자가 join하는지 확인합니다.

프로세스 종료 훅에서 호출할 때는 훅 자체의 시간 제한과 재진입도 고려합니다.

daemon 지정은 shutdown 구현을 대체하지 않습니다.

사용자 스레드가 프로세스를 붙잡는 문제와 안전하게 상태를 닫는 문제를 따로 해결합니다.

종료 로그는 요청 시각과 실제 작업자 종료 시각을 구분합니다.

연습 문제

큐에 세 작업을 넣고 poison marker로 비우기를 요청하되 100ms 안에 작업자가 끝나지 않으면 interrupt하세요.

메서드 반환 시 작업자는 살아 있지 않아야 합니다.

정답과 두 단계 중단

먼저 정상 비우기 기회를 주고 시간 제한 join 뒤 생존하면 interrupt와 무기한 join으로 정리를 끝냅니다. 이 정리까지 100ms 안에 끝난다는 보장은 없습니다.

exercise/DeadlineDrainSolution.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class DeadlineDrainSolution {
    private static final int STOP = -1;

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
        Thread worker =
                new Thread(
                        () -> {
                            try {
                                for (; ; ) {
                                    int value = queue.take();
                                    if (value == STOP) return;
                                    Thread.sleep(10);
                                    System.out.println("done=" + value);
                                }
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        },
                        "deadline-drain");
        worker.start();
        queue.add(1);
        queue.add(2);
        queue.add(3);
        queue.add(STOP);
        worker.join(100);
        if (worker.isAlive()) {
            worker.interrupt();
            worker.join();
        }
        System.out.println("stopped=" + !worker.isAlive() + ", remaining=" + queue.size());
    }
}

마지막 remaining에는 종료 표식도 포함될 수 있다
출력된 done 로그	가능한 `remaining`	큐와 처리 중 값
없음	3 또는 4	첫 값을 꺼낸 뒤 중단되었을 수도 있음
`done=1`	2 또는 3	다음 값을 꺼냈는지에 따라 달라짐
`done=1`, `done=2`	1 또는 2	셋째 값이 처리 중이었을 수도 있음
`done=1`부터 `done=3`까지	0 또는 1	STOP까지 소비했으면 0 · 남았으면 1

없음: 가능한 remaining: 3 또는 4
큐와 처리 중 값: 첫 값을 꺼낸 뒤 중단되었을 수도 있음
done=1: 가능한 remaining: 2 또는 3
큐와 처리 중 값: 다음 값을 꺼냈는지에 따라 달라짐
done=1, done=2: 가능한 remaining: 1 또는 2
큐와 처리 중 값: 셋째 값이 처리 중이었을 수도 있음
done=1부터 done=3까지: 가능한 remaining: 0 또는 1
큐와 처리 중 값: STOP까지 소비했으면 0 · 남았으면 1

모든 행은 마지막 stopped=true를 전제로 합니다. remaining은 STOP을 포함한 큐 항목 수이며 이미 꺼낸 값은 제외합니다. 100ms 뒤 취소로 전환할 수 있지만 이어지는 무기한 join까지 100ms 안에 끝난다는 뜻은 아닙니다.

graceful shutdown은 플래그 하나나 yield 힌트가 아닙니다.

acceptance 차단, 큐 정책, interrupt 가능한 대기, 시간 제한 전환, 최종 join을 한 소유자가 책임질 때 종료 상태를 예측할 수 있습니다.