You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "vldpyatkov (via GitHub)" <gi...@apache.org> on 2024/08/21 21:29:27 UTC
[PR] IGNITE-20869 CompletableFuture with orTimeout has noticeable performa… [ignite-3]
vldpyatkov opened a new pull request, #4261:
URL: https://github.com/apache/ignite-3/pull/4261
…nce impact
https://issues.apache.org/jira/browse/IGNITE-20869
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731044590
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
Review Comment:
Because of that comment above,. the protocol has to get a response in any case. It is a main difference between timeout worker for Ignite node and the client. That is also a reason why I do not use one class in both cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1728971976
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -104,6 +114,9 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
/** Executor for async operation listeners. */
private final Executor asyncContinuationExecutor;
+ /** Timeout worker. */
+ private TimeoutWorker timeoutWorker;
Review Comment:
```suggestion
private final TimeoutWorker timeoutWorker;
```
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
+ }
+ }
+
+ try {
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException e) {
+ log.info("The timeout worker was interrupted, probably the client is stopping.");
+ }
+
+ updateHeartbeat();
+ }
+
+ } catch (Throwable t) {
+ // Call FH here.
+
+ throw new IgniteInternalException(t);
+ }
+ }
+ }
+
+ /**
+ * Timeout object.
+ * The class is a wrapper over the complete future.
+ */
+ private static class TimeoutObject {
+ /** End time. */
Review Comment:
```suggestion
/** End time (milliseconds since Unix epoch). */
```
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
Review Comment:
Why is it declared as `ConcurrentMap` and not just `Map`?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/FeatureTimeoutBenchmark.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMillis;
+
+import com.lmax.disruptor.dsl.Disruptor;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Feature timeout benchmark.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FeatureTimeoutBenchmark {
Review Comment:
What does 'feature' mean here? Is it just 'future'?
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -87,7 +97,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
private final AtomicLong reqId = new AtomicLong(1);
/** Pending requests. */
- private final Map<Long, ClientRequestFuture<?>> pendingReqs = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, TimeoutObject> pendingReqs = new ConcurrentHashMap<>();
Review Comment:
Do we really need this change from `Map`?
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
Review Comment:
Why is this line commented out?
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -287,10 +302,9 @@ private CompletableFuture<NetworkMessage> invoke0(ClusterNode recipient, Channel
long correlationId = createCorrelationId();
- CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<NetworkMessage>()
- .orTimeout(timeout, TimeUnit.MILLISECONDS);
+ CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<>();
- requestsMap.put(correlationId, responseFuture);
+ requestsMap.put(correlationId, new TimeoutObject(timeout > 0 ? coarseCurrentTimeMillis() + timeout : 0, responseFuture));
Review Comment:
It seems that `orTimeout(0, SECONDS)` will timeout immediately, but the `TimeoutObject` treats 0 as 'no timeout at all', so this changes the behavior. I think we need to establish what we want here first.
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -764,4 +782,116 @@ private InetSocketAddress getFromCacheOrCreateResolved(ClusterNode recipientNode
return address != null ? address : createResolved(recipientNode.address());
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
Review Comment:
```suggestion
/** Active operations. */
```
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
+ }
+ }
+
+ try {
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException e) {
+ log.info("The timeout worker was interrupted, probably the client is stopping.");
+ }
+
+ updateHeartbeat();
+ }
+
+ } catch (Throwable t) {
+ // Call FH here.
Review Comment:
If we are going to call a FH, let's add a TODO for this
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
+ }
+ }
+
+ try {
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException e) {
+ log.info("The timeout worker was interrupted, probably the client is stopping.");
+ }
+
+ updateHeartbeat();
+ }
+
+ } catch (Throwable t) {
+ // Call FH here.
+
+ throw new IgniteInternalException(t);
+ }
+ }
+ }
+
+ /**
+ * Timeout object.
+ * The class is a wrapper over the complete future.
+ */
+ private static class TimeoutObject {
+ /** End time. */
+ private final long endTime;
+
+ /** Target future. */
+ private final ClientRequestFuture<?> fut;
+
+ /**
+ * Constructor.
+ *
+ * @param endTime End timestamp in milliseconds.
+ * @param fut Target future.
+ */
+ public TimeoutObject(long endTime, ClientRequestFuture<?> fut) {
+ this.endTime = endTime;
+ this.fut = fut;
+ }
+
+ /**
+ * Gets end timestamp.
+ *
+ * @return End timestamp in milliseconds.
+ */
+ public long getEndTime() {
+ return endTime;
+ }
+
+ /**
+ * Gets a target future.
+ *
+ * @return A future.
+ */
+ public ClientRequestFuture<?> getFuture() {
Review Comment:
```suggestion
public ClientRequestFuture<?> future() {
```
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
Review Comment:
Is this check really needed? If the future is already completed, its subsequent completion will have no effect, and it will take just one volatile read (which we pay for `isDone()`)
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
Review Comment:
`TimeoutWorker` classes seem almost identical and `TimeoutObject` classes only differ in the element type of the future. It seems that it's really easy to have just one generic `TimeoutObject`. As for `TimeoutWorker`, most its logic is identical, and differences can be extracted to an abstract method, to avoid code duplication.
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
+ }
+ }
+
+ try {
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException e) {
+ log.info("The timeout worker was interrupted, probably the client is stopping.");
+ }
+
+ updateHeartbeat();
+ }
+
+ } catch (Throwable t) {
+ // Call FH here.
+
+ throw new IgniteInternalException(t);
+ }
+ }
+ }
+
+ /**
+ * Timeout object.
+ * The class is a wrapper over the complete future.
+ */
+ private static class TimeoutObject {
+ /** End time. */
+ private final long endTime;
+
+ /** Target future. */
+ private final ClientRequestFuture<?> fut;
+
+ /**
+ * Constructor.
+ *
+ * @param endTime End timestamp in milliseconds.
+ * @param fut Target future.
+ */
+ public TimeoutObject(long endTime, ClientRequestFuture<?> fut) {
+ this.endTime = endTime;
+ this.fut = fut;
+ }
+
+ /**
+ * Gets end timestamp.
+ *
+ * @return End timestamp in milliseconds.
+ */
+ public long getEndTime() {
Review Comment:
```suggestion
public long endTime() {
```
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Does this mean a thread per client? Isn't this too much?
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
Why is a copy of the map created here? Isn't concurrent map safe with respect to iteration happening concurrently to modifications?
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
Review Comment:
The map is not sorted, so why is it a queue?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/FeatureTimeoutBenchmark.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMillis;
+
+import com.lmax.disruptor.dsl.Disruptor;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Feature timeout benchmark.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FeatureTimeoutBenchmark {
+ private static AtomicLong ID_GEN = new AtomicLong();
+
+ /** Active operation. */
+ public ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ private IgniteThread timeoutWorker;
+
+ private ConcurrentMap<Long, CompletableFuture<?>> futs;
+
+ private Disruptor<TimeoutObject> disruptor;
+
+ @Param({"false", "true"})
+ private boolean useFutureEmbeddedTimeout;
+
+ /**
+ * Prepare to start the benchmark.
+ */
+ @Setup
+ public void setUp() {
+ if (useFutureEmbeddedTimeout) {
+ futs = new ConcurrentHashMap<>();
+ } else {
+ requestsMap = new ConcurrentHashMap<>();
+ timeoutWorker = new IgniteThread("benchmark", "timeout-worker", new TimeoutRunnable(requestsMap));
+ timeoutWorker.start();
+ }
+ }
+
+ /**
+ * Closes resources.
+ */
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ if (useFutureEmbeddedTimeout) {
+ for (CompletableFuture<?> fut : futs.values()) {
+ if (!fut.isDone()) {
+ try {
+ fut.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ assert e.getCause() instanceof TimeoutException : "Unexpected exception type: " + e.getCause().getClass();
+ } catch (TimeoutException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ futs.clear();
+ futs = null;
+ } else {
+ assert waitForCondition(requestsMap::isEmpty, 10_000);
+
+ timeoutWorker.interrupt();
+ timeoutWorker = null;
+ }
+ }
+
+ /**
+ * Benchmark for KV upsert via embedded client.
+ */
+ @Benchmark
+ public void test() {
+ if (useFutureEmbeddedTimeout) {
+ for (int i = 0; i < 10; i++) {
+ var fut = new CompletableFuture<Void>();
+
+ futs.put(ID_GEN.incrementAndGet(), fut);
+
+ fut.orTimeout(10, TimeUnit.MILLISECONDS);
+
+ if (futs.size() > 100_000) {
Review Comment:
Why is this needed?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java:
##########
@@ -121,6 +120,10 @@ else if (err instanceof TimeoutException) // Translate timeout exception.
blockedMsgs.add(msgData);
+ if (timeoutMs > 0) {
+ fut.orTimeout(timeoutMs, TimeUnit.MILLISECONDS);
Review Comment:
Could you please elaborate why this is better? We still call `orTimeout()`, just later.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/FeatureTimeoutBenchmark.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMillis;
+
+import com.lmax.disruptor.dsl.Disruptor;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Feature timeout benchmark.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FeatureTimeoutBenchmark {
+ private static AtomicLong ID_GEN = new AtomicLong();
+
+ /** Active operation. */
+ public ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ private IgniteThread timeoutWorker;
+
+ private ConcurrentMap<Long, CompletableFuture<?>> futs;
+
+ private Disruptor<TimeoutObject> disruptor;
+
+ @Param({"false", "true"})
+ private boolean useFutureEmbeddedTimeout;
+
+ /**
+ * Prepare to start the benchmark.
+ */
+ @Setup
+ public void setUp() {
+ if (useFutureEmbeddedTimeout) {
+ futs = new ConcurrentHashMap<>();
+ } else {
+ requestsMap = new ConcurrentHashMap<>();
+ timeoutWorker = new IgniteThread("benchmark", "timeout-worker", new TimeoutRunnable(requestsMap));
+ timeoutWorker.start();
+ }
+ }
+
+ /**
+ * Closes resources.
+ */
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ if (useFutureEmbeddedTimeout) {
+ for (CompletableFuture<?> fut : futs.values()) {
+ if (!fut.isDone()) {
+ try {
+ fut.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ assert e.getCause() instanceof TimeoutException : "Unexpected exception type: " + e.getCause().getClass();
+ } catch (TimeoutException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ futs.clear();
+ futs = null;
+ } else {
+ assert waitForCondition(requestsMap::isEmpty, 10_000);
+
+ timeoutWorker.interrupt();
+ timeoutWorker = null;
+ }
+ }
+
+ /**
+ * Benchmark for KV upsert via embedded client.
+ */
+ @Benchmark
+ public void test() {
+ if (useFutureEmbeddedTimeout) {
+ for (int i = 0; i < 10; i++) {
+ var fut = new CompletableFuture<Void>();
+
+ futs.put(ID_GEN.incrementAndGet(), fut);
+
+ fut.orTimeout(10, TimeUnit.MILLISECONDS);
+
+ if (futs.size() > 100_000) {
+ futs = new ConcurrentHashMap<>();
+ }
+ }
+ } else {
+ for (int i = 0; i < 10; i++) {
+ requestsMap.put(ID_GEN.incrementAndGet(), new TimeoutObject(
+ System.currentTimeMillis() + 10,
+ new CompletableFuture()
+ ));
+
+ if (requestsMap.size() > 100_000) {
+ requestsMap = new ConcurrentHashMap<>();
+ }
+ }
+ }
+ }
+
+ /**
+ * Benchmark's entry point.
+ */
+ public static void main(String[] args) throws RunnerException {
+ Options opt = new OptionsBuilder()
+ .include(".*" + FeatureTimeoutBenchmark.class.getSimpleName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutRunnable implements Runnable {
+ /** Active operation. */
Review Comment:
```suggestion
/** Active operations. */
```
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java:
##########
@@ -87,23 +87,22 @@ public ClusterService clusterService() {
) {
CompletableFuture<Message> fut = new CompletableFuture<>();
- fut.orTimeout(timeoutMs, TimeUnit.MILLISECONDS).
- whenComplete((res, err) -> {
- assert !(res == null && err == null) : res + " " + err;
+ fut.whenComplete((res, err) -> {
Review Comment:
This change seems to change the semantics. Before the change, `whenComplete()` callback would see the `TimeoutException`, but now it won't.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/FeatureTimeoutBenchmark.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMillis;
+
+import com.lmax.disruptor.dsl.Disruptor;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Feature timeout benchmark.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FeatureTimeoutBenchmark {
+ private static AtomicLong ID_GEN = new AtomicLong();
+
+ /** Active operation. */
+ public ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ private IgniteThread timeoutWorker;
+
+ private ConcurrentMap<Long, CompletableFuture<?>> futs;
+
+ private Disruptor<TimeoutObject> disruptor;
+
+ @Param({"false", "true"})
+ private boolean useFutureEmbeddedTimeout;
+
+ /**
+ * Prepare to start the benchmark.
+ */
+ @Setup
+ public void setUp() {
+ if (useFutureEmbeddedTimeout) {
+ futs = new ConcurrentHashMap<>();
+ } else {
+ requestsMap = new ConcurrentHashMap<>();
+ timeoutWorker = new IgniteThread("benchmark", "timeout-worker", new TimeoutRunnable(requestsMap));
+ timeoutWorker.start();
+ }
+ }
+
+ /**
+ * Closes resources.
+ */
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ if (useFutureEmbeddedTimeout) {
+ for (CompletableFuture<?> fut : futs.values()) {
+ if (!fut.isDone()) {
+ try {
+ fut.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ assert e.getCause() instanceof TimeoutException : "Unexpected exception type: " + e.getCause().getClass();
+ } catch (TimeoutException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ futs.clear();
+ futs = null;
+ } else {
+ assert waitForCondition(requestsMap::isEmpty, 10_000);
+
+ timeoutWorker.interrupt();
+ timeoutWorker = null;
+ }
+ }
+
+ /**
+ * Benchmark for KV upsert via embedded client.
Review Comment:
It's not about KV at all :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731429663
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
It is not clear where this thread has to be stopped.
A static variable does not seem like a good solution to me.
I believe it is better to do it on a separate ticket.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731034601
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
Review Comment:
Fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732341646
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
Review Comment:
I have rewritten this place.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732339290
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Right now, we start/stop it in the general life cycle. I do not see a reason to make it a demon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731063843
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
I do not know how I can do it right now.
May I left TODO here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731499882
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
+ }
+ }
+
+ try {
+ Thread.sleep(sleepInterval);
+ } catch (InterruptedException e) {
+ log.info("The timeout worker was interrupted, probably the client is stopping.");
+ }
+
+ updateHeartbeat();
+ }
+
+ } catch (Throwable t) {
+ // Call FH here.
Review Comment:
Added TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731504051
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
Review Comment:
I moved this class to another place, where it is available for both usages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731496703
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -764,4 +782,116 @@ private InetSocketAddress getFromCacheOrCreateResolved(ClusterNode recipientNode
return address != null ? address : createResolved(recipientNode.address());
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
Review Comment:
I moved the code in one palce.
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -764,4 +782,116 @@ private InetSocketAddress getFromCacheOrCreateResolved(ClusterNode recipientNode
return address != null ? address : createResolved(recipientNode.address());
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
Review Comment:
I moved the code to one place.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729132437
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java:
##########
@@ -121,6 +120,10 @@ else if (err instanceof TimeoutException) // Translate timeout exception.
blockedMsgs.add(msgData);
+ if (timeoutMs > 0) {
+ fut.orTimeout(timeoutMs, TimeUnit.MILLISECONDS);
Review Comment:
It is a code for tests in which we block RAFT commands.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731503030
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -287,10 +302,9 @@ private CompletableFuture<NetworkMessage> invoke0(ClusterNode recipient, Channel
long correlationId = createCorrelationId();
- CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<NetworkMessage>()
- .orTimeout(timeout, TimeUnit.MILLISECONDS);
+ CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<>();
- requestsMap.put(correlationId, responseFuture);
+ requestsMap.put(correlationId, new TimeoutObject(timeout > 0 ? coarseCurrentTimeMillis() + timeout : 0, responseFuture));
Review Comment:
I took this code from our repository. It was for a thin client.
Anyway, timeout 0 is something specific that will be used in a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729127654
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java:
##########
@@ -87,23 +87,22 @@ public ClusterService clusterService() {
) {
CompletableFuture<Message> fut = new CompletableFuture<>();
- fut.orTimeout(timeoutMs, TimeUnit.MILLISECONDS).
- whenComplete((res, err) -> {
- assert !(res == null && err == null) : res + " " + err;
+ fut.whenComplete((res, err) -> {
Review Comment:
We still set the timeout on sending the operation.
It just has excessive usage of `orTimeoput`, that I removbed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729049745
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Do you think we can share the worker among clients started on one JVM?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729093737
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Why not :) The collection is concurrent anyway
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729040986
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -87,7 +97,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
private final AtomicLong reqId = new AtomicLong(1);
/** Pending requests. */
- private final Map<Long, ClientRequestFuture<?>> pendingReqs = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, TimeoutObject> pendingReqs = new ConcurrentHashMap<>();
Review Comment:
Because the way that we use the map requires it has to be concurrent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729044675
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
Review Comment:
If someone uses a plain map here, they get a compile error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1730890823
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
We iterate over the map to copy it anyway, it can't be faster than just iterating. Something must be wrong with the benchmark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732313502
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -287,10 +302,9 @@ private CompletableFuture<NetworkMessage> invoke0(ClusterNode recipient, Channel
long correlationId = createCorrelationId();
- CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<NetworkMessage>()
- .orTimeout(timeout, TimeUnit.MILLISECONDS);
+ CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<>();
- requestsMap.put(correlationId, responseFuture);
+ requestsMap.put(correlationId, new TimeoutObject(timeout > 0 ? coarseCurrentTimeMillis() + timeout : 0, responseFuture));
Review Comment:
Thin client treats 0 as "no timeout" (which is documented in the config javadoc), this might be different in other places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov merged PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729090244
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
This approach is better than that where we use a concurrent iterator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729094774
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
Ok, but I just imagined: what if we have a million entries in that map...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1730893142
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -764,4 +782,116 @@ private InetSocketAddress getFromCacheOrCreateResolved(ClusterNode recipientNode
return address != null ? address : createResolved(recipientNode.address());
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
Review Comment:
Duplicate code. As mentioned above, we can put this into core module and reuse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731609129
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Is a demon thread an option?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729133920
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/FeatureTimeoutBenchmark.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMillis;
+
+import com.lmax.disruptor.dsl.Disruptor;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Feature timeout benchmark.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FeatureTimeoutBenchmark {
+ private static AtomicLong ID_GEN = new AtomicLong();
+
+ /** Active operation. */
+ public ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ private IgniteThread timeoutWorker;
+
+ private ConcurrentMap<Long, CompletableFuture<?>> futs;
+
+ private Disruptor<TimeoutObject> disruptor;
+
+ @Param({"false", "true"})
+ private boolean useFutureEmbeddedTimeout;
+
+ /**
+ * Prepare to start the benchmark.
+ */
+ @Setup
+ public void setUp() {
+ if (useFutureEmbeddedTimeout) {
+ futs = new ConcurrentHashMap<>();
+ } else {
+ requestsMap = new ConcurrentHashMap<>();
+ timeoutWorker = new IgniteThread("benchmark", "timeout-worker", new TimeoutRunnable(requestsMap));
+ timeoutWorker.start();
+ }
+ }
+
+ /**
+ * Closes resources.
+ */
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ if (useFutureEmbeddedTimeout) {
+ for (CompletableFuture<?> fut : futs.values()) {
+ if (!fut.isDone()) {
+ try {
+ fut.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ assert e.getCause() instanceof TimeoutException : "Unexpected exception type: " + e.getCause().getClass();
+ } catch (TimeoutException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ futs.clear();
+ futs = null;
+ } else {
+ assert waitForCondition(requestsMap::isEmpty, 10_000);
+
+ timeoutWorker.interrupt();
+ timeoutWorker = null;
+ }
+ }
+
+ /**
+ * Benchmark for KV upsert via embedded client.
+ */
+ @Benchmark
+ public void test() {
+ if (useFutureEmbeddedTimeout) {
+ for (int i = 0; i < 10; i++) {
+ var fut = new CompletableFuture<Void>();
+
+ futs.put(ID_GEN.incrementAndGet(), fut);
+
+ fut.orTimeout(10, TimeUnit.MILLISECONDS);
+
+ if (futs.size() > 100_000) {
Review Comment:
I avoided GC here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729100274
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
Review Comment:
But we would have to create exception instances. This code is not in a low-latency path, and I was worried about GC here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732215385
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -395,7 +419,7 @@ private <T> void complete(ClientRequestFuture<T> pendingReq, ClientMessageUnpack
private void processNextMessage(ClientMessageUnpacker unpacker) throws IgniteException {
if (protocolCtx == null) {
// Process handshake.
- complete(pendingReqs.remove(-1L), unpacker);
+ complete((ClientRequestFuture<?>) pendingReqs.remove(-1L).future(), unpacker);
Review Comment:
Let's make `TimeoutObject` generic with `T` being the type inside the `CompletableFuture`. This will allow to avoid these casts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731145209
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
Do you have any questions, what exactly is not clear?
* Create a separate class
* Put a static concurrent map there
* Create one thread for timeout checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732208463
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
+ updateHeartbeat();
+
+ timeoutObject = entry.getValue();
+
+ assert timeoutObject != null : "Unexpected null on the timeout queue.";
+
+ if (timeoutObject.getEndTime() > 0 && now > timeoutObject.getEndTime()) {
+ CompletableFuture<?> fut = timeoutObject.getFuture();
+
+ if (!fut.isDone()) {
+ fut.completeExceptionally(new TimeoutException());
+ }
+
+ // Client-facing future will fail with a timeout, but internal ClientRequestFuture will stay in the map -
+ // otherwise we'll fail with "protocol breakdown" error when a late response arrives from the server.
+ // requestsMap.remove(entry.getKey(), timeoutObject)
Review Comment:
Then why do you leave the commented-out line? You explained why you don't remove from the map, isn't this enough?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1730888379
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -158,6 +178,8 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
}
+ new IgniteThread(timeoutWorker).start();
Review Comment:
One timeout worker for all clients is enough. Maybe even one timeout worker per JVM for all Ignite needs, we can put it in the core module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732616857
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -314,13 +335,16 @@ private <T> ClientRequestFuture<T> send(
long id,
@Nullable PayloadWriter payloadWriter,
@Nullable PayloadReader<T> payloadReader,
- @Nullable CompletableFuture<PayloadInputChannel> notificationFut) {
+ @Nullable CompletableFuture<PayloadInputChannel> notificationFut,
+ long timeout
+ ) {
if (closed()) {
throw new IgniteClientConnectionException(CONNECTION_ERR, "Channel is closed", endpoint());
}
ClientRequestFuture<T> fut = new ClientRequestFuture<>(payloadReader, notificationFut);
- pendingReqs.put(id, fut);
+
+ pendingReqs.put(id, new TimeoutObject<>(timeout > 0 ? coarseCurrentTimeMillis() + timeout : 0, fut));
Review Comment:
Can we make `TimeoutObject` an interface? We already use a custom future class (`ClientRequestFuture`), it can hold the timeout as well, so we avoid an allocation of `TimeoutObject` wrapper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729090244
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
This approach is better than one where we use a concurrent iterator. I tried on the benchmark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1729090509
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -87,7 +97,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
private final AtomicLong reqId = new AtomicLong(1);
/** Pending requests. */
- private final Map<Long, ClientRequestFuture<?>> pendingReqs = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, TimeoutObject> pendingReqs = new ConcurrentHashMap<>();
Review Comment:
Why? :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732346669
##########
modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java:
##########
@@ -287,10 +302,9 @@ private CompletableFuture<NetworkMessage> invoke0(ClusterNode recipient, Channel
long correlationId = createCorrelationId();
- CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<NetworkMessage>()
- .orTimeout(timeout, TimeUnit.MILLISECONDS);
+ CompletableFuture<NetworkMessage> responseFuture = new CompletableFuture<>();
- requestsMap.put(correlationId, responseFuture);
+ requestsMap.put(correlationId, new TimeoutObject(timeout > 0 ? coarseCurrentTimeMillis() + timeout : 0, responseFuture));
Review Comment:
That is exactly why I wrote it in this way (TimeoutWorker#body).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1732482546
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -395,7 +419,7 @@ private <T> void complete(ClientRequestFuture<T> pendingReq, ClientMessageUnpack
private void processNextMessage(ClientMessageUnpacker unpacker) throws IgniteException {
if (protocolCtx == null) {
// Process handshake.
- complete(pendingReqs.remove(-1L), unpacker);
+ complete((ClientRequestFuture<?>) pendingReqs.remove(-1L).future(), unpacker);
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#discussion_r1731422572
##########
modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java:
##########
@@ -782,4 +803,120 @@ private class HeartbeatTask extends TimerTask {
}
}
}
+
+ /**
+ * Timeout object worker.
+ */
+ private static class TimeoutWorker extends IgniteWorker {
+ /** Worker sleep interval. */
+ private final long sleepInterval = getLong("IGNITE_TIMEOUT_WORKER_SLEEP_INTERVAL", 500);
+
+ /** Active operation. */
+ public final ConcurrentMap<Long, TimeoutObject> requestsMap;
+
+ /**
+ * Constructor.
+ *
+ * @param log Logger.
+ * @param igniteInstanceName Name of the Ignite instance this runnable is used in.
+ * @param name Worker name. Note that in general thread name and worker (runnable) name are two different things. The same
+ * worker can be executed by multiple threads and therefore for logging and debugging purposes we separate the two.
+ * @param requestsMap Active operations.
+ */
+ public TimeoutWorker(
+ IgniteLogger log,
+ String igniteInstanceName,
+ String name,
+ ConcurrentMap<Long, TimeoutObject> requestsMap
+ ) {
+ super(log, igniteInstanceName, name, null);
+
+ this.requestsMap = requestsMap;
+ }
+
+ @Override
+ protected void body() {
+ try {
+ TimeoutObject timeoutObject;
+
+ while (!isCancelled()) {
+ long now = coarseCurrentTimeMillis();
+
+ for (Entry<Long, TimeoutObject> entry : new HashMap<>(requestsMap).entrySet()) {
Review Comment:
I tested it again and did not find any difference. I will use the map iterator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20869 Got rid of using CompletableFuture#orTimeout method on operations hot path [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on PR #4261:
URL: https://github.com/apache/ignite-3/pull/4261#issuecomment-2309617554
@vldpyatkov could you please try running `ClientHeartbeatBenchmark` with the following modification to enable timeouts:
```
client = (TcpIgniteClient) IgniteClient.builder()
.addresses("127.0.0.1:" + testServer.port())
+ .operationTimeout(5000)
.build();
```
On my machine, I don't see any difference between main and your branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org