[Java] 기존 동시성 프로그래밍의 한계와 새롭게 도입될 구조적 동시성(Structured Concurrency)

 

 

1. 기존 동시성 프로그래밍의 한계와 새롭게 도입될 구조적 동시성
(Structured Concurrency)

---

[ 비구조적 동시성의 한계 ]

ExecutorService를 이용한 동시성 적용

개발을 하다 보면 하나의 작업이 여러 개의 하위 작업(Task)들로 나누어지는 경우가 있다. 일반적인 단일 스레드 애플리케이션에서는 하위 작업들이 순차적으로 실행될 것이다. 하지만 만약 각각의 작업이 서로 독립적이고 하드웨어 리소스가 충분하다면, 이를 동시에 실행하여 전체 작업을 더 빠르게 그리고 더 적은 지연 시간으로 처리할 수 있을 것이다.

 

 

예를 들어 User와 Order API를 호출하여 얻어온 결과를 사용하는 코드가 있다고 하자. 이때 각각의 I/O 작업이 자체 스레드에서 동시에 실행된다면 더 빠를 것이므로, Java5에 도입된 ExecutorService를 활용하여 간편하게 동시성을 확보할 수 있다. ExecutorService는 바로 Future 객체를 반환하고 동시에 요청을 실행시키는데, get이라는 블로킹 요청을 호출하여 하위 작업들을 join 할 수 있다.

Response handle() throws ExecutionException, InterruptedException {
    Future<String>  user  = esvc.submit(() -> findUser());
    Future<Integer> order = esvc.submit(() -> fetchOrder());
    String theUser  = user.get();   // Join findUser
    int    theOrder = order.get();  // Join fetchOrder
    return new Response(theUser, theOrder);
}

 

 

 

ExecutorService를 이용한 비구조적 동시성의 문제

Java 21에 도입된 가상 스레드를 통해 각 I/O 작업에 스레드를 할당하는 것이 비용-효율적이게 개선 되었지만, 그 결과로 생길 수 있는 엄청난 수의 스레드를 관리하는 것은 여전히 어렵다. 왜냐하면 각각의 하위 작업이 동시에 실행되기 때문에, 각각은 독립적으로 성공 또는 실패할 수 있기 때문이다. 여기서 실패는 예외가 던져짐을 의미하며, 위의 코드에서는 작업에서 예외가 발생할 경우 스레드의 라이프사이클을 이해하는 것은 상당히 복잡할 수 있다.

• findUser에서 예외가 발생한 경우
  • handle 메서드는 예외를 던지지만, fetchOrder는 별도의 스레드에서 계속 실행됨
  • 최상의 경우 스레드 누수로 끝나겠지만, 최악의 경우 fetchOrder가 다른 작업을 방해할 수 있음
• findUser가 실행되는 동안 fetchOrder에서 예외가 발생한 경우
  • handle 메서드는 취소되지 않고 user.get을 위해 블로킹되어 불필요하게 대기함
  • findUser가 완료되고 user.get이 값을 반환한 후에야 order.get이 예외를 던지고 handle이 실패함
• handle 메서드에서 예외가 발생한 경우
  • handle을 실행하는 스레드가 중단되어도 중단이 하위 작업들로 전파되지 않음
  • handle 메서드가 실패하여도 findUser와 fetchOrder를 위한 스레드는 계속해서 실행되며, 모두 스레드 누수가 발생함

 

 

문제는 프로그램이 작업과 하위 작업의 관계를 따라 논리적으로 구조화되어 있지만, 이러한 관계가 실질적으로 존재하지 않는다는 것이다. 이로 인해 오류가 발생할 여지가 더 많아질 뿐만 아니라 오류를 진단하고 문제를 해결하기가 더 어려워진다. 예를 들어 스레드 덤프와 같은 모니터링 도구는 작업 관계 정보가 없으므로 각각의 개별 스레드 호출 스택에 handle, findUser, fetchOrder를 표시할 것이다.

이렇듯 동시성 구조가 없는 제한없는 동시성 패턴을 비구조적 동시성(Unstructured Concurrency)이라고 한다. 작업과 하위 작업 간의 관계가 유용하더라도 이를 강제하거나 심지어 추적하지 않는다. 그래서 취소 혹은 예외가 발생한 케이스 등에 대비해 개발자가 직접 이에 대한 처리를 해야 한다.

 

 

 

[ 구조적 동시성의 개념과 필요성 ]

구조적 동시성의 개념과 필요성

구조적 동시성(Structured Concurrency)은 작업과 하위 작업 간의 자연스러운 관계를 유지하여 더 읽기 쉽고, 유지 보수가 용이하며, 안정적인 동시성 코드를 작성할 수 있도록 도와주는 동시성 프로그래밍 접근 방식이다. 구조적 동시성이라는 용어는 Martin Sústrik에 의해 만들어졌고, Nathaniel J. Smith에 의해 대중화됐다. 그리고 Erlang의 hierarchical supervisor와 같은 다른 언어의 아이디어가 구조적 동시성의 오류 처리 설계에 영향을 미쳤다고 한다. 코틀린에서 제공하는 코루틴(Coroutine) 역시 구조적 동시성의 구현이라고 볼 수 있다.

즉, 구조적 동시성이란 코드의 구문 구조(syntactic structure)를 사용해 서로 다른 스레드에서 실행 중인 연관된 작업들을 하나의 작업 단위 (single unit of work) 로 취급하는 방식으로, 이를 통해 오류 처리나 취소를 간소화하고 안정성과 관측성을 높이고자 한다.

코드 상으로 작업과 하위 작업 간의 관계가 명확하고 이를 통해 런타임에 관계가 재정의되기 때문에 동시성 프로그래밍이 더 쉽고 안정적이며 관찰하기 쉬워진 것이다. 구문 구조는 하위 작업의 라이프사이클을 표현하고 스레드 내 호출 스택과 유사한 스레드 간의 계층 구조를 런타임에 표현할 수 있게 해준다. 이러한 표현을 통해 오류 전파 및 취소는 물론 동시 진행 중인 프로그램을 의미 있게 관찰할 수 있다.

스레드 취소나 예외 상황에 대한 모든 케이스를 개발자가 직접 처리하는 것은 비효율적이므로 자동화된 API를 제공하여 사용하도록 하는 것이 훨씬 바람직할 것이다. 따라서 자바는 구조적 동시성(Structured Concurrency) API를 도입해 동시성 프로그래밍을 간소화할 수 있는 기능을 제공하고자 하였다.

 

 

 

구조적 동시성의 기본 원칙

구조적 동시성은 다음의 간단한 원칙에서 비롯된다.

“한 작업이 동시 진행 중인 하위 작업들로 분할되면, 그들 모두 같은 위치(작업의 코드 블록)로 돌아간다.”

구조적 동시성에서 하위 작업은 상위 작업을 대신하여 작동하고, 상위 작업은 하위 작업의 결과를 기다리며 실패 여부를 모니터링한다. 다중 스레드를 위한 구조적 동시성의 강력함은 다음의 2가지 아이디어에서 비롯된다.

• 코드 블록을 통해 실행 흐름에 대해 진입점과 출구점을 잘 정의함
• 코드의 구문 중첩을 반영하는 방식으로 작업의 라이프사이클을 엄격하게 중첩시킴

 

코드 블록의 진입점과 종료점이 잘 정의되어 있기 때문에 동시 진행 중인 하위 작업의 라이프사이클은 상위 작업의 구문 블록에 국한된다. 다른 하위 작업의 수명 역시 상위 작업의 수명 내에 중첩되어 있기 때문에 하나의 단위로 추론하고 관리될 수 있다. 따라서 deadline 같은 정책을 전체 작업 트리에 적용하고, 통합 가시성 도구에서 하위 작업을 상위 작업에 종속된 것으로 표시할 수 있다.

 

 

 

구조적 동시성의 목표

자바가 구조적 동시성을 도입하고자 하는 목표는 다음과 같다.

• 스레드 누수 및 취소 지연과 같은 취소와 종료로 인한 위험을 제거할 수 있는 동시성 프로그래밍 스타일을 장려함
• 동시성 코드의 관측성을 높임

 

구조적 동시성을 도입한다고 하여 ExecutorService나 Future 같은 java.util.concurrent 패키지의 구성들을 대체하려고 하는 것은 아니다. 또한 이번 스펙(JEP-453) 작업이 최종적인 구조적 동시성 API를 정의한 것도 아니며, 최종적인 구조적 동시성 API는 향후 JDK 릴리스나 서드 파티 애플리케이션에서 정의할 수 있을 것이다.

 

 

 

[ 가상 스레드와 구조적 동시성 ]

구조적 동시성은 가상 스레드와 매우 잘 어울린다고 볼 수 있다. 가상 스레드는 매우 많이 존재할 수 있을 뿐만 아니라, I/O를 포함하는 모든 동시 동작의 단위까지 표현할 수 있을 만큼 저렴하다. 즉, 서버 애플리케이션은 구조화된 동시성을 사용하여 수천 또는 수백만 건의 요청을 한 번에 처리할 수 있는 것이다. 각 요청을 처리하는 작업을 새로운 가상 스레드에 할당하고, 동시 실행을 위해 하위 작업을 제출하여 작업이 분산되면 각 하위 작업에 새로운 가상 스레드를 할당하는 것이다.

즉, 가상 스레드는 풍부한 스레드를 제공하고, 구조적 동시성은 이를 정확하고 강력하게 조정하며 통합 모니터링 도구에서 개발자가 이해한 대로 스레드를 표시할 수 있게 도와준다. 이를 통해 유지 관리가 가능하고 안정적이며, 관측 가능한 서버 애플리케이션을 더 쉽게 구축할 수 있을 것이다.

 

 

 

 

 

2. Java에 도입될 구조적 동시성 API

---

[ StructuredTaskScope API ]

StructuredTaskScope API 사용 예시

구조적 동시성 API의 주요 클래스는 java.util.concurrent 패키지의 StructuredTaskScope이다. 이 클래스를 통해 하나의 작업을 여러 개의 동시 하위 작업으로 구성하고 이를 하나의 단위로 조정할 수 있다. 하위 작업은 개별적으로 fork된 후에 자신의 스레드에서 실행되며, 하나의 단위로 결합되거나 취소된다. 즉, 하위 작업의 성공 또는 실패가 상위 작업에서 집계되어 처리되는 것이다. StructuredTaskScope는 하위 작업의 수명을 명확한 어휘 범위(Lexical Scope)로 제한하여 작업과 하위 작업의 모든 상호 작용(포크, 결합, 취소, 오류 처리 및 결과 작성)이 이루어지도록 한다. 아래는 StructuredTaskScope를 사용하는 예시 코드이다.

Response handle() throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Supplier<String>  user  = scope.fork(() -> findUser());
        Supplier<Integer> order = scope.fork(() -> fetchOrder());

        scope.join()            // Join both subtasks
            .throwIfFailed();  // ... and propagate errors

        // Here, both subtasks have succeeded, so compose their results
        return new Response(user.get(), order.get());
    }
}

 

 

구조적 동시성을 적용하지 않았을 때와 달리 연관된 스레드의 라이프사이클을 손쉽게 이해할 수 있다. 모든 조건에서 스레드의 수명은 try-with-resources 문으로 제한되며 여러 가지 중요한 속성들을 보장한다.

• 단락을 통한 오류 처리: findUser 또는 fetchOrder 하위 작업 중 하나가 실패하면 다른 하위 작업이 아직 완료되지 않은 경우 취소된다. 이는 ShutdownOnFailure에 의해 구현된 종료 정책에 의해 관리되며, 다른 정책도 적용 가능하다.
• 취소 전파: handle를 실행 중인 스레드가 join을 호출하기 전이나 도중에 중단되면 스레드가 스코프를 종료할 때 두 하위 작업이 자동으로 취소된다.
• 명확성: 위의 코드는 구조가 명확하다. 하위 작업을 설정하고, 완료되거나 취소될 때까지 기다린 다음, 성공 또는 실패 여부를 결정한다.
• 관찰 가능성: 스레드 덤프에는 작업 계층 구조가 명확하게 표시된다. findUser 및 fetchOrder를 실행하는 스레드는 자식 범위로 표시된다.

 

 

 

StructuredTaskScope API 자세히 살펴보기

StructuredTaskScope는 다음과 같은 기능들을 제공한다.

public class StructuredTaskScope<T> implements AutoCloseable {

    public <U extends T> Subtask<U> fork(Callable<? extends U> task);
    public void shutdown();

    public StructuredTaskScope<T> join() throws InterruptedException;
    public StructuredTaskScope<T> joinUntil(Instant deadline) 
        throws InterrupteException, TimeoutException;
    public void close();

    protected void handleComplete(Subtask<? extends T> handle);
    protected final void ensureOwnerAndJoined();

}

 

 

그리고 이를 활용하는 StructuredTaskScope의 작업 흐름은 다음과 같다.

1. 스코프를 생성한다. 스코프를 생성하는 스레드가 오너에 해당한다.
2. fork 메서드를 사용하여 스코프에서 하위 작업을 fork 한다.
3. 하위 작업 또는 스코프의 소유자는 언제든지 스코프의 shutdown 메서드를 호출하여 완료되지 않은 하위 작업을 취소하고 새로운 하위 작업의 fork를 방지할 수 있다.
4. 스코프의 오너는 스코프 내의 모든 하위 작업을 하나의 단위로 결합한다. 그리고 join 메서드를 호출하여 모든 하위 작업이 완료(성공 여부와 무관)되거나 shutdown을 통해 취소될 때까지 기다릴 수 있다. 또는 joinUntil 메서드를 호출하여 주어진 시간까지 기다릴 수 있다.
5. join 후에는 하위 작업의 오류가 있다면 이를 다루고 그 결과를 처리한다.
6. try-with-resources에 의해 암시적으로 스코프를 닫는다. 이렇게 하면 아직 종료되지 않은 경우 스코프가 종료되고, 취소되었지만 아직 완료되지 않은 아직 모든 하위 작업이 완료될 때까지 기다린다.

 

 

 

종료 정책(shutdown policies)

동시 진행 중인 하위 작업을 다룰 때는 불필요한 작업을 피하기 위해 단락 패턴(short-circuiting pattern)을 사용하는 것이 일반적이다. 예를 들어 하위 작업 중 하나가 실패하면 모든 하위 작업을 취소하거나, 반대로 모든 하위 작업 중 하나가 성공하면 모든 하위 작업을 취소하는 것이 합리적일 때가 있다.

StructuredTaskScope의 두 서브클래스, ShutdownOnFailure와 ShutdownOnSuccess는 각각 첫 번째 하위 작업이 실패하거나 성공할 때 스코프를 종료하는 정책으로 이러한 패턴을 지원한다. 종료 정책은 예외를 처리하는 중앙 집중식 방법을 추가로 제공하며, 성공 결과를 가져올 수도 있다. 이는 전체 범위를 하나의 단위로 취급하는 구조화된 동시성의 사상에 부합한다.

다음은 여러 작업을 동시에 실행하고 그중 하나라도 실패하면 실패하는 실패 시의 종료 정책이 있는 StructuredTaskScope이다.

<T> List<T> runAll(List<Callable<T>> tasks) throws InterruptedException, ExecutionException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        List<? extends Supplier<T>> suppliers = tasks.stream().map(scope::fork).toList();
        scope.join()
             .throwIfFailed();  // Propagate exception if any subtask fails
        // Here, all tasks have succeeded, so compose their results
        return suppliers.stream().map(Supplier::get).toList();
    }
}

 

 

다음은 첫 번째 성공한 하위 작업의 결과를 반환하는 성공 시 종료 정책이 있는 StructuredTaskScope이다.

<T> T race(List<Callable<T>> tasks, Instant deadline) throws InterruptedException, ExecutionException, TimeoutException {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<T>()) {
        for (var task : tasks) {
            scope.fork(task);
        }
        return scope.joinUntil(deadline)
                    .result();  // Throws if none of the subtasks completed successfully
    }
}

 

 

위의 두 가지 종료 정책이 기본적으로 제공되는 것으로서, 개발자가 임의의 정책을 만들어 사용할 수도 있다.

 

 

 

 

출처

• JEP-453
• https://www.youtube.com/watch?v=WsCJYQDPrrE