JEP 428: Structured Concurrency

Project loom 에서 개발 한 '구조화 된 동시성' 이다

이 기능은 말 그대로 동시성 자체를 구조화 시킨다

개발자는 여러 작업들을 나누고 구조화 하여 소프트웨어의 복잡성을 관리한다

하지만 스레드는 그러한 구조화 기능이 없다.

각 하위 작업들은 독립적으로 실패(예외)하거나 성공한다

만약 어떠한 작업 스레드가 구조화된 프로세스를 가지고 있어서, 하위 작업 중 하나라도 실패하면 실패해야 하는 경우를 생각해보자.
이 때의 실패는 스레드의 수명을 이해하는 것이 굉장히 복잡해진다.

상위 스레드에서 하위 스레드 2개가 돌아가는 형태를 가정해보자
이 때 하위 스레드의 이름을 A, B 라고 하겠다.

• 가정 1: 하위 스레드중 하나가 실패 한 경우
  상위 스레드가 작업을 시작했다.
  이 때 하위 스레드 A가 예외를 발생시켰고 상위 스레드는 그 예외를 받는다.
  그럼 남은 하위 스레드 B 는 어떻게 되는가?
  보통은 기껏해야 리소스 낭비 정도겠지만, 최악의 경우 다른 스레드를 방해할 수도 있다.
• 가정 2: 상위 스레드가 실패한 경우
  같은 가정으로, 하위 스레드 2개를 수행시켰다.
  상위 스레드가 예외를 발생하며 중지되었다.
  이 때 스레드 A, B 는 계속 실행 중일 것이다.
• 가정 3: 빠른 실패
  하위 스레드 2개 중 A 는 오랜 시간이 걸리는 작업이다.
  그런데 B 가 빠른 시간내에 예외를 발생시켰다.
  이 때 상위 스레드는 B 의 실패를 전달받게 되지만 프로세스를 종료하지 않는다.
  A 가 끝날 때 까지 기다리는 것이다.

그럼 이러한 문제를 어떻게 해결 하였을까?

StructuredTaskScope

StructuredTaskScope 클래스는 JEP 428 의 주요 클래스이다.

이 클래스를 사용하면 개발자가 작업을 동시 하위 작업의 그룹으로 구성하고 이들을 하나의 단위로 조정 할 수 있다.

예시를 돌리던 도중 문제 발생

문제 해결

해결 1

/Users/pollra/develop/studies/project_loom_example/src/main/java/com/pollra/project/loom/step/StepTree.java:3: error: package jdk.incubator.concurrent is not visible
import jdk.incubator.concurrent.StructuredTaskScope;
                    ^
  (package jdk.incubator.concurrent is declared in module jdk.incubator.concurrent, which is not in the module graph)

해당 프로젝트에 인큐베이터 기능을 사용하고 있기 때문에 모듈 시스템을 명시적으로 설정해야 한다

모듈 파일 생성

생성 위치는 src/main/java 위치에 생성한다.

module project.loom.example.main {
    requires jdk.incubator.concurrent;
}

해결 2

> Task :Main.main() FAILED
3 actionable tasks: 1 executed, 2 up-to-date
WARNING: Using incubator modules: jdk.incubator.concurrent
Exception in thread "main" java.lang.UnsupportedOperationException: Preview Features not enabled, need to run with --enable-preview
	at java.base/jdk.internal.misc.PreviewFeatures.ensureEnabled(PreviewFeatures.java:49)
	at jdk.incubator.concurrent/jdk.incubator.concurrent.StructuredTaskScope.<init>(StructuredTaskScope.java:302)
	at project.loom.example.main/com.pollra.project.loom.Main.run(Main.java:17)
	at project.loom.example.main/com.pollra.project.loom.Main.main(Main.java:12)

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':Main.main()'.
> Process 'command '/Users/pollra/.sdkman/candidates/java/current/bin/java'' finished with non-zero exit value 1

또 다른 에러가 나왔다

--enable-preview 를 인수로 추가하라는 이야기 이다

이전 글에서 해당 오류를 gradle 명령어를 직접 컨트롤 하는 것으로 해결 했지만 매번 예시를 돌릴 때 마다 불편해서 다른 방법을 찾아봤다

gradle 버전을 낮추고 사용하면 IDE 에서 gradle 버전으로 인한 에러를 해결할 수 있다는 글을 보고 gradle 버전을 낮춰본다

Java 19 는 7.6 부터 지원하기 때문에 7.6 설치

해결 3

IDE 를 통해 구동시키니 이번엔 다른 오류가 떴다.

오류해결에 집중하다보니 기록을 까먹고 오류를 바로 해결해버렸는데, 결과적으로 해결 방법은 다음과 같다.

--add-modules jdk.incubator.concurrent --enable-preview

이제 메인 함수를 실행시켜보자

import jdk.incubator.concurrent.StructuredTaskScope;

import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class Main {
    public static void main(String[] args) {
        Integer run = run();
        System.out.println("run = " + run);
    }

    public static Integer run() {
        StructuredTaskScope scope = new StructuredTaskScope<Object>();
        try {
            Future<User> future1 = scope.fork(() -> findUser());
            Future<Integer> future2 = scope.fork(() -> longTimeFunction());

            scope.join();

            return future1.get().id + future2.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        } finally {
            scope.shutdown();
        }
    }

    public static User findUser() {
        throw new RuntimeException();
//        return new User(1, "pollra");
    }

    public static Integer longTimeFunction() {
        try {
            Thread.sleep(Duration.ofSeconds(5));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 1;
    }

    public record User(
            Integer id,
            String name
    ){}
}

실행은 성공적이다

하지만 문제가 있다. Duration.ofSeconds(5) 로 걸어 둔 sleep 시간을 그대로 다 기다린다는 것이다

가이드에서는 아래와 같은 코드를 제시하고 있다

<T> List<T> runAll(List<Callable<T>> tasks) throws Throwable {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        List<Future<T>> futures = tasks.stream().map(scope::fork).toList();
        scope.join();
        scope.throwIfFailed(e -> e);  // Propagate exception as-is if any fork fails
        // Here, all tasks have succeeded, so compose their results
        return futures.stream().map(Future::resultNow).toList();
    }
}

main 함수의 코드를 조금 변경하여 scope.throwIfFailed 함수를 통해 실패를 기다리지 않고 종료시켜보자

package com.pollra.project.loom;

import jdk.incubator.concurrent.StructuredTaskScope;

import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class Main {
    public static void main(String[] args) {
        Integer run = run();
        System.out.println("run = " + run);
    }

    public static Integer run() {
        var scope = new StructuredTaskScope.ShutdownOnFailure();
        try {
            Future<Integer> future1 = scope.fork(() -> findUser());
            Future<Integer> future2 = scope.fork(() -> longTimeFunction());

            scope.join();
            scope.throwIfFailed(e -> new RuntimeException());

            return future1.get() + future2.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        } finally {
            scope.shutdown();
            scope.close();
        }
    }

    public static Integer findUser() {
        System.out.println("findUser(): 바로 종료 되어야 함");
        throw new RuntimeException();
    }

    public static Integer longTimeFunction() {
        try {
            System.out.println("longTimeFunction(): 기다림");
            Thread.sleep(Duration.ofSeconds(10));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return 1;
    }
}

중요한 변경사항은 아래와 같다 (세세한 변경사항은 기록하지 않았다)

• StructuredTaskScope -> StructuredTaskScope.ShutdownOnFailure
• scope.join(); 코드 다음 줄에
  scope.throwIfFailed(e -> new RuntimeException()); 코드 추가
• Thread.sleep(Duration.ofSeconds(5)); -> 10 초로 변경

아래는 실행 결과이다

시작하는 것을 보기 힘들 정도로 바로 종료되는 모습을 확인 할 수 있다

이렇게 2 project loom 2일차를 마쳐본다

다음은 스케줄러 이론을 공부해 볼 예정이다.