Virtual thread pinning issue와 java 24에서 해소 방법

virtual thread

java 21부터 virtual thread가 나온 것은 대부분의 알고 있는 사실이다. 간단하게 virtual thread는 기존 jvm에 사용하는 thread가 os kernal thread에 매핑되어 사용되던 걸 carrier thread (platform thread)에 virtual thread(향후 vt)를 사용하여 내부적 요청에 vt를 carrier thead에 mount, unmont하여 kernal os를 덜 사용하는 전략을 사용하는 thread를 말한다.

 

내부적으로 실행되는 로직을 확인해보자. openjdk에 있는 테스트용 VThreadRunner.java 코드를 이용해서 호출해보면서 내부적으로 VThread에서 사용되는 carrier thread 호출 scheduler는 fork join pool을 사용하는 걸 알 수 있다.

/*
 * Copyright (c) 2022, 2023, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package javas.main;

import java.lang.reflect.Field;
import java.time.Duration;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Helper class to support tests running tasks a in virtual thread.
 */
public class VThreadRunner {
    private VThreadRunner() { }

    /**
     * Characteristic value signifying that initial values for inheritable
     * thread locals are not inherited from the constructing thread.
     */
    public static final int NO_INHERIT_THREAD_LOCALS = 1 << 2;

    /**
     * Represents a task that does not return a result but may throw
     * an exception.
     */
    @FunctionalInterface
    public interface ThrowingRunnable {
        /**
         * Runs this operation.
         */
        void run() throws Exception;
    }

    /**
     * Run a task in a virtual thread and wait for it to terminate.
     * If the task completes with an exception then it is thrown by this method.
     * If the task throws an Error then it is wrapped in an RuntimeException.
     *
     * @param name thread name, can be null
     * @param characteristics thread characteristics
     * @param task the task to run
     * @throws Exception the exception thrown by the task
     */
    public static void run(String name,
                           int characteristics,
                           ThrowingRunnable task) throws Exception {
        AtomicReference<Exception> exc = new AtomicReference<>();
        Runnable target =  () -> {
            try {
                task.run();
            } catch (Error e) {
                exc.set(new RuntimeException(e));
            } catch (Exception e) {
                exc.set(e);
            }
        };

        Thread.Builder builder = Thread.ofVirtual();
        if (name != null)
            builder.name(name);
        if ((characteristics & NO_INHERIT_THREAD_LOCALS) != 0)
            builder.inheritInheritableThreadLocals(false);
        Thread thread = builder.start(target);

        // wait for thread to terminate
        while (thread.join(Duration.ofSeconds(10)) == false) {
            System.out.println("-- " + thread + " --");
            for (StackTraceElement e : thread.getStackTrace()) {
                System.out.println("  " + e);
            }
        }

        Exception e = exc.get();
        if (e != null) {
            throw e;
        }
    }

    /**
     * Run a task in a virtual thread and wait for it to terminate.
     * If the task completes with an exception then it is thrown by this method.
     * If the task throws an Error then it is wrapped in an RuntimeException.
     *
     * @param name thread name, can be null
     * @param task the task to run
     * @throws Exception the exception thrown by the task
     */
    public static void run(String name, ThrowingRunnable task) throws Exception {
        run(name, 0, task);
    }

    /**
     * Run a task in a virtual thread and wait for it to terminate.
     * If the task completes with an exception then it is thrown by this method.
     * If the task throws an Error then it is wrapped in an RuntimeException.
     *
     * @param characteristics thread characteristics
     * @param task the task to run
     * @throws Exception the exception thrown by the task
     */
    public static void run(int characteristics, ThrowingRunnable task) throws Exception {
        run(null, characteristics, task);
    }

    /**
     * Run a task in a virtual thread and wait for it to terminate.
     * If the task completes with an exception then it is thrown by this method.
     * If the task throws an Error then it is wrapped in an RuntimeException.
     *
     * @param task the task to run
     * @throws Exception the exception thrown by the task
     */
    public static void run(ThrowingRunnable task) throws Exception {
        run(null, 0, task);
    }

    /**
     * Returns the virtual thread scheduler.
     */
    private static ForkJoinPool defaultScheduler() {
        try {
            var clazz = Class.forName("java.lang.VirtualThread");
            var field = clazz.getDeclaredField("DEFAULT_SCHEDULER");
            field.setAccessible(true);
            return (ForkJoinPool) field.get(null);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * Sets the virtual thread scheduler's target parallelism.
     * @return the previous parallelism level
     */
    public static int setParallelism(int size) {
        return defaultScheduler().setParallelism(size);
    }

    /**
     * Ensures that the virtual thread scheduler's target parallelism is at least
     * the given size. If the target parallelism is less than the given size then
     * it is changed to the given size.
     * @return the previous parallelism level
     */
    public static int ensureParallelism(int size) {
        ForkJoinPool pool = defaultScheduler();
        int parallelism = pool.getParallelism();
        if (size > parallelism) {
            pool.setParallelism(size);
        }
        return parallelism;
    }
}

 

그리고 contiton을 생성해서 task를 실행시키는데 이때 사용되는 VThreadContinuation을 생성한다. 이 Continuation은 실제 해당 vt가 실행될 때 호출되는 메소드이고 wrap이외에 onPinned가 구현되어 있는데 onPinned는 내부적으로 vt가 pinning되었을 때 실행되는 메소드이다.

 

이 Vt를 실행시키는 runContinuation을 선언하는데 이 runContinuation은 vt가 실행되는 yield상태에서 호출되고 바로 호출 하는 것이 아닌 위에 생성했던 scheduler에 스케줄 처리되면서 실행된다.

condition내부에는 vt를 mount하고 종료되면 unmount 하는 코드가 들어있는 걸 확인할 수 있다.

 

virtual thread의 효과적인 사용처

virtual thread를 사용할 때 cpu intensive한 환경이 아닌 io intensive한 환경에서 사용하라고 한다. 그이유는 virtual thread의 경우 i/o bound 작업에 효율성이 있어 높은 처리량을 주기위해서 설계되었기 때문에 cpu bound가 높은 환경에서는 오히려 오버헤드가 늘어날 수 있다. 또한 virtual thread는 가볍게 여러개 생성되면서 여러 작업을 수행할 수 있게 해주는데 cpu intensive한 환경에서는 어울리지 않으며 컨텍스트 스위칭 비용에 대한 효과를 보기 어렵다.

 

virtual thread의 문제

carrier thread에서는 vt를 할당받아서 내부적으로 수행하게 되는데 synchronized 되어 있는 코드를 수행하게 되면 pinning이 발생하는 문제가 생긴다. 그 이유는 synchronized안에는 동시성 제어를 위한 monitor를 들고 있으며 이 걸 사용해서 lock, unlock을 수행한다. 내부적으로 이 monitor에 대한 소유는 carrier thread가 보유하게 되기 때문에 blocking되는 상황에서는 carrior thread자체가 blocking 되는 이슈가 있다. 이는 synchronized가 jvm 레벨에서 구현되어 있기 때문이다.

 

이해 대한 해결법으로 ReentrantLock을 사용하면 해소가 가능하다고 한다. ReentrantLock는 synchronized와 다르게 FIFO 큐로 관리되어있어 lock에 대한 점유 순서를 보장하고 있기 때문에 VT에서는 효율적인 스케줄링이 가능하기 때문에 필요에 따라서 Carrier Thread에서 mount/unmount가 가능하다. 이는 java level에서 구현되어 있어서 가능한 부분이다.

출처 : https://www.reddit.com/r/java/comments/13ze03y/question_about_virtual_threads_and_their/?rdt=60476

https://stackoverflow.com/questions/78671922/why-reentrantlock-is-better-for-virtual-threads-than-synchronized

From the java community on Reddit

 

넷플릭스에서 찾은 관련된 이슈 

https://netflixtechblog.com/java-21-virtual-threads-dude-wheres-my-lock-3052540e231d

 

한번 테스트 해보자.

 

package javas.main.wedul;

import lombok.Synchronized;

public class Wedul {

    @Synchronized
    public synchronized void start() throws InterruptedException {
        System.out.println("start");
        Thread.sleep(1000);
    }
}

synchronized가 붙어있는 Wedul클래스의 start라는 메소드가 있다고 가정하자

 

이때 두개의 virtual thread가 Wedul.start()를 실행시킨다고 가정해보면 synchronized가 pinning을 발생시키는지 확이할 수 있다.

이때 여러개의 carrier thread가 존재할 경우 vt로 인한 pinning이 아닌 여러개의 carrier thread가 실행될 수 있으니 vm option을 주어 개수를 제한한다. 그리고 trace mode가 켜져있어야 하기 때문에 해당 옵션도 켜준다.

 

-Djdk.virtualThreadScheduler.parallelism=1 
-Djdk.virtualThreadScheduler.maxPoolSize=1 
-Djdk.virtualThreadScheduler.minRunnable=1 
-Djdk.tracePinnedThreads=short

 

그럼 위에 사용했던 VThreadRunner를 사용해서 thread를 실행시켜보자.

package javas.main.wedul;

import javas.main.VThreadRunner;

public class JavaApplication {

    public static void main(String[] args) throws Exception {
        Wedul w = new Wedul();
        VThreadRunner.run(w::start);
        VThreadRunner.run(w::start);
    }


}

 

pinning 이 발생해서 VT 생성시에 추가되었던 VThreadContinuation에 onPinned가 실행된다.

 

 

java 24에서 virtual thread 해소

https://openjdk.org/jeps/491

JEP 491: Synchronize Virtual Threads without Pinning

 

openjdk에 올라온 내용을 보면 jdk24부터는 synchroize를 사용하더라도 pinning이 되지 않도록 수정했다고 하는데 그 주요 내용은 carrier thread가 synchorinzed를 만났을 때 monitor를 점유하는게 아닌 vt가 점유하게 하고 그 vt를 unmount시키도록 하는 방법이다.

 

 

똑같은 코드를 jdk24로 수행해보니 pinning 경고가 사라진걸 확인 할 수 있다.

 

 

결론

virtual thread를 사용한다면 특히 synchronized가 많은 mysql drive를 사용한다면 jdk 24이상부터 사용하자.

 

 

 

위 예시 코드

https://github.com/weduls/virtual-theadtest