Встроенный HTTP-клиент Java 21 закрепляет поток несущей

Question

Я использую Java Corretto 21.0.0.35.1 build 21+35-LTS и встроенный HTTP-клиент Java для получения ответа в виде файла . Я делаю параллельные запросы, используя виртуальные потоки, и по большей части это работает хорошо. Однако иногда во время моего тестирования возникает событие «Закрепление», как видно из трассировки стека ниже.

Я полагал, что JDK был обновлен для полной поддержки виртуальных потоков, и, насколько я понимаю, HTTP-клиент вообще не должен закреплять поток-носитель. Однако похоже, что это событие закрепления иногда происходит при чтении и (автоматическом) закрытии файла.InputStream.

Ожидается ли такое поведение или это все еще ошибка в JDK?

Код:

      HttpResponse<InputStream> response = httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream());
try (InputStream responseBody = response.body()) {
  return parser.parse(responseBody); // LINE 52 in the trace below
}

След

      * Pinning event captured:
  java.lang.VirtualThread.parkOnCarrierThread(java.lang.VirtualThread.java:687)
  java.lang.VirtualThread.park(java.lang.VirtualThread.java:603)
  java.lang.System$2.parkVirtualThread(java.lang.System$2.java:2639)
  jdk.internal.misc.VirtualThreads.park(jdk.internal.misc.VirtualThreads.java:54)
  java.util.concurrent.locks.LockSupport.park(java.util.concurrent.locks.LockSupport.java:219)
  java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.util.concurrent.locks.AbstractQueuedSynchronizer.java:754)
  java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.util.concurrent.locks.AbstractQueuedSynchronizer.java:990)
  java.util.concurrent.locks.ReentrantLock$Sync.lock(java.util.concurrent.locks.ReentrantLock$Sync.java:153)
  java.util.concurrent.locks.ReentrantLock.lock(java.util.concurrent.locks.ReentrantLock.java:322)
  sun.nio.ch.SocketChannelImpl.implCloseNonBlockingMode(sun.nio.ch.SocketChannelImpl.java:1091)
  sun.nio.ch.SocketChannelImpl.implCloseSelectableChannel(sun.nio.ch.SocketChannelImpl.java:1124)
  java.nio.channels.spi.AbstractSelectableChannel.implCloseChannel(java.nio.channels.spi.AbstractSelectableChannel.java:258)
  java.nio.channels.spi.AbstractInterruptibleChannel.close(java.nio.channels.spi.AbstractInterruptibleChannel.java:113)
  jdk.internal.net.http.PlainHttpConnection.close(jdk.internal.net.http.PlainHttpConnection.java:427)
  jdk.internal.net.http.PlainHttpConnection.close(jdk.internal.net.http.PlainHttpConnection.java:406)
  jdk.internal.net.http.Http1Response.lambda$readBody$1(jdk.internal.net.http.Http1Response.java:355)
  jdk.internal.net.http.Http1Response$$Lambda+0x00007f4cb5e6c438.749276779.accept(jdk.internal.net.http.Http1Response$$Lambda+0x00007f4cb5e6c438.749276779.java:-1)
  jdk.internal.net.http.ResponseContent$ChunkedBodyParser.onError(jdk.internal.net.http.ResponseContent$ChunkedBodyParser.java:185)
  jdk.internal.net.http.Http1Response$BodyReader.onReadError(jdk.internal.net.http.Http1Response$BodyReader.java:677)
  jdk.internal.net.http.Http1AsyncReceiver.checkForErrors(jdk.internal.net.http.Http1AsyncReceiver.java:302)
  jdk.internal.net.http.Http1AsyncReceiver.flush(jdk.internal.net.http.Http1AsyncReceiver.java:268)
  jdk.internal.net.http.Http1AsyncReceiver$$Lambda+0x00007f4cb5e31228.555093431.run(jdk.internal.net.http.Http1AsyncReceiver$$Lambda+0x00007f4cb5e31228.555093431.java:-1)
  jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.run(jdk.internal.net.http.common.SequentialScheduler$LockingRestartableTask.java:182)
  jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.run(jdk.internal.net.http.common.SequentialScheduler$CompleteRestartableTask.java:149)
  jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.run(jdk.internal.net.http.common.SequentialScheduler$SchedulableTask.java:207)
  jdk.internal.net.http.HttpClientImpl$DelegatingExecutor.execute(jdk.internal.net.http.HttpClientImpl$DelegatingExecutor.java:177)
  jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(jdk.internal.net.http.common.SequentialScheduler.java:282)
  jdk.internal.net.http.common.SequentialScheduler.runOrSchedule(jdk.internal.net.http.common.SequentialScheduler.java:251)
  jdk.internal.net.http.Http1AsyncReceiver.onReadError(jdk.internal.net.http.Http1AsyncReceiver.java:516)
  jdk.internal.net.http.Http1AsyncReceiver.lambda$handlePendingDelegate$3(jdk.internal.net.http.Http1AsyncReceiver.java:380)
  jdk.internal.net.http.Http1AsyncReceiver$$Lambda+0x00007f4cb5e33ca0.84679411.run(jdk.internal.net.http.Http1AsyncReceiver$$Lambda+0x00007f4cb5e33ca0.84679411.java:-1)
  jdk.internal.net.http.Http1AsyncReceiver$Http1AsyncDelegateSubscription.cancel(jdk.internal.net.http.Http1AsyncReceiver$Http1AsyncDelegateSubscription.java:163)
  jdk.internal.net.http.common.HttpBodySubscriberWrapper$SubscriptionWrapper.cancel(jdk.internal.net.http.common.HttpBodySubscriberWrapper$SubscriptionWrapper.java:92)
  jdk.internal.net.http.ResponseSubscribers$HttpResponseInputStream.close(jdk.internal.net.http.ResponseSubscribers$HttpResponseInputStream.java:653)

  com.acme.service.server.StatusClient.getResponse(com.acme.service.server.StatusClient.java:52)
  com.acme.service.server.StatusClient_ClientProxy.getResponse(com.acme.service.server.StatusClient_ClientProxy.java:-1)
  com.acme.client.Request.execute(com.acme.client.Request.java:96)
  com.acme.service.server.serviceStatusProvider.getStatusHistorys(com.acme.service.server.serviceStatusProvider.java:237)
  com.acme.service.api.RemoteStatusCheck.getStatusHistory(com.acme.service.api.RemoteStatusCheck.java:163)
  com.acme.service.api.RemoteStatusCheck.lambda$doChecks$0(com.acme.service.api.RemoteStatusCheck.java:132)
  com.acme.service.api.RemoteStatusCheck$$Lambda+0x00007f4cb9f0d8d0.979953307.call(com.acme.service.api.RemoteStatusCheck$$Lambda+0x00007f4cb9f0d8d0.979953307.java:-1)
  java.util.concurrent.FutureTask.run(java.util.concurrent.FutureTask.java:317)
  java.lang.VirtualThread.runWith(java.lang.VirtualThread.java:341)
  java.lang.VirtualThread.run(java.lang.VirtualThread.java:311)
  java.lang.VirtualThread$VThreadContinuation$1.run(java.lang.VirtualThread$VThreadContinuation$1.java:192)
  jdk.internal.vm.Continuation.enter0(jdk.internal.vm.Continuation.java:320)
  jdk.internal.vm.Continuation.enter(jdk.internal.vm.Continuation.java:312)
  jdk.internal.vm.Continuation.enterSpecial(jdk.internal.vm.Continuation.java:-1)

Answer 1

Метод(строки 108–115 в Temurin-21+35 (сборка 21+35-LTS), но, вероятно, во всех производных OpenJDK) реализуется как:

      public final void close() throws IOException {
    synchronized (closeLock) {
        if (closed)
            return;
        closed = true;
        implCloseChannel();
    }
}

Строка 113 в трассировке стека соответствуетвызов, который также соответствует предыдущей строке в вашей трассировке стека, и он находится в середине этого синхронизированного блока. Виртуальные потоки будут закреплены, если они припаркованы/заблокированы блоками, поэтому они и закрепляются.

Другими словами, учитывая код в его нынешнем виде, закрепление является ожидаемым и правильным поведением и, следовательно, не является ошибкой.

Является ли использование здесь недосмотром при избавлении от синхронизированных блоков в JDK, или есть конкретная причина, по которой это все еще используется., Я не знаю. Учитывая, что это частный объект блокировки, я думаю, от него можно избавиться (т. е. он не является частью «API» канала), заменив его наили что-то подобное, но, возможно, есть другие причины, по которым стоит оставить это на данный момент.

Я спросил об этом в списке nio-dev в теме. Должен ли AbstractInterruptibleChannel.close() по-прежнему использовать синхронизированный блок?

Алан Бейтман ответил там :

Мы решили, что этого делать не стоит, потому что SO_LINGER устанавливается редко. Временное закрепление из-за конфликта в readLock или writeLock при закрытии допустимо.

Тем временем мы работаем над снятием ограничения на синхронизируемые блоки. Мы надеемся, что скоро что-нибудь появится в репозитории ткацкого станка.