diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java index 4128efff44..301ccd27c1 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java @@ -37,6 +37,7 @@ import java.util.List; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.TimeUnit; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; @@ -146,12 +147,12 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED } private volatile boolean peerNoRfc7540Priorities; - private static final long STREAM_TIMEOUT_GRANULARITY_MILLIS = 1000; - private long lastStreamTimeoutCheckMillis; + private static final long STREAM_TIMEOUT_GRANULARITY_NANOS = TimeUnit.SECONDS.toNanos(1); + private long lastStreamTimeoutCheckNanos; - private static final long VALIDATE_AFTER_INACTIVITY_GRANULARITY_MILLIS = 1000; + private static final long VALIDATE_AFTER_INACTIVITY_GRANULARITY_NANOS = TimeUnit.SECONDS.toNanos(1); private final Timeout validateAfterInactivity; - private volatile long lastActivityTime; + private volatile long lastActivityNanos; AbstractH2StreamMultiplexer( final ProtocolIOSession ioSession, @@ -199,7 +200,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED } this.lowMark = H2Config.INIT.getInitialWindowSize() / 2; this.streamListener = streamListener; - this.lastActivityTime = System.currentTimeMillis(); + this.lastActivityNanos = System.nanoTime(); this.validateAfterInactivity = validateAfterInactivity; } @@ -539,8 +540,8 @@ public final void onOutput() throws HttpException, IOException { if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0 && remoteSettingState == SettingsHandshake.ACKED) { final long t = TimeValue.isPositive(validateAfterInactivity) ? - Math.max(validateAfterInactivity.toMilliseconds(), VALIDATE_AFTER_INACTIVITY_GRANULARITY_MILLIS) : 0; - final boolean hasBeenIdleTooLong = t > 0 && System.currentTimeMillis() - lastActivityTime > t; + Math.max(validateAfterInactivity.toNanoseconds(), VALIDATE_AFTER_INACTIVITY_GRANULARITY_NANOS) : 0; + final boolean hasBeenIdleTooLong = t > 0 && System.nanoTime() - lastActivityNanos > t; if (hasBeenIdleTooLong && ioSession.hasCommands() && pingHandlers.isEmpty()) { final Timeout socketTimeout = ioSession.getSocketTimeout(); ioSession.setSocketTimeout(Timeout.ofSeconds(5)); @@ -1510,7 +1511,7 @@ class H2StreamChannelImpl implements H2StreamChannel { private final AtomicInteger outputWindow; private volatile boolean localClosed; - private volatile long localResetTime; + private volatile long localResetNanos = Long.MIN_VALUE; H2StreamChannelImpl(final int id, final int initialInputWindowSize, final int initialOutputWindowSize) { this.id = id; @@ -1676,7 +1677,7 @@ public boolean localReset(final int code) throws IOException { return false; } localClosed = true; - localResetTime = System.currentTimeMillis(); + localResetNanos = System.nanoTime(); final RawFrame resetStream = frameFactory.createResetStream(id, code); commitFrameInternal(resetStream); @@ -1687,8 +1688,8 @@ public boolean localReset(final int code) throws IOException { } @Override - public long getLocalResetTime() { - return localResetTime; + public long getLocalResetNanos() { + return localResetNanos; } @Override @@ -1743,14 +1744,14 @@ private void checkStreamTimeouts(final long nowNanos) throws IOException { } private void validateStreamTimeouts() throws IOException { - final long nowMillis = System.currentTimeMillis(); - if ((nowMillis - lastStreamTimeoutCheckMillis) >= STREAM_TIMEOUT_GRANULARITY_MILLIS) { - lastStreamTimeoutCheckMillis = nowMillis; - checkStreamTimeouts(System.nanoTime()); + final long nowNanos = System.nanoTime(); + if ((nowNanos - lastStreamTimeoutCheckNanos) >= STREAM_TIMEOUT_GRANULARITY_NANOS) { + lastStreamTimeoutCheckNanos = nowNanos; + checkStreamTimeouts(nowNanos); } } private void updateLastActivity() { - this.lastActivityTime = System.currentTimeMillis(); + this.lastActivityNanos = System.nanoTime(); } } \ No newline at end of file diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java index 404360dcea..7bd10613d1 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java @@ -32,6 +32,7 @@ import java.nio.channels.CancelledKeyException; import java.nio.charset.CharacterCodingException; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -50,7 +51,7 @@ class H2Stream implements StreamControl { - private static final long LINGER_TIME = 1000; // 1 second + private static final long LINGER_TIME_NANOS = TimeUnit.SECONDS.toNanos(1); private final H2StreamChannel channel; private final H2StreamHandler handler; @@ -124,8 +125,8 @@ AtomicInteger getInputWindow() { } private boolean isPastLingerDeadline() { - final long localResetTime = channel.getLocalResetTime(); - return localResetTime > 0 && localResetTime + LINGER_TIME < System.currentTimeMillis(); + final long localResetNanos = channel.getLocalResetNanos(); + return localResetNanos != Long.MIN_VALUE && System.nanoTime() - localResetNanos > LINGER_TIME_NANOS; } boolean isClosedPastLingerDeadline() { diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2StreamChannel.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2StreamChannel.java index d86159ef9a..ed99bcf4f9 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2StreamChannel.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2StreamChannel.java @@ -68,10 +68,10 @@ default void terminate() { } } - long getLocalResetTime(); + long getLocalResetNanos(); default boolean isLocalReset() { - return getLocalResetTime() > 0; + return getLocalResetNanos() != Long.MIN_VALUE; } } diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java index 6ea7414e70..afcfffd3d2 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java @@ -143,8 +143,8 @@ protected void validateSession( final TimeValue timeValue = validateAfterInactivity; if (TimeValue.isNonNegative(timeValue)) { final long lastAccessTime = Math.min(ioSession.getLastReadTime(), ioSession.getLastWriteTime()); - final long deadline = lastAccessTime + timeValue.toMilliseconds(); - if (deadline <= System.currentTimeMillis()) { + final long deadline = lastAccessTime + timeValue.toNanoseconds(); + if (deadline <= System.nanoTime()) { ioSession.enqueue(new StaleCheckCommand(callback::execute), Command.Priority.NORMAL); return; } diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java index 8a17b8124e..1e3b43456a 100644 --- a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java +++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java @@ -1706,23 +1706,23 @@ void testKeepAliveDisabledNeverEmitsPing() throws Exception { Assertions.assertTrue(frames.stream().noneMatch(FrameStub::isPing), "Disabled policy must never emit PING"); } - private static long getValidateAfterInactivityGranularityMillis() throws Exception { - final Field field = AbstractH2StreamMultiplexer.class.getDeclaredField("VALIDATE_AFTER_INACTIVITY_GRANULARITY_MILLIS"); + private static long getValidateAfterInactivityGranularityNanos() throws Exception { + final Field field = AbstractH2StreamMultiplexer.class.getDeclaredField("VALIDATE_AFTER_INACTIVITY_GRANULARITY_NANOS"); field.setAccessible(true); return field.getLong(null); } - private static void setLastActivityTime(final AbstractH2StreamMultiplexer mux, final long millis) throws Exception { - final Field field = AbstractH2StreamMultiplexer.class.getDeclaredField("lastActivityTime"); + private static void setLastActivityNanos(final AbstractH2StreamMultiplexer mux, final long nanos) throws Exception { + final Field field = AbstractH2StreamMultiplexer.class.getDeclaredField("lastActivityNanos"); field.setAccessible(true); - field.setLong(mux, millis); + field.setLong(mux, nanos); } private static void makeMuxIdle(final AbstractH2StreamMultiplexer mux, final Timeout validateAfterInactivity) throws Exception { - final long granularityMillis = getValidateAfterInactivityGranularityMillis(); - final long configuredMillis = validateAfterInactivity != null ? validateAfterInactivity.toMilliseconds() : 0; - final long effectiveMillis = configuredMillis > 0 ? Math.max(configuredMillis, granularityMillis) : 0; - setLastActivityTime(mux, System.currentTimeMillis() - effectiveMillis - 10); + final long granularityNanos = getValidateAfterInactivityGranularityNanos(); + final long configuredNanos = validateAfterInactivity != null ? validateAfterInactivity.toNanoseconds() : 0; + final long effectiveNanos = configuredNanos > 0 ? Math.max(configuredNanos, granularityNanos) : 0; + setLastActivityNanos(mux, System.nanoTime() - effectiveNanos - TimeUnit.MILLISECONDS.toNanos(10)); } @Test diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/nio/pool/TestH2ConnPool.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/nio/pool/TestH2ConnPool.java index b9bf2558c4..83651e6035 100644 --- a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/nio/pool/TestH2ConnPool.java +++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/nio/pool/TestH2ConnPool.java @@ -28,6 +28,7 @@ import java.net.InetSocketAddress; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.apache.hc.core5.concurrent.FutureCallback; @@ -97,8 +98,9 @@ void testValidateSessionEnqueuesStaleCheck() { final IOSession session = Mockito.mock(IOSession.class); Mockito.when(session.isOpen()).thenReturn(true); - Mockito.when(session.getLastReadTime()).thenReturn(0L); - Mockito.when(session.getLastWriteTime()).thenReturn(0L); + final long farPastNanos = System.nanoTime() - TimeUnit.DAYS.toNanos(1); + Mockito.when(session.getLastReadTime()).thenReturn(farPastNanos); + Mockito.when(session.getLastWriteTime()).thenReturn(farPastNanos); @SuppressWarnings("unchecked") final Callback callback = (Callback) Mockito.mock(Callback.class); diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java index f8936a4bd1..fd110531af 100644 --- a/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java +++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/benchmark/HttpBenchmark.java @@ -473,7 +473,7 @@ private Results doExecute(final HttpAsyncRequester requester, final Stats stats) final long deadline = config.getTimeLimit() != null ? config.getTimeLimit().toMilliseconds() : Long.MAX_VALUE; - final long startTime = System.currentTimeMillis(); + final long startNanos = System.nanoTime(); for (int i = 0; i < workers.length; i++) { workers[i].execute(); @@ -485,7 +485,7 @@ private Results doExecute(final HttpAsyncRequester requester, final Stats stats) System.out.println("...done"); } - final long endTime = System.currentTimeMillis(); + final long endNanos = System.nanoTime(); for (int i = 0; i < workers.length; i++) { workers[i].releaseResources(); @@ -499,7 +499,7 @@ private Results doExecute(final HttpAsyncRequester requester, final Stats stats) requestUri.toASCIIString(), stats.getContentLength(), config.getConcurrencyLevel(), - endTime - startTime, + TimeUnit.NANOSECONDS.toMillis(endNanos - startNanos), stats.getSuccessCount(), stats.getFailureCount(), stats.getKeepAliveCount(), diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/SocksProxy.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/SocksProxy.java index dc7629ef8a..2999ad080c 100644 --- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/SocksProxy.java +++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/SocksProxy.java @@ -37,6 +37,7 @@ import java.net.SocketAddress; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; import org.apache.hc.core5.net.InetAddressUtils; @@ -254,7 +255,7 @@ public void start() throws IOException { } public void shutdown(final TimeValue timeout) throws InterruptedException { - final long waitUntil = System.currentTimeMillis() + timeout.toMilliseconds(); + final long deadlineNanos = System.nanoTime() + timeout.toNanoseconds(); Thread t = null; lock.lock(); try { @@ -272,18 +273,18 @@ public void shutdown(final TimeValue timeout) throws InterruptedException { handler.shutdown(); } while (!this.handlers.isEmpty()) { - final long waitTime = waitUntil - System.currentTimeMillis(); - if (waitTime > 0) { - wait(waitTime); + final long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos > 0) { + wait(TimeUnit.NANOSECONDS.toMillis(remainingNanos)); } } } finally { lock.unlock(); } if (t != null) { - final long waitTime = waitUntil - System.currentTimeMillis(); - if (waitTime > 0) { - t.join(waitTime); + final long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos > 0) { + t.join(TimeUnit.NANOSECONDS.toMillis(remainingNanos)); } } } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/concurrent/BasicFuture.java b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/BasicFuture.java index c1e0ea93a2..6fddfa6a91 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/concurrent/BasicFuture.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/BasicFuture.java @@ -110,23 +110,26 @@ public T get(final long timeout, final TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { Args.notNull(unit, "Time unit"); final long msecs = unit.toMillis(timeout); - final long startTime = (msecs <= 0) ? 0 : System.currentTimeMillis(); - long waitTime = msecs; + final long waitNanos = unit.toNanos(timeout); + final long startNanos = (waitNanos <= 0) ? 0 : System.nanoTime(); + long remainingNanos = waitNanos; try { lock.lock(); if (this.completed) { return getResult(); - } else if (waitTime <= 0) { - throw TimeoutValueException.fromMilliseconds(msecs, msecs + Math.abs(waitTime)); + } else if (remainingNanos <= 0) { + throw TimeoutValueException.fromMilliseconds(msecs, msecs); } else { for (; ; ) { - condition.await(waitTime, TimeUnit.MILLISECONDS); + condition.await(remainingNanos, TimeUnit.NANOSECONDS); if (this.completed) { return getResult(); } - waitTime = msecs - (System.currentTimeMillis() - startTime); - if (waitTime <= 0) { - throw TimeoutValueException.fromMilliseconds(msecs, msecs + Math.abs(waitTime)); + remainingNanos = waitNanos - (System.nanoTime() - startNanos); + if (remainingNanos <= 0) { + final long elapsedMillis = TimeUnit.NANOSECONDS.toMillis( + System.nanoTime() - startNanos); + throw TimeoutValueException.fromMilliseconds(msecs, elapsedMillis); } } } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java index a7c7cffa0a..025739e33a 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java @@ -267,7 +267,7 @@ public final void enumAvailable(final Callback callback) { } public final void closeIdle(final TimeValue idleTime) { - final long deadline = System.currentTimeMillis() - (TimeValue.isPositive(idleTime) ? idleTime.toMilliseconds() : 0); + final long deadline = System.nanoTime() - (TimeValue.isPositive(idleTime) ? idleTime.toNanoseconds() : 0); for (final PoolEntry poolEntry: sessionPool.values()) { if (poolEntry.session != null) { lock.lock(); diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractSingleCoreIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractSingleCoreIOReactor.java index fc3de36f1f..2d59f07e93 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractSingleCoreIOReactor.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractSingleCoreIOReactor.java @@ -109,14 +109,14 @@ public void execute() { @Override public final void awaitShutdown(final TimeValue waitTime) throws InterruptedException { Args.notNull(waitTime, "Wait time"); - final long deadline = System.currentTimeMillis() + waitTime.toMilliseconds(); - long remaining = waitTime.toMilliseconds(); + final long deadlineNanos = System.nanoTime() + waitTime.toNanoseconds(); + long remainingNanos = waitTime.toNanoseconds(); lock.lock(); try { while (this.status.get().compareTo(IOReactorStatus.SHUT_DOWN) < 0) { - condition.await(remaining, TimeUnit.MILLISECONDS); - remaining = deadline - System.currentTimeMillis(); - if (remaining <= 0) { + condition.await(remainingNanos, TimeUnit.NANOSECONDS); + remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { return; } } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java index 73cbfc20da..eaa318c5aa 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java @@ -206,25 +206,26 @@ enum Status { void setSocketTimeout(Timeout timeout); /** - * Returns timestamp of the last read event. + * Returns monotonic nanosecond timestamp of the last read event. * - * @return timestamp. + * @return nanosecond timestamp obtained from {@link System#nanoTime()}. */ long getLastReadTime(); /** - * Returns timestamp of the last write event. + * Returns monotonic nanosecond timestamp of the last write event. * - * @return timestamp. + * @return nanosecond timestamp obtained from {@link System#nanoTime()}. */ long getLastWriteTime(); /** - * Returns timestamp of the last I/O event including socket timeout reset. + * Returns monotonic nanosecond timestamp of the last I/O event including + * socket timeout reset. * * @see #getSocketTimeout() * - * @return timestamp. + * @return nanosecond timestamp obtained from {@link System#nanoTime()}. */ long getLastEventTime(); diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java index 042b593c01..2cedd7ab36 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java @@ -78,10 +78,10 @@ public IOSessionImpl(final String type, final SelectionKey key, final SocketChan this.id = String.format(type + "-%010d", COUNT.getAndIncrement()); this.handlerRef = new AtomicReference<>(); this.status = new AtomicReference<>(Status.ACTIVE); - final long currentTimeMillis = System.currentTimeMillis(); - this.lastReadTime = currentTimeMillis; - this.lastWriteTime = currentTimeMillis; - this.lastEventTime = currentTimeMillis; + final long nowNanos = System.nanoTime(); + this.lastReadTime = nowNanos; + this.lastWriteTime = nowNanos; + this.lastEventTime = nowNanos; } @Override @@ -211,7 +211,7 @@ public Timeout getSocketTimeout() { @Override public void setSocketTimeout(final Timeout timeout) { this.socketTimeout = Timeout.defaultsToInfinite(timeout); - this.lastEventTime = System.currentTimeMillis(); + this.lastEventTime = System.nanoTime(); } @Override @@ -226,13 +226,13 @@ public int write(final ByteBuffer src) throws IOException { @Override public void updateReadTime() { - lastReadTime = System.currentTimeMillis(); + lastReadTime = System.nanoTime(); lastEventTime = lastReadTime; } @Override public void updateWriteTime() { - lastWriteTime = System.currentTimeMillis(); + lastWriteTime = System.nanoTime(); lastEventTime = lastWriteTime; } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java index 57f727194f..ae25e10a6b 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionRequest.java @@ -73,7 +73,7 @@ public IOSessionRequest( this.closeableRef = new AtomicReference<>(); // Set the time when this request is created - this.enqueueTime = System.currentTimeMillis(); + this.enqueueTime = System.nanoTime(); } public void completed(final ProtocolIOSession ioSession) { @@ -136,7 +136,7 @@ public String toString() { // Getter for enqueueTime @Internal - public long getEnqueueTime() { + public long getEnqueueNanos() { return enqueueTime; } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOWorkerStats.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOWorkerStats.java index 75ec5fea72..f17e75d0ad 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOWorkerStats.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOWorkerStats.java @@ -45,6 +45,6 @@ public interface IOWorkerStats { int pendingChannelCount(); // Cheap - long lastSelectMilli(); + long lastSelectNano(); } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalChannel.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalChannel.java index acf8c8afc0..ff18166a70 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalChannel.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalChannel.java @@ -57,12 +57,12 @@ final void handleIOEvent(final int ops) { } } - final boolean checkTimeout(final long currentTimeMillis) { + final boolean checkTimeout(final long nowNanos) { final Timeout timeout = getTimeout(); if (!timeout.isDisabled()) { - final long timeoutMillis = timeout.toMilliseconds(); - final long deadlineMillis = getLastEventTime() + timeoutMillis; - if (currentTimeMillis > deadlineMillis) { + final long timeoutNanos = timeout.toNanoseconds(); + final long deadlineNanos = getLastEventTime() + timeoutNanos; + if (nowNanos > deadlineNanos) { try { onTimeout(timeout); } catch (final CancelledKeyException ex) { diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalConnectChannel.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalConnectChannel.java index a69ca198a9..e90161611d 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalConnectChannel.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalConnectChannel.java @@ -44,7 +44,7 @@ final class InternalConnectChannel extends InternalChannel { private final InternalDataChannel dataChannel; private final IOEventHandlerFactory eventHandlerFactory; private final IOReactorConfig reactorConfig; - private final long creationTimeMillis; + private final long creationNanos; InternalConnectChannel( final SelectionKey key, @@ -60,7 +60,7 @@ final class InternalConnectChannel extends InternalChannel { this.dataChannel = dataChannel; this.eventHandlerFactory = eventHandlerFactory; this.reactorConfig = reactorConfig; - this.creationTimeMillis = System.currentTimeMillis(); + this.creationNanos = System.nanoTime(); } @Override @@ -70,8 +70,8 @@ void onIOEvent(final int readyOps) throws IOException { socketChannel.finishConnect(); } //check out connectTimeout - final long now = System.currentTimeMillis(); - if (checkTimeout(now)) { + final long nowNanos = System.nanoTime(); + if (checkTimeout(nowNanos)) { if (reactorConfig.getSocksProxyAddress() == null) { dataChannel.upgrade(eventHandlerFactory.createHandler(dataChannel, sessionRequest.attachment)); key.attach(dataChannel); @@ -95,7 +95,7 @@ Timeout getTimeout() { @Override long getLastEventTime() { - return creationTimeMillis; + return creationNanos; } @Override diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/MultiCoreIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/MultiCoreIOReactor.java index 957a772eb4..3799c9d7ff 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/MultiCoreIOReactor.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/MultiCoreIOReactor.java @@ -88,23 +88,25 @@ public final void initiateShutdown() { @Override public final void awaitShutdown(final TimeValue waitTime) throws InterruptedException { Args.notNull(waitTime, "Wait time"); - final long deadline = System.currentTimeMillis() + waitTime.toMilliseconds(); - long remaining = waitTime.toMilliseconds(); + final long deadlineNanos = System.nanoTime() + waitTime.toNanoseconds(); + long remainingNanos = waitTime.toNanoseconds(); for (int i = 0; i < this.ioReactors.length; i++) { final IOReactor ioReactor = this.ioReactors[i]; if (ioReactor.getStatus().compareTo(IOReactorStatus.SHUT_DOWN) < 0) { - ioReactor.awaitShutdown(TimeValue.of(remaining, TimeUnit.MILLISECONDS)); - remaining = deadline - System.currentTimeMillis(); - if (remaining <= 0) { + ioReactor.awaitShutdown(TimeValue.of(remainingNanos, TimeUnit.NANOSECONDS)); + remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { return; } } } for (int i = 0; i < this.threads.length; i++) { final Thread thread = this.threads[i]; - thread.join(remaining); - remaining = deadline - System.currentTimeMillis(); - if (remaining <= 0) { + final long millis = TimeUnit.NANOSECONDS.toMillis(remainingNanos); + final int nanos = (int) (remainingNanos - TimeUnit.MILLISECONDS.toNanos(millis)); + thread.join(millis, nanos); + remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { return; } } diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java index dbda722e33..cd2b278a67 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java @@ -42,6 +42,7 @@ import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -72,8 +73,9 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect private final Queue requestQueue; private final AtomicBoolean shutdownInitiated; private final long selectTimeoutMillis; - private volatile long lastTimeoutCheckMillis; - private volatile long lastSelectMillis; + private final long selectTimeoutNanos; + private volatile long lastTimeoutCheckNanos; + private volatile long lastSelectNanos; private final IOReactorMetricsListener threadPoolListener; // Atomic variables for tracking total wait time and count of processed requests @@ -100,6 +102,7 @@ class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements Connect this.channelQueue = new ConcurrentLinkedQueue<>(); this.requestQueue = new ConcurrentLinkedQueue<>(); this.selectTimeoutMillis = this.reactorConfig.getSelectInterval().toMilliseconds(); + this.selectTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(this.selectTimeoutMillis); } void enqueueChannel(final ChannelEntry entry) throws IOReactorShutdownException { @@ -135,7 +138,7 @@ void doExecute() throws IOException { } // Process selected I/O events - lastSelectMillis = System.currentTimeMillis(); + lastSelectNanos = System.nanoTime(); if (readyCount > 0) { processEvents(this.selector.selectedKeys()); } @@ -176,11 +179,11 @@ private void initiateSessionShutdown() { } private void validateActiveChannels() { - final long currentTimeMillis = System.currentTimeMillis(); - if ((currentTimeMillis - this.lastTimeoutCheckMillis) >= this.selectTimeoutMillis) { - this.lastTimeoutCheckMillis = currentTimeMillis; + final long nowNanos = System.nanoTime(); + if ((nowNanos - this.lastTimeoutCheckNanos) >= this.selectTimeoutNanos) { + this.lastTimeoutCheckNanos = nowNanos; for (final SelectionKey key : this.selector.keys()) { - checkTimeout(key, currentTimeMillis); + checkTimeout(key, nowNanos); } } } @@ -255,10 +258,10 @@ private void processClosedSessions() { } } - private void checkTimeout(final SelectionKey key, final long nowMillis) { + private void checkTimeout(final SelectionKey key, final long nowNanos) { final InternalChannel channel = (InternalChannel) key.attachment(); if (channel != null) { - channel.checkTimeout(nowMillis); + channel.checkTimeout(nowNanos); } } @@ -336,7 +339,7 @@ private void processPendingConnectionRequests() { for (int i = 0; i < MAX_CHANNEL_REQUESTS && (sessionRequest = this.requestQueue.poll()) != null; i++) { if (threadPoolListener != null) { // Calculate wait time safely without keeping long-lived state - final long waitTimeMillis = System.currentTimeMillis() - sessionRequest.getEnqueueTime(); + final long waitTimeMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sessionRequest.getEnqueueNanos()); // Accumulate total wait time and increment count atomically totalWaitTime.addAndGet(waitTimeMillis); @@ -505,8 +508,8 @@ public int pendingChannelCount() { } @Override - public long lastSelectMilli() { - return lastSelectMillis; + public long lastSelectNano() { + return lastSelectNanos; } } diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestInternalChannel.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestInternalChannel.java index cfd2356098..f747d46c2e 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestInternalChannel.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestInternalChannel.java @@ -28,6 +28,7 @@ import java.io.IOException; import java.nio.channels.CancelledKeyException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -118,10 +119,10 @@ void handleIoEventClosesOnException() { @Test void checkTimeoutInvokesCallback() { try (TestChannel channel = new TestChannel()) { - channel.lastEventTime = 0; + channel.lastEventTime = System.nanoTime() - TimeUnit.SECONDS.toNanos(10); channel.timeout = Timeout.ofMilliseconds(1); - final boolean result = channel.checkTimeout(10); + final boolean result = channel.checkTimeout(System.nanoTime()); Assertions.assertFalse(result); Assertions.assertTrue(channel.timedOut.get()); @@ -133,7 +134,7 @@ void checkTimeoutSkipsWhenDisabled() { try (TestChannel channel = new TestChannel()) { channel.timeout = Timeout.DISABLED; - final boolean result = channel.checkTimeout(System.currentTimeMillis()); + final boolean result = channel.checkTimeout(System.nanoTime()); Assertions.assertTrue(result); Assertions.assertFalse(channel.timedOut.get()); diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocksProxyProtocolHandler.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocksProxyProtocolHandler.java index 9b6b1a1821..64e06823de 100644 --- a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocksProxyProtocolHandler.java +++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSocksProxyProtocolHandler.java @@ -136,7 +136,7 @@ private static final class TestIOSession implements IOSession { this.open = true; this.eventMask = 0; this.socketTimeout = Timeout.DISABLED; - this.lastReadTime = System.currentTimeMillis(); + this.lastReadTime = System.nanoTime(); this.lastWriteTime = this.lastReadTime; this.lastEventTime = this.lastReadTime; } @@ -153,7 +153,7 @@ public ByteChannel channel() { @Override public void setEventMask(final int ops) { this.eventMask = ops; - this.lastEventTime = System.currentTimeMillis(); + this.lastEventTime = System.nanoTime(); } @Override @@ -220,7 +220,7 @@ public Timeout getSocketTimeout() { @Override public void setSocketTimeout(final Timeout timeout) { this.socketTimeout = timeout; - this.lastEventTime = System.currentTimeMillis(); + this.lastEventTime = System.nanoTime(); } @Override @@ -240,13 +240,13 @@ public long getLastEventTime() { @Override public void updateReadTime() { - this.lastReadTime = System.currentTimeMillis(); + this.lastReadTime = System.nanoTime(); this.lastEventTime = this.lastReadTime; } @Override public void updateWriteTime() { - this.lastWriteTime = System.currentTimeMillis(); + this.lastWriteTime = System.nanoTime(); this.lastEventTime = this.lastWriteTime; }