+ * Triggered only once per dispatcher lifecycle. + *
+ * Not triggered if dispatcher is stopped normally with {@link VSDispatcher#close()}.
+ */
abstract void onDisconnected();
- abstract void onReconnected();
+ /**
+ * Triggered when the first connection is established.
+ */
+ abstract void onConnected();
/**
- * @return true if transport is already known to be unrecoverable
+ * Triggered when transport is stopped (e.g. connection lost) but may be recoverable.
+ *
+ * @return true if transport is already known to be unrecoverable (and recovery should be stopped right away)
*/
- abstract boolean onTransportStopped(VSocketRecoveryInfo recoveryInfo);
+ abstract boolean onTransportRecoveryStart(VSocketRecoveryInfo recoveryInfo);
/**
+ * Triggered when transport recovery have to stop (because of timeout or dispatcher shutdown).
+ *
* @return true if transport was permanently lost (can't be recovered anymore)
*/
- abstract boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo);
-}
\ No newline at end of file
+ abstract boolean onTransportRecoveryStop(VSocketRecoveryInfo recoveryInfo);
+}
diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannelImpl.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannelImpl.java
index eb89ef03..65365cf0 100644
--- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannelImpl.java
+++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannelImpl.java
@@ -315,6 +315,7 @@ public void close (boolean terminate) {
} catch (ConnectionAbortedException x) {
LOGGER.log (Level.FINE, "Error sending disconnect.", x);
} catch (InterruptedException x) {
+ Thread.currentThread().interrupt();
LOGGER.log (Level.FINE, "Sending disconnect interrupted.", x);
} catch (Exception x) {
LOGGER.log (Level.WARNING, "Error sending disconnect", x);
diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClient.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClient.java
index a5fcad63..65206c2a 100644
--- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClient.java
+++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClient.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2024 EPAM Systems, Inc
+ * Copyright 2026 EPAM Systems, Inc
*
* See the NOTICE file distributed with this work for additional information
* regarding copyright ownership. Licensed under the Apache License,
@@ -17,26 +17,29 @@
package com.epam.deltix.util.vsocket;
import com.epam.deltix.util.ContextContainer;
+import com.epam.deltix.util.concurrent.QuickExecutor;
+import com.epam.deltix.qsrv.hf.spi.conn.DisconnectEventListener;
import com.epam.deltix.util.io.GUID;
import com.epam.deltix.util.io.IOUtil;
-import com.epam.deltix.util.io.aeron.DXAeron;
import com.epam.deltix.util.io.offheap.OffHeap;
import com.epam.deltix.util.lang.Disposable;
-import com.epam.deltix.util.concurrent.QuickExecutor;
-import com.epam.deltix.qsrv.hf.spi.conn.DisconnectEventListener;
import com.epam.deltix.util.lang.DisposableListener;
import com.epam.deltix.util.time.GlobalTimer;
import com.epam.deltix.util.time.TimeKeeper;
+import com.epam.deltix.util.time.TimerRunner;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import org.jetbrains.annotations.ApiStatus;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.VisibleForTesting;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket;
import java.io.*;
-import java.net.*;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
import java.util.Date;
-import java.util.TimerTask;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.logging.Level;
@@ -53,10 +56,14 @@ public class VSClient extends ConnectionStateListener implements Disposable, Dis
public static final boolean SSL_TERMINATION = Boolean.getBoolean(SSL_TERMINATION_PROPERTY);
//private static final int MAX_TRANSPORT_RECONNECT_ATTEMPTS = Integer.getInteger("TimeBase.network.VSClient.maxTransportReconnectAttempts", 5);
- private static final int TRANSPORT_RECONNECT_ATTEMPT_INTERVAL = Integer.getInteger("TimeBase.network.VSClient.transportReconnectAttemptInterval", 1000);
+ private final int transportReconnectAttemptInterval;
+ private final int socketSendBufferSize;
+ private final int socketReceiveBufferSize;
private static final int RE_ATTEMPT_EXTRA_DELAY = 10; // Extra delay to avoid situation when we re-schedule task due to timer jitter
+ private static final VSClientOptions DEFAULT_CLIENT_OPTIONS = new VSClientOptions();
+
private String host;
private int port;
private int numTransportChannels = 3;
@@ -66,19 +73,19 @@ public class VSClient extends ConnectionStateListener implements Disposable, Dis
private final Object dispatcherLock = new Object();
private String clientId;
- private long serverTime = -1;
+ private long serverTime = -1;
private volatile DisconnectEventListener listener;
private int reconnectInterval;
private VSCompression serverCompression;
- private int soTimeout = Integer.getInteger("TimeBase.network.VSClient.soTimeout", 5000);
- private int timeout = Integer.getInteger("TimeBase.network.VSClient.timeout", 5000);
+ private int soTimeout;
+ private int timeout;
private boolean enableSSL = false;
private final boolean sslTermination;
private int sslPort = 0;
- private SSLContext sslContext;
+ private SSLContext sslContext;
private final ContextContainer contextContainer;
@@ -88,7 +95,8 @@ public class VSClient extends ConnectionStateListener implements Disposable, Dis
private volatile boolean closed = false;
// Elements should be sorted (when possible) by time of last reconnection attempt however there is no strict enforcement for this.
- // New broken sockets should be added to the head of the queue
+ // New broken sockets should be added to the head of the queue.
+ // Broken sockets that failed to reconnect should be added to the tail of the queue.
private final ConcurrentLinkedDeque It returns true during reconnecting phase because it would be inconsistent to return false,
+ * considering that reconnecting state does not trigger "disconnected" event.
+ *
+ * In most cases you should use {@link #tryGetConnectionStatus()} instead.
+ *
+ * @return true if connected or reconnecting, false otherwise
+ */
public boolean isConnected() {
- return dispatcher != null && dispatcher.hasAvailableTransport();
+ return dispatcher != null && dispatcher.isConnectedOrReconnecting();
+ }
+
+ /**
+ * Checks if client is fully connected right now.
+ * Will not wait and will return false if reconnecting.
+ *
+ * @return true if connected and NOT reconnecting, false otherwise
+ */
+ public boolean isConnectedAndNotReconnecting() {
+ return dispatcher != null && dispatcher.isConnectedAndNotReconnecting();
}
/**
* Return true, if it has CONNECTED state.
- * Return false, if it has DISCONNECTED state.
- * Otherwise, waits at least {@link #reconnectInterval} until status gets CONNECTED or DISCONNECTED.
+ * Return false, if it has DISCONNECTED/DISCONNECTING state.
+ * Otherwise, waits until status gets CONNECTED or DISCONNECTED.
*
* @return true if connected, false if disconnected
*/
@@ -376,12 +420,11 @@ private Socket setupSocket() throws IOException {
socket.setTcpNoDelay(true);
// Sets socket buffer sizes.
- // Please note that later socket also will be additionally configured in VSocketImpl.setUpSocket() method.
- // However, that happens only after socket gets connected.
// It's important to configure receive buffer size before connection is established
// to allow it to use TCP window size greater than 64kb.
// That's why we have to do that here.
- VSocketImpl.configureBufferSizes(socket);
+ socket.setReceiveBufferSize(this.socketReceiveBufferSize);
+ socket.setSendBufferSize(this.socketSendBufferSize);
// Connect
socket.connect(socketAddress, timeout);
@@ -578,7 +621,7 @@ VSocket openTransport () throws IOException {
assert numBytesRecieved == 0; // new connections should have = 0;
this.serverCompression = Enum.valueOf(VSCompression.class, compression);
}
-
+
ok = true;
} finally {
if (!ok)
@@ -614,6 +657,7 @@ else if (serverCompression == VSCompression.ON)
try {
vsc.sendConnect ();
} catch (InterruptedException x) {
+ Thread.currentThread().interrupt();
throw new InterruptedIOException ();
}
@@ -633,7 +677,11 @@ private void close(boolean waitForChannelsToFinish) {
VSDispatcher d = dispatcher;
// If dispatcher is null, then we already disconnected or even never were connected.
- triggerDisconnectEvent = d != null;
+ // If dispatcher is in shutdown state, then disconnect event already was triggered.
+ // Note that this check does not give 100% guarantee that disconnect event will be triggered no more than once
+ // because of race between checking isShutdownState() and calling d.setStateListener(null).
+ // However, in practice this should be sufficient.
+ triggerDisconnectEvent = d != null && !d.isShutdownState();
if (d != null) {
d.setStateListener(null);
@@ -645,9 +693,14 @@ private void close(boolean waitForChannelsToFinish) {
dispatcher = null;
}
-
- // https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/-/issues/1269
// Trigger a disconnect event, so any disconnect listeners can be notified.
+ // https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/-/issues/1269
+ // However, this also results that onDisconnect event will be triggered even if no "unexpected disconnect" actually happened.
+ // So while VSDispatcher does not trigger disconnect event if it shut down gracefully, VSClient.close() will still trigger it.
+ // TODO: Decide if we want to call .onDisconnected() in case of normal shutdown.
+ // TODO: This should be reviewed after TickDBClient refactor. We may want to completely remove this call
+ // as updated VSDispatcher already triggers disconnect event on unexpected disconnects
+ // and state change that is caused by TickDBClient closing the connection may be handled in TickDBClient itself.
if (triggerDisconnectEvent) {
DisconnectEventListener listenerRef = listener;
if (listenerRef != null) {
@@ -661,7 +714,7 @@ public void setDisconnectedListener(DisconnectEventListener
}
@Override
- boolean onTransportStopped(VSocketRecoveryInfo recoveryInfo) {
+ boolean onTransportRecoveryStart(VSocketRecoveryInfo recoveryInfo) {
if (dispatcher != null) {
// TODO: Ensure that we can't get duplicate instance of socket in the broken list
broken.addFirst(recoveryInfo);
@@ -674,7 +727,7 @@ boolean onTransportStopped(VSocketRecoveryInfo recoveryInfo) {
}
@Override
- boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) {
+ boolean onTransportRecoveryStop(VSocketRecoveryInfo recoveryInfo) {
try {
synchronized (recoveryInfo) {
while (recoveryInfo.isRecoveryAttemptInProgress()) {
@@ -697,12 +750,13 @@ boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) {
return recoveryFailed || (!removed && !recoveryInfo.isRecoverySucceeded());
}
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
return true;
}
}
@Override
- void onReconnected() {
+ void onConnected() {
if (listener != null)
listener.onReconnected();
}
@@ -730,8 +784,6 @@ public void disposed (VSDispatcher d) {
contextContainer.getQuickExecutor().shutdownInstance();
dispatcher = null;
-
- onDisconnected();
}
}
}
@@ -748,4 +800,15 @@ public int getSSLPort() {
public String toString () {
return ("VSClient (" + host + ":" + port + ")");
}
+
+ /** Returns default client options based on system properties, with "sslTermination" set to provided value */
+ private static VSClientOptions withSslTermination(boolean sslTermination) {
+ if (sslTermination == DEFAULT_CLIENT_OPTIONS.isSslTermination()) {
+ return DEFAULT_CLIENT_OPTIONS;
+ } else {
+ VSClientOptions result = new VSClientOptions();
+ result.setSslTermination(sslTermination);
+ return result;
+ }
+ }
}
diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClientOptions.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClientOptions.java
new file mode 100644
index 00000000..f69e71f5
--- /dev/null
+++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSClientOptions.java
@@ -0,0 +1,82 @@
+package com.epam.deltix.util.vsocket;
+
+/*
+ * Copyright 2026 EPAM Systems, Inc
+ *
+ * See the NOTICE file distributed with this work for additional information
+ * regarding copyright ownership. Licensed 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.
+ */
+import org.jetbrains.annotations.ApiStatus;
+
+@ApiStatus.Experimental
+public class VSClientOptions {
+ // Init fields with default values
+
+ private int socketSendBufferSize = VSocketImpl.SOCKET_SEND_BUFFER_SIZE;
+ private int socketReceiveBufferSize = VSocketImpl.SOCKET_RECEIVE_BUFFER_SIZE;
+ private boolean sslTermination = VSClient.SSL_TERMINATION;
+ private int transportReconnectAttemptInterval = Integer.getInteger("TimeBase.network.VSClient.transportReconnectAttemptInterval", 1000);
+
+ // Controls socket timeout ("soTimeout") during the initial handshake. After handshake is complete, soTimeout is always "0" (infinite).
+ private int handshakeSocketTimeout = Integer.getInteger("TimeBase.network.VSClient.soTimeout", 5000);
+ private int socketConnectTimeout = Integer.getInteger("TimeBase.network.VSClient.timeout", 5000);
+
+
+ public int getSocketReceiveBufferSize() {
+ return socketReceiveBufferSize;
+ }
+
+ public void setSocketReceiveBufferSize(int socketReceiveBufferSize) {
+ this.socketReceiveBufferSize = socketReceiveBufferSize;
+ }
+
+ public int getSocketSendBufferSize() {
+ return socketSendBufferSize;
+ }
+
+ public void setSocketSendBufferSize(int socketSendBufferSize) {
+ this.socketSendBufferSize = socketSendBufferSize;
+ }
+
+ public int getHandshakeSocketTimeout() {
+ return handshakeSocketTimeout;
+ }
+
+ public void setHandshakeSocketTimeout(int handshakeSocketTimeout) {
+ this.handshakeSocketTimeout = handshakeSocketTimeout;
+ }
+
+ public boolean isSslTermination() {
+ return sslTermination;
+ }
+
+ public void setSslTermination(boolean sslTermination) {
+ this.sslTermination = sslTermination;
+ }
+
+ public int getSocketConnectTimeout() {
+ return socketConnectTimeout;
+ }
+
+ public void setSocketConnectTimeout(int socketConnectTimeout) {
+ this.socketConnectTimeout = socketConnectTimeout;
+ }
+
+ public int getTransportReconnectAttemptInterval() {
+ return transportReconnectAttemptInterval;
+ }
+
+ public void setTransportReconnectAttemptInterval(int transportReconnectAttemptInterval) {
+ this.transportReconnectAttemptInterval = transportReconnectAttemptInterval;
+ }
+}
diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcher.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcher.java
index 32f599e4..c3082a66 100644
--- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcher.java
+++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcher.java
@@ -27,14 +27,20 @@
import com.epam.deltix.util.time.TimerRunner;
import com.epam.deltix.util.memory.DataExchangeUtils;
import net.jcip.annotations.GuardedBy;
+import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
import java.io.EOFException;
import java.io.IOException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
/**
@@ -51,7 +57,14 @@ public final class VSDispatcher implements Disposable {
@GuardedBy("transportChannels")
private final ObjectHashSet Corresponding transport channel will be closed after this method returns.
+ *
+ * This method is expected to block until logical transport gets recovered or declared unrecoverably broken.
+ * In case of recovery failure, expected to trigger dispatcher shutdown, as loss of single transport channel
+ * means loss of data and inconsistent state for client and server.
+ *
+ * Multiple transport channels may be lost concurrently, so this method may be executed concurrently.
+ * In that case, threads may compete for changing dispatcher state.
+ */
void transportStopped (VSTransportChannel channel, Throwable ex) {
- IOException iex = ex instanceof IOException ? (IOException)ex : null;
+ if (!(ex instanceof Exception)) {
+ // This means major failure, possibly OOM or other serious error.
+ VSProtocol.LOGGER.log(Level.SEVERE, "Critical error on transport channel. Remote address: " + channel.socket.getRemoteAddress() + ". Dispatcher: " + this, ex);
+ // Just close dispatcher right away
+ close();
+ return;
+ }
Level disconnectLogLevel = ex instanceof EOFException ? Level.FINE : Level.INFO;
if (VSProtocol.LOGGER.isLoggable(disconnectLogLevel)) {
@@ -290,166 +374,279 @@ void transportStopped (VSTransportChannel channel, Throwa
}
long startTime = System.currentTimeMillis();
- long endTime = startTime + reconnectInterval;
+ long endTime = startTime + lingerInterval;
+ boolean registered = false; // true if we have registered this transport in "recoveringTransports" phaser
boolean wasCheckedIn;
- synchronized (transportChannels) {
- if (transportChannels.isEmpty()) // already closed
- return;
+ try {
+ synchronized (transportChannels) {
+ VSDispatcherState currentState = state.get();
+ if (currentState == VSDispatcherState.DISCONNECTING || currentState == VSDispatcherState.DISCONNECTED) {
+ // Dispatcher is already closing or closed, no need to recover transport
+ return;
+ }
- if (!transportChannels.remove(channel)) // check that channel already removed
- return;
+ if (!transportChannels.remove(channel)) // check if that channel is already removed
+ return;
- synchronized (freeChannels) {
- wasCheckedIn = freeChannels.remove(channel);
- assert wasCheckedIn == !channel.checkedOut;
- freeChannels.notifyAll();
- }
+ // From this point we consider that we are recovering this transport channel.
+
+ // Counter incremented before state change,
+ // so that should be impossible to see CONNECTING with 0 recovering transports and still pending recovery attempt.
+ recoveringTransports.register();
+ registered = true;
+ state.compareAndSet(VSDispatcherState.CONNECTED, VSDispatcherState.RECONNECTING);
+
+ synchronized (freeChannels) {
+ wasCheckedIn = freeChannels.remove(channel);
+ assert wasCheckedIn == !channel.checkedOut;
+ freeChannels.notifyAll();
+ }
- if (!wasCheckedIn) {
- // Try to wait for the channel to become checked in
- wasCheckedIn = waitForTransportCheckIn(channel, startTime, endTime);
if (!wasCheckedIn) {
- if (VSProtocol.LOGGER.isLoggable(Level.INFO)) {
- VSProtocol.LOGGER.log(Level.INFO, "Error waiting to reconnect (transport was not checked in).");
+ // Try to wait for the channel to become checked in
+ wasCheckedIn = waitForTransportCheckIn(channel, startTime, endTime);
+ if (!wasCheckedIn) {
+ if (VSProtocol.LOGGER.isLoggable(Level.INFO)) {
+ VSProtocol.LOGGER.log(Level.INFO, "Error waiting to reconnect (transport was not checked in).");
+ }
}
}
}
+ ConnectionStateListener stateListener = this.stateListener;
- state = VSDispatcherState.CONNECTING;
- }
-
- boolean transportIsUnrecoverablyBroken = false;
+ boolean transportIsUnrecoverablyBroken = false;
+ try {
+ // trying to recover transport
+ VSocketRecoveryInfo recoveryInfo = new VSocketRecoveryInfo(channel.socket, endTime);
- // trying to recover transport
- VSocketRecoveryInfo recoveryInfo = new VSocketRecoveryInfo(channel.socket, startTime);
+ long now = System.currentTimeMillis();
+ if (wasCheckedIn && (now < endTime) && !isShutdownState()) {
- long now = System.currentTimeMillis();
- if (wasCheckedIn && (now < endTime)) {
+ if (stateListener != null) {
+ if (stateListener.onTransportRecoveryStart(recoveryInfo)) {
+ transportIsUnrecoverablyBroken = true;
+ }
+ }
- if (stateListener != null) {
- if (stateListener.onTransportStopped(recoveryInfo)) {
+ if (remoteConnected && !transportIsUnrecoverablyBroken) {
+ // System.out.println("WAITED: remoteConnected=" + remoteConnected + " transportIsUnrecoverablyBroken=" + transportIsUnrecoverablyBroken);
+ try {
+ // We loop here waiting for recovery to complete or timeout to expire or dispatcher to be closed.
+ synchronized (recoveryInfo) {
+ long timeToWait;
+ while ((timeToWait = endTime - now) > 0 && recoveryInfo.isWaitingForRecovery() && remoteConnected && !isShutdownState()) {
+ recoveryInfo.wait(timeToWait);
+ if (recoveryInfo.isWaitingForRecovery() && remoteConnected) {
+ now = System.currentTimeMillis();
+ }
+ }
+ if (recoveryInfo.isRecoveryFailed()) {
+ transportIsUnrecoverablyBroken = true;
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ if (VSProtocol.LOGGER.isLoggable(Level.FINE))
+ VSProtocol.LOGGER.log(Level.FINE, "Error waiting to reconnect.", e);
+ }
+ } else {
+ //System.out.println("NOT WAITED: remoteConnected=" + remoteConnected + " transportIsUnrecoverablyBroken=" + transportIsUnrecoverablyBroken);
+ }
+ } else {
transportIsUnrecoverablyBroken = true;
+ if (VSProtocol.LOGGER.isLoggable(Level.FINE)) {
+ VSProtocol.LOGGER.log(Level.FINE, "Cancelled recovery of failed connection because of timeout on waiting for check-in from other thread. Remote address: " + getRemoteAddress());
+ }
}
- }
- if (remoteConnected && !transportIsUnrecoverablyBroken) {
- // System.out.println("WAITED: remoteConnected=" + remoteConnected + " transportIsUnrecoverablyBroken=" + transportIsUnrecoverablyBroken);
- try {
- // Try to wait for connection restore
- state = VSDispatcherState.CONNECTING;
-
- synchronized (recoveryInfo) {
- long timeToWait;
- while ((timeToWait = endTime - now) > 0 && recoveryInfo.isWaitingForRecovery() && remoteConnected) {
- recoveryInfo.wait(timeToWait);
- if (recoveryInfo.isWaitingForRecovery() && remoteConnected) {
- now = System.currentTimeMillis();
- }
- }
- if (recoveryInfo.isRecoveryFailed()) {
- transportIsUnrecoverablyBroken = true;
+ // In general, VSDispatcher don't have to shut down if transport recovery fails,
+ // because other transport channels may remain functional.
+ // However, in our current design, loss of single transport channel means loss of data
+ // and inconsistent state for client and server, so we have to shut down the dispatcher.
+ // The decision to shut down the dispatcher is delegated to the state listener.
+ if (stateListener != null) {
+ if (stateListener.onTransportRecoveryStop(recoveryInfo)) {
+ transportIsUnrecoverablyBroken = true;
+ }
+ }
+ } finally {
+ if (transportIsUnrecoverablyBroken || isShutdownState()) {
+ // We lost this transport channel and were unable to recover it (because of explicit error, timeout or triggered shutdown state).
+ // This means it is not possible to recover from this state, and we have to properly close the dispatcher.
+ // We need to close all remaining connections and explicitly notify user about that.
+
+ // Record a copy of state listener reference before updating state because it may be changed concurrently.
+ // If save "stateListener" before state update, then we can be sure that
+ // if we had non-null listener before state update, then we will have non-null listener for thread that gets "triggerDisconnectedEvent".
+ var savedStateListener = this.stateListener;
+
+ // Try to set state to DISCONNECTING, before decrementing recoveringTransports counter,
+ // so other thread will not switch into CONNECTED state if this was the last recovering transport.
+
+ VSDispatcherState stateBeforeUpdate = state.getAndUpdate(prevState -> {
+ switch (prevState) {
+ case CONNECTED:
+ // Should not happen
+ return VSDispatcherState.DISCONNECTING;
+ case RECONNECTING:
+ return VSDispatcherState.DISCONNECTING;
+ case DISCONNECTING:
+ return prevState; // remain in DISCONNECTING
+ case DISCONNECTED:
+ return prevState; // remain in DISCONNECTED
+ default:
+ throw new IllegalStateException("Unexpected dispatcher state: " + prevState);
}
+ });
+ // Disconnected event should be triggered only if we changed the state.
+ // So that event should be triggered only once per dispatcher lifetime.
+ // Also, it disables trigger of onDisconnected event if dispatcher is closed normally via direct call to close().
+ boolean triggerDisconnectedEvent = stateBeforeUpdate == VSDispatcherState.CONNECTED || stateBeforeUpdate == VSDispatcherState.RECONNECTING;
+
+ registered = false;
+ recoveringTransports.arriveAndDeregister();
+
+ processChannelRecoveryFailure(ex, triggerDisconnectedEvent, savedStateListener);
+ } else {
+ // Successfully recovered this transport channel
+ registered = false;
+ recoveringTransports.arriveAndDeregister();
+ int remaining = recoveringTransports.getUnarrivedParties();
+ if (!recoveringTransports.isTerminated() && remaining == 0) {
+ // All lost transport channels are recovered, try to update state to CONNECTED
+ state.getAndUpdate(prev -> {
+ if (prev == VSDispatcherState.RECONNECTING) {
+ return VSDispatcherState.CONNECTED;
+ } else {
+ return prev; // Keep state unchanged
+ }
+ });
}
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- if (VSProtocol.LOGGER.isLoggable(Level.FINE))
- VSProtocol.LOGGER.log(Level.FINE, "Error waiting to reconnect.", e);
}
- } else {
- //System.out.println("NOT WAITED: remoteConnected=" + remoteConnected + " transportIsUnrecoverablyBroken=" + transportIsUnrecoverablyBroken);
}
- } else {
- transportIsUnrecoverablyBroken = true;
- if (VSProtocol.LOGGER.isLoggable(Level.FINE)) {
- VSProtocol.LOGGER.log(Level.FINE, "Cancelled recovery of failed connection because of timeout on waiting for check-in from other thread. Remote address: " + getRemoteAddress());
- }
- }
- if (stateListener != null) {
- if (stateListener.onTransportBroken(recoveryInfo)) {
- transportIsUnrecoverablyBroken = true;
+ } finally {
+ if (registered) {
+ // In case of any unexpected error, ensure that we release the counter
+ recoveringTransports.arriveAndDeregister();
}
}
+ }
- if (transportIsUnrecoverablyBroken) {
- // We lost this transport channel and were unable to recover it.
- // This means that we lost at least some data and can't recover from this state.
- // We need to close all remaining connections and explicitly notify use about that.
-
- boolean wasConnected = remoteConnected;
-
- // mark that we lost transport completely
- state = VSDispatcherState.DISCONNECTED;
+ /**
+ * Triggered when transport channel recovery has failed and dispatcher must be disconnected.
+ * @param triggerClose if true, then current thread is the one that first detected unrecoverable transport failure and responsible for shutdown
+ */
+ private void processChannelRecoveryFailure(Throwable ex, boolean triggerClose, @Nullable ConnectionStateListener stateListenerCopy) {
+ boolean wasConnected = remoteConnected;
- // notify all waiting for transport that connection is lost
- onRemoteClosed();
+ // notify all waiting for transport that connection is lost
+ onRemoteClosed();
- if (ex instanceof SocketException || ex instanceof EOFException || ex instanceof SocketTimeoutException) {
- if (VSProtocol.LOGGER.isLoggable(Level.FINE))
- VSProtocol.LOGGER.log(Level.FINE, "Exception on transport channel. Remote address: " + getRemoteAddress(), ex);
- } else {
- VSProtocol.LOGGER.log(Level.SEVERE, "Exception on transport channel. Remote address: " + getRemoteAddress(), ex);
- }
+ if (ex instanceof SocketException || ex instanceof EOFException || ex instanceof SocketTimeoutException) {
+ if (VSProtocol.LOGGER.isLoggable(Level.FINE))
+ VSProtocol.LOGGER.log(Level.FINE, "Exception on transport channel. Remote address: " + getRemoteAddress(), ex);
+ } else {
+ VSProtocol.LOGGER.log(Level.SEVERE, "Exception on transport channel. Remote address: " + getRemoteAddress(), ex);
+ }
- if (wasConnected) {
- VSProtocol.LOGGER.log(Level.WARNING, "Disconnecting due to unrecoverable transport channel loss. Remote address: " + getRemoteAddress(), ex);
- } else {
- VSProtocol.LOGGER.log(Level.FINER, "Disconnecting (re-triggered) due to unrecoverable transport channel loss. Remote address: " + getRemoteAddress(), ex);
- }
+ if (wasConnected) {
+ VSProtocol.LOGGER.log(Level.WARNING, "Disconnecting due to unrecoverable transport channel loss. Remote address: " + getRemoteAddress(), ex);
+ } else {
+ VSProtocol.LOGGER.log(Level.FINER, "Disconnecting (re-triggered) due to unrecoverable transport channel loss. Remote address: " + getRemoteAddress(), ex);
+ }
- // and then notify all channels that we lost transport
- synchronized (channels) {
- for (VSChannelImpl vsChannel : channels)
- if (vsChannel != null)
- vsChannel.onDisconnected(iex);
+ // and then notify all channels that we lost transport
+ IOException iex = ex instanceof IOException ? (IOException)ex : null;
+ synchronized (channels) {
+ for (VSChannelImpl vsChannel : channels) {
+ if (vsChannel != null) {
+ // TODO: Review. Calling onDisconnected while holding "channels" lock may lead to deadlocks
+ vsChannel.onDisconnected(iex);
+ }
}
+ }
+ if (triggerClose) {
// notify state listener that connections lost
- if (stateListener != null)
- stateListener.onDisconnected();
+ if (stateListenerCopy != null) {
+ if (VSProtocol.LOGGER.isLoggable(Level.FINER)) {
+ VSProtocol.LOGGER.log(Level.FINER, "Notifying state listener about disconnection. Remote address: " + getRemoteAddress());
+ }
+ stateListenerCopy.onDisconnected();
+ }
close();
} else {
- state = VSDispatcherState.CONNECTED;
+ // If this thread is not responsible for closing dispatcher,
+ // just wait until dispatcher gets closed by other thread.
+ boolean success;
+ try {
+ success = closeLatch.await(lingerInterval + 1_000, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Failed waiting for dispatcher to close after transport recovery failure.", e);
+ }
+ if (!success) {
+ VSProtocol.LOGGER.log(Level.WARNING, "Timeout waiting for dispatcher to close after transport recovery failure. Remote address: " + getRemoteAddress());
+ // No other thread closed the dispatcher in a timely manner, close it ourselves. Even if it may break order between .onDisconnected() and .disposed() events.
+ close();
+ }
}
}
+ boolean isShutdownState() {
+ VSDispatcherState value = state.get();
+ return value == VSDispatcherState.DISCONNECTING || value == VSDispatcherState.DISCONNECTED;
+ }
+
/**
* Return true, if it has CONNECTED state.
- * Return false, if it has DISCONNECTED state.
- * Otherwise, waits at least {@link #reconnectInterval} until status gets CONNECTED or DISCONNECTED.
+ * Return false, if it has INITIAL, DISCONNECTED or DISCONNECTING state.
+ * Otherwise, waits at least {@link #lingerInterval} until status gets CONNECTED or DISCONNECTED.
*
* @return true if connected, false if disconnected
*/
public boolean tryGetConnectionStatus() {
-
- // set timeout > reconnectInterval
- int timeout = reconnectInterval * 2;
-
- long timeLimit = TimeKeeper.currentTime + timeout;
- if (timeLimit < 0) // overflow check
- timeLimit = Long.MAX_VALUE;
-
- long period = Math.min(timeout, 1000);
- try {
- while (TimeKeeper.currentTime < timeLimit) {
- if (state == VSDispatcherState.CONNECTED)
- return true;
- else if (state == VSDispatcherState.DISCONNECTED)
+ // Get current phase before checking state
+ int phase = recoveringTransports.getPhase();
+
+ // Read current status
+ switch (state.get()) {
+ case CONNECTED:
+ return true;
+ case INITIAL:
+ case DISCONNECTED:
+ case DISCONNECTING:
+ return false;
+ case RECONNECTING:
+ // Wait below
+ }
+
+ // What for the phase to change
+ recoveringTransports.awaitAdvance(phase);
+
+ // Check new status
+ switch (state.get()) {
+ case CONNECTED:
+ return true;
+ case INITIAL:
+ case DISCONNECTED:
+ case DISCONNECTING:
+ return false;
+ case RECONNECTING:
+ default: {
+ // Special case: we are still in reconnection state, even after phase advanced.
+ if (recoveringTransports.isTerminated()) {
return false;
- Thread.sleep(period);
+ }
+ // It may be possible that the waiting transport count was just decremented to zero
+ // but state was not updated yet. Check that.
+ return recoveringTransports.getUnarrivedParties() == 0;
}
- } catch (InterruptedException e) {
}
-
- if (state == VSDispatcherState.CONNECTED)
- return true;
- else if (state == VSDispatcherState.DISCONNECTED)
- return false;
-
- return false;
}
/**
@@ -466,7 +663,7 @@ private boolean waitForTransportCheckIn(VSTransportChannel channel, long now, lo
boolean checkedIn = false;
try {
- while (now < endTime && !checkedIn && !disposed.get()) {
+ while (now < endTime && !checkedIn && !isShutdownState()) {
transportChannels.wait(endTime - now);
now = System.currentTimeMillis();
synchronized (freeChannels) {
@@ -493,7 +690,7 @@ public void closeTransport() throws IOException, Interru
}
}
- public void checkIn (VSTransportChannel tc) {
+ void checkIn (VSTransportChannel tc) {
synchronized (transportChannels) {
if (transportChannels.contains(tc)) {
synchronized (freeChannels) {
@@ -502,6 +699,7 @@ public void checkIn (VSTransportChannel tc) {
freeChannels.notify();
}
} else {
+ // This was removed from dispatcher, possibly channel recovery in progress.
synchronized (freeChannels) {
tc.checkedOut = false;
}
@@ -517,8 +715,9 @@ VSTransportChannel checkOut ()
{
synchronized (freeChannels) {
for (;;) {
- if (state != VSDispatcherState.CONNECTED && !remoteConnected)
+ if (isShutdownState() && !remoteConnected) {
throw new ConnectionAbortedException("Connection aborted from remote side [" + getRemoteAddress() + "]");
+ }
if (!freeChannels.isEmpty ()) {
VSTransportChannel channel = freeChannels.pop();
@@ -565,7 +764,7 @@ public void close(boolean wait) {
void onRemoteClosed() {
remoteConnected = false;
-
+
// notify all waiting threads in checkOut()
synchronized (freeChannels) {
freeChannels.notifyAll();
@@ -573,9 +772,11 @@ void onRemoteClosed() {
}
private void sendClosing() {
- if (!remoteConnected || state != VSDispatcherState.CONNECTED)
+ // TODO: In theory we can try to check if there are any free transport channels and try to use them to send
+ // the close message. But in practice, if we are not connected anymore, then we are not very likely to succeed.
+ if (!remoteConnected || state.get() != VSDispatcherState.CONNECTED)
return;
-
+
VSTransportChannel channel = null;
try {
byte[] buffer = new byte[2];
@@ -590,10 +791,22 @@ private void sendClosing() {
}
}
+ @Override
public void close () {
sendClosing();
-
+
+ // Change state to DISCONNECTING if it was not DISCONNECTED already.
+ // This state change disables triggering of stateListener.onDisconnected() on transport channel error.
+ // So normal dispatcher.close() will not trigger onDisconnected() event.
+ state.getAndUpdate(prevState -> {
+ if (prevState == VSDispatcherState.DISCONNECTED) {
+ return VSDispatcherState.DISCONNECTED;
+ } else {
+ return VSDispatcherState.DISCONNECTING;
+ }
+ });
+
synchronized (transportChannels) {
for (VSTransportChannel tc : transportChannels)
Util.close (tc);
@@ -602,7 +815,6 @@ public void close () {
transportChannels.notify();
}
- state = VSDispatcherState.DISCONNECTED;
remoteConnected = false;
// disable free channels to prevent locking on code below
@@ -631,8 +843,15 @@ public void close () {
t.cancel(); // stop timer thread
timer = null; // for GC
- if (disposed.compareAndSet(false, true))
- notifyListeners();
+ VSDispatcherState prevState = state.getAndUpdate(x -> VSDispatcherState.DISCONNECTED);
+ if (prevState != VSDispatcherState.DISCONNECTED) {
+ // This may be triggered only once per dispatcher lifetime
+ notifyDisposedEventListeners();
+ }
+
+ recoveringTransports.forceTermination();
+
+ closeLatch.countDown();
}
VSChannelImpl newChannel (int inCapacity, int outCapacity, boolean compressed) {
@@ -643,6 +862,9 @@ VSChannelImpl newChannel (int inCapacity, int outCapacity, boolean
}
synchronized (channels) {
+ if (activeChannels >= channelsLimit && channelsLimit > 0)
+ throw new IllegalStateException("Attempt to create new channel above channels limit = " + channelsLimit);
+
int localId = channels.indexOf (null);
boolean extend = localId < 0;
@@ -654,7 +876,7 @@ VSChannelImpl newChannel (int inCapacity, int outCapacity, boolean
index += isClient ? -1 : 1;
vsc = new VSChannelImpl (this, inCapacity, outCapacity, compressed, localId, index, contextContainer);
-
+
if (extend) {
channels.add (vsc);
} else {
@@ -667,7 +889,7 @@ VSChannelImpl newChannel (int inCapacity, int outCapacity, boolean
}
return (vsc);
- }
+ }
public VSChannel [] getVirtualChannels () {
synchronized (channels) {
@@ -703,6 +925,7 @@ long getLatency() {
try {
return (tc = checkOut()).getLatency();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
return 0;
} catch (ConnectionAbortedException e) {
return 0;
@@ -727,21 +950,21 @@ void channelClosed (VSChannelImpl vsc) {
}
}
- public void addDisposableListener(DisposableListener listener) {
+ public void addDisposableListener(DisposableListener Run the test with -Xmx500m to see the problem.
+ */
+@SuppressWarnings("NewClassNamingConvention")
+public class Test_VSocketChannelLeak {
+ private static final boolean enableCloseFix = true;
+ private static final int ITERATIONS = 1000;
+ private static final int PAYLOAD_SIZE = 1_024 * 1_024; // 1 MB
+ private static final boolean waitForFreeMem = false; // Needed for CI env - it's slow
+
+// public static void main (String [] args) throws Exception {
+// testImpl();
+// //Thread.sleep(Long.MAX_VALUE);
+// }
+
+ @Ignore // Fails on CI
+ @Test(timeout = 60_000)
+ public void test() throws IOException, InterruptedException {
+ testImpl();
+ }
+
+ @SuppressWarnings("Convert2Lambda")
+ private static void testImpl() throws IOException, InterruptedException {
+ VSServer server = new VSServer(0);
+
+ AtomicLong openChannels = new AtomicLong();
+ AtomicLong closedChannels = new AtomicLong();
+
+ ExecutorService executorService = Executors.newCachedThreadPool();
+
+ server.setConnectionListener(new VSConnectionListener() {
+ @Override
+ public void connectionAccepted(QuickExecutor executor, VSChannel serverChannel) {
+ openChannels.incrementAndGet();
+
+ // This payload will be kept in memory until the channel is closed
+ byte[] payload = new byte[PAYLOAD_SIZE];
+ payload[0] = 1;
+
+ serverChannel.addDisposableListener(new DisposableListener<>() {
+ @Override
+ public void disposed(VSChannel resource) {
+ byte val = payload[11];
+ if (val != 0) {
+ System.out.println("Should never happen");
+ }
+ // Intentionally do not remove the listener from the channel to release the memory only if channel is released
+ }
+ });
+
+ if (enableCloseFix) {
+ executorService.submit(() -> {
+ DataInputStream dis = serverChannel.getDataInputStream();
+ //noinspection TryFinallyCanBeTryWithResources
+ try {
+ while (true) {
+ try {
+ dis.readByte();
+ } catch (EOFException e) {
+ // Graceful close
+ break;
+ } catch (IOException e) {
+ break;
+ }
+ }
+ } finally {
+ serverChannel.close(); // This will release the memory
+ closedChannels.incrementAndGet();
+ }
+ });
+ }
+ }
+ });
+ server.setDaemon(true);
+ server.start();
+ System.out.println("Server started on " + server.getLocalPort());
+
+ try {
+ createConnections("localhost", server.getLocalPort());
+ } finally {
+ System.out.println("Open channels: " + openChannels.get());
+ System.out.println("Closed channels: " + closedChannels.get());
+ }
+
+ server.close();
+ executorService.shutdown();
+ }
+
+ public static void createConnections(String host, int port)
+ throws IOException, InterruptedException {
+
+ VSClient client = new VSClient(host, port);
+ client.connect();
+
+ // This loop fails with OutOfMemoryError
+ for (int i = 0; i < ITERATIONS; i++) {
+ VSChannel s = client.openChannel();
+ s.close(false);
+
+ if (waitForFreeMem && (Runtime.getRuntime().freeMemory() < Runtime.getRuntime().totalMemory() / 5)) {
+ // Wait if we below 20% of free memory
+ System.gc();
+ Thread.sleep(200);
+ } else {
+ Thread.yield();
+ }
+ }
+ long usedMemory1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
+
+ System.gc();
+ Thread.sleep(3000);
+
+ long usedMemory2 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
+ System.out.println("Used memory before gc: " + usedMemory1);
+ System.out.println("Used memory after gc: " + usedMemory2);
+
+ client.close();
+
+ if (usedMemory2 > ITERATIONS * PAYLOAD_SIZE) {
+ throw new RuntimeException("Memory leak detected");
+ }
+ }
+}
diff --git a/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocketEcho.java b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocketEcho.java
index 60da33bb..60bf03cd 100644
--- a/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocketEcho.java
+++ b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocketEcho.java
@@ -23,7 +23,7 @@
import java.io.*;
public class Test_VSocketEcho {
- public static void main (String args []) throws Throwable {
+ public static void main (String[] args) throws Throwable {
int port = SocketTestUtilities.parsePort(args);
VSServer server = TestVServerSocketFactory.createEchoVServer(port);
diff --git a/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocket_Correctness.java b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocket_Correctness.java
new file mode 100644
index 00000000..16952ec6
--- /dev/null
+++ b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/Test_VSocket_Correctness.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2024 EPAM Systems, Inc
+ *
+ * See the NOTICE file distributed with this work for additional information
+ * regarding copyright ownership. Licensed 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 com.epam.deltix.util.vsocket;
+
+import com.epam.deltix.util.io.BasicIOUtil;
+import com.epam.deltix.util.vsocket.util.SocketTestUtilities;
+import com.epam.deltix.util.vsocket.util.TestVServerSocketFactory;
+import org.junit.Test;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Random;
+
+/**
+ * Sends random data to an echo server and checks that correct data is sent back.
+ */
+@SuppressWarnings("SameParameterValue")
+public class Test_VSocket_Correctness {
+
+ public static final int WRITE_COUNT = 1_000_000;
+
+ @Test
+ public void testSocket() throws Throwable {
+ Test_VSocket_Correctness.main(new String[0]);
+ }
+
+ public static void main(String[] args) throws Throwable {
+ int port = SocketTestUtilities.parsePort(args);
+
+ VSServer server = TestVServerSocketFactory.createBinaryEchoVServer(port);
+ server.setDaemon(true);
+ server.start();
+ System.out.println("Server started on " + server.getLocalPort());
+
+ try {
+ client("localhost", server.getLocalPort());
+ } finally {
+ server.close();
+ }
+ }
+
+ private static void client(String host, int port) {
+ try (VSClient client = new VSClient(host, port)) {
+ client.connect();
+
+ assertEchoClientCorrectness(client, WRITE_COUNT);
+ } catch (IOException x) {
+ throw new UncheckedIOException(x);
+ }
+ }
+
+ static void assertEchoClientCorrectness(VSClient client, int writeCount) {
+ Random rngSrc = new Random(0);
+ Random rngDst = new Random(0);
+
+ // Fills data with values 1..127
+ byte[] data = makeTestData();
+
+ try (VSChannel channel = client.openChannel()) {
+ DataOutputStream os = channel.getDataOutputStream();
+
+ // Data writer thread
+ new Thread(() -> {
+ Thread.currentThread().setName("PRODUCER");
+ sendData(rngSrc, os, data, writeCount);
+ }).start();
+
+ // Reader
+ readData(channel, data, rngDst, writeCount);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static void sendData(Random rngSrc, DataOutputStream os, byte[] data, int writeCount) {
+ long totalSent = 0;
+ try {
+ for (int i = 0; i < writeCount; i++) {
+ int size = generateNextSize(rngSrc);
+ //rngSrc.nextBytes(data);
+ os.write(data, 0, size);
+ totalSent += size;
+ os.write(Byte.MIN_VALUE);
+ totalSent += 1;
+ }
+ os.flush();
+ System.out.println("Producer finished to send data. Sent: " + totalSent + " bytes");
+ } catch (IOException x) {
+ throw new UncheckedIOException(x);
+ }
+ }
+
+ private static void readData(VSChannel channel, byte[] expected, Random rngDst, int readCount) throws IOException {
+ long totalRead = 0;
+ DataInputStream is = channel.getDataInputStream();
+ byte[] actual = new byte[8 * 1024];
+ for (int i = 0; i < readCount; i++) {
+ int size = generateNextSize(rngDst);
+ //rngDst.nextBytes(expected);
+ BasicIOUtil.readFully(is, actual, 0, size);
+
+ int mismatch = Arrays.mismatch(expected, 0, size, actual, 0, size);
+ if (mismatch >= 0) {
+ throw new AssertionError("Data mismatch at position " + (totalRead + mismatch));
+ }
+ totalRead += size;
+
+ int single = is.read();
+ if (single < 0) {
+ throw new AssertionError("Unexpected end of stream");
+ }
+ if (single - 256 != Byte.MIN_VALUE) {
+ throw new AssertionError("Unexpected value " + single + " at position " + totalRead);
+ }
+ totalRead += 1;
+ }
+ System.out.println("Consumer finished to read data. Read: " + totalRead + " bytes");
+ }
+
+ private static int generateNextSize(Random rngSrc) {
+ return 10 + rngSrc.nextInt(1000);
+ }
+
+ private static byte[] makeTestData() {
+ byte[] data = new byte[8 * 1024];
+ byte val = 0;
+ for (int ii = 0; ii < data.length; ii++) {
+ val++;
+ data[ii] = val;
+ if (val == Byte.MAX_VALUE) {
+ val = 0;
+ }
+ }
+ return data;
+ }
+}
diff --git a/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/VSDispatcherTest.java b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/VSDispatcherTest.java
index 089c9ed8..62393536 100644
--- a/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/VSDispatcherTest.java
+++ b/java/timebase/api/src/test/java/com/epam/deltix/util/vsocket/VSDispatcherTest.java
@@ -33,7 +33,7 @@
*/
public class VSDispatcherTest {
- @Test (timeout = 5_000) // Note: test timeout must be greater than reconnectInterval + 2000ms
+ @Test (timeout = 10_000) // Note: test timeout must be greater than reconnectInterval + 2000ms
public void testNoHangsOnConcurrentDisconnects() throws IOException, InterruptedException {
int reconnectInterval = 2000;
@@ -45,16 +45,16 @@ void onDisconnected() {
}
@Override
- void onReconnected() {
+ void onConnected() {
}
@Override
- boolean onTransportStopped(VSocketRecoveryInfo recoveryInfo) {
+ boolean onTransportRecoveryStart(VSocketRecoveryInfo recoveryInfo) {
return false;
}
@Override
- boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) {
+ boolean onTransportRecoveryStop(VSocketRecoveryInfo recoveryInfo) {
return true;
}
});
@@ -115,14 +115,22 @@ boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) {
out.write(buffer, 0, buffer.length);
}
- assertTrue(dispatcher.hasAvailableTransport());
+ assertTrue(dispatcher.isConnectedOrReconnecting());
System.out.println("Emulating broken transports...");
startBarrier.countDown();
- Thread.sleep(100); // Let threads get into blocked state
+
+ // Let threads get into waiting for recovery state,
+ // but maximum of 1sec, which is half of the waiting time for recovery
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(100);
+ if (errorThreads.stream().allMatch(thread -> thread.getState() == Thread.State.TIMED_WAITING)) {
+ break;
+ }
+ }
// Now no transports should be available
- assertFalse(dispatcher.hasAvailableTransport());
+ assertFalse(dispatcher.isConnectedAndNotReconnecting());
// Emulate Flusher thread
VSChannelImpl vsChannel = channels.get(0);
@@ -140,5 +148,8 @@ boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) {
for (Thread thread : errorThreads) {
thread.join();
}
+
+ Test_ClientReconnect.waitUntil(5000, "Dispatcher did not reach DISCONNECTED state in time",
+ () -> dispatcher.getInternalState() == VSDispatcherState.DISCONNECTED);
}
-}
\ No newline at end of file
+}
diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/SafeDisconnectableEventHandler.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/SafeDisconnectableEventHandler.java
new file mode 100644
index 00000000..40937753
--- /dev/null
+++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/SafeDisconnectableEventHandler.java
@@ -0,0 +1,66 @@
+package com.epam.deltix.qsrv.hf.tickdb.comm.client;
+
+import com.epam.deltix.gflog.api.Log;
+import com.epam.deltix.gflog.api.LogFactory;
+import com.epam.deltix.qsrv.hf.spi.conn.DisconnectEventListener;
+import com.epam.deltix.qsrv.hf.spi.conn.Disconnectable;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Same as {@link com.epam.deltix.qsrv.hf.spi.conn.DisconnectableEventHandler} but catches and logs exceptions from listeners
+ * instead of propagating them.
+ *
+ * Helps implement the {@link Disconnectable} interface.
+ * Doesn't maintain a connection status, so
+ * The {@code TickDBClient} takes ownership of the provided {@code Oauth2Client}
+ * and will automatically close it when the TimeBase connection is closed.
+ *
+ * This method must be called before opening the connection.
+ *
+ * @param oauth2Client the OAuth2 client to associate with this connection
+ */
public void setOauth2Client(Oauth2Client oauth2Client) {
- this.userPrincipalResolver.setOauth2Client(oauth2Client);
+ this.setOauth2Client(oauth2Client, false);
}
/**
- * Sets user access token to login to the Timebase server when OAUTH type of authentication defined on server.
+ * Assigns an {@link Oauth2Client} to the current TimeBase connection.
+ *
+ * If {@code external} is {@code false}, the {@code TickDBClient} takes ownership
+ * of the provided client and will automatically close it when the TimeBase
+ * connection is closed.
+ *
+ * If {@code external} is {@code true}, the {@code TickDBClient} will use the
+ * provided client but will not manage its lifecycle. In this case, the
+ * caller is responsible for closing the {@code Oauth2Client}.
+ *
+ * This method must be called before opening the connection.
+ *
+ * @param oauth2Client the OAuth2 client to associate with this connection
+ * @param external {@code true} if the client is managed externally and should
+ * not be closed by {@code TickDBClient}; {@code false} if
+ * ownership should be transferred to {@code TickDBClient}
+ */
+ public void setOauth2Client(Oauth2Client oauth2Client, boolean external) {
+ this.userPrincipalResolver.setOauth2Client(oauth2Client, external);
+ }
+
+ @VisibleForTesting
+ UserPrincipalResolver getUserPrincipalResolver() {
+ return userPrincipalResolver;
+ }
+
+ /**
+ * Sets user access token for login into Timebase server when OAUTH type of authentication defined on server.
* @param token Access token
*/
public void setAccessToken(String token) {
@@ -376,7 +431,12 @@ private VSClient getConnectedVSClient() throws IOException {
if (address != null)
connection.setClientAddress(address, idd);
- connection.setNumTransportChannels(isRemoteConnection ? 1 : getConnectionsNumber(isRemoteConnection));
+ // Use explicitly configured number of transport channels or determine it automatically
+ int transportChannels = numTransportChannels != null ?
+ numTransportChannels :
+ (isRemoteConnection ? 1 : getConnectionsNumber(true));
+
+ connection.setNumTransportChannels(transportChannels);
connection.setTimeout(timeout);
connection.setDisconnectedListener(listener);
connection.setSslContext(SSLClientContextProvider.getSSLContext());
@@ -402,9 +462,7 @@ protected VSChannel createChannel(ChannelType type, boolean autoCommit,
int inCapacity;
int outCapacity;
- int defaultCapacity = isRemoteConnection ? DEFAULT_REMOTE_CHANNEL_SIZE : DEFAULT_LOCAL_CHANNEL_SIZE;
-
- int configuredCapacity = channelBufferSize > 0 ? channelBufferSize : defaultCapacity;
+ int configuredCapacity = channelBufferSize > 0 ? channelBufferSize : defaultChannelCapacity;
switch (type) {
case Input:
@@ -507,6 +565,24 @@ public synchronized long getServerStartTime() {
public void open(boolean readOnly) {
if (syncOpen(readOnly))
onReconnected();
+
+ boolean insideOfEventHandler = TickDBClient.insideOfEventHandler.get().isTrue();
+ if (insideOfEventHandler) {
+ LOGGER.warn("open() is called from inside of event handler. This is not intended API usage. "
+ + "This can lead to unexpected deadlocks "
+ + "and revokes guaranties on handler event order execution. "
+ + "Please make sure that TickDBClient.open() is called outside of it's own event handlers. "
+ + "Event handler thread: %s").with(Thread.currentThread().getName());
+ } else {
+ // Wait for existing events to be processed before we return from open() method.
+ // This is necessary to avoid situation when a listener that added after open() method call,
+ // receives events that were generated during the connection recovery process.
+ if (!this.connectionNotifier.waitForSubmittedEvents(eventHandlerWaitTimeout)) {
+ LOGGER.warn("Some of connection listener events were not processed within timeout after open() method call. " +
+ "Please make sure that event handlers are processing events in a timely manner and do not block. " +
+ "Thread for open(): %s").with(Thread.currentThread().getName());
+ }
+ }
}
private synchronized boolean syncOpen (boolean readOnly) {
@@ -1120,6 +1196,14 @@ public boolean isReadOnly () {
}
public void close () {
+ boolean insideOfEventHandler = TickDBClient.insideOfEventHandler.get().isTrue();
+ if (insideOfEventHandler) {
+ LOGGER.warn("close() is called from inside of event handler. This is not intended API usage. "
+ + "This can lead to unexpected deadlocks "
+ + "and revokes guaranties on handler event order execution. "
+ + "Please make sure that TickDBClient.close() is called outside of it's own event handlers. "
+ + "Event handler thread: %s").with(Thread.currentThread().getName());
+ }
connMgr.cancelReconnect();
@@ -1154,17 +1238,63 @@ public void close () {
isOpen = false;
}
+ int eventId = -1;
+ synchronized (this) {
+ boolean wasConnected = connMgr.isConnected();
+ if (wasConnected) {
+ connMgr.disconnected();
+ // Warning: here we submit a task for QuickExecutor
+ eventId = connectionNotifier.addDisconnectEvent();
+ }
+ }
+
+ if (eventId >= 0) {
+ // Wait for disconnect event to be processed, by event handlers,
+ // so we can guarantee that any handler installed by client before close()
+ // will be executed before close() returns.
+
+ // However if we are inside of event handler right now,
+ // then the client code called close() from the handler,
+ // so waiting for event to be processed is pointless (and will cause a deadlock)
+ if (!insideOfEventHandler) {
+ if (!connectionNotifier.waitForEventProcessing(eventId, eventHandlerWaitTimeout)) {
+ LOGGER.warn("Some of connection listener events were not processed within timeout after close() method call. " +
+ "Please make sure that event handlers are processing events in a timely manner and do not block. " +
+ "Thread for close(): %s").with(Thread.currentThread().getName());
+ }
+ }
+ }
+
+ userPrincipalResolver.close();
+
+ boolean asyncExecutorShutdown = false;
+ QuickExecutor quickExecutor = contextContainer.getQuickExecutor();
+
// shutdown QuickExecutor only if 'open'
if (shutdown) {
- contextContainer.getQuickExecutor().shutdownInstance();
+ // This have to be done after notifying listeners, as we use QE to execute callbacks
+
+ if (insideOfEventHandler) {
+ // Warning: if we are inside of event handler right now,
+ // then this call will basically mean interruption of THIS thread.
+ // So to avoid the deadlock and waiting for executor to interrupt this thread,
+ // we delegate the shutdown call to another thread,
+ // so current thead can be able to be stopped properly by the executor.
+ asyncExecutorShutdown = true;
+ } else {
+ quickExecutor.shutdownInstance();
+ }
+
// We can be already stopped due to a connection loss
aeronContext.stopIfStarted();
}
- if (connMgr.isConnected())
- connMgr.disconnected();
-
userPrincipalResolver.close();
+
+ if (asyncExecutorShutdown) {
+ LOGGER.warn("Shutting down QuickExecutor from inside of event handler unsing separate thread");
+ new Thread(quickExecutor::shutdownInstance, "TickDBClient-QE-Shutdown").start();
+ }
}
public File[] getDbDirs() {
@@ -1259,7 +1389,7 @@ public synchronized MetaData getMetaData () {
// DisconnectableImpl.Reconnector impl.
@Override
- public boolean tryReconnect(int numAttempts, long timeSinceDisconnected, ReconnectableImpl helper) throws Exception {
+ public boolean tryReconnect(int numAttempts, long timeSinceDisconnected, TickDBReconnectableImpl helper) {
if (isOpen)
open(isReadOnly);
@@ -1267,16 +1397,22 @@ public boolean tryReconnect(int numAttempts, long timeSinceDisconnected, Reconne
}
// Disconnectable impl.
+ /**
+ * Warning: code that installs this listener is also responsible to remove it eventually
+ * using {@link #removeDisconnectEventListener(DisconnectEventListener)}, otherwise it can cause memory leaks.
+ * Usually this is should be done just before closing the client or immediately after closing it.
+ */
@Override
public void addDisconnectEventListener(DisconnectEventListener listener) {
- connMgr.addDisconnectEventListener(listener);
+ connectionNotifier.disconnectListeners.addDisconnectEventListener(listener);
}
@Override
public void removeDisconnectEventListener(DisconnectEventListener listener) {
- connMgr.removeDisconnectEventListener(listener);
+ connectionNotifier.disconnectListeners.removeDisconnectEventListener(listener);
}
+
@Override
public boolean isSecured() {
assertOpen();
@@ -1298,6 +1434,8 @@ public boolean isConnected() {
}
void onSessionDisconnected() {
+ // TODO: This isConnected() check should be done under lock.
+ // For now it's left as is to find out instability source of Test_Reconnect.
if (connMgr.isConnected()) {
synchronized (this) {
session = new SessionClient(this, serverProtocolVersion);
@@ -1310,27 +1448,36 @@ void onSessionDisconnected() {
}
private void onDisconnected() {
- if (connMgr.isConnected()) {
- connMgr.scheduleReconnect();
- // listeners can actually stop reconnecting using "close"
- connMgr.disconnected();
+ synchronized (this) {
+ boolean wasConnected = connMgr.isConnected();
- // closing session
- synchronized (this) {
- session = Util.close(session);
+ if (wasConnected) {
+ connMgr.scheduleReconnect();
+ // listeners can actually stop reconnecting using "close"
+ connMgr.disconnected();
+
+ // closing session
+ if (session != null) {
+ LOGGER.info("Closing session due to disconnection");
+ }
+ Util.close(session);
+ session = null;
+ connectionNotifier.addDisconnectEvent();
}
}
}
private void onReconnected() {
- if (!connMgr.isConnected()) {
+ synchronized (this) {
+ boolean wasConnected = connMgr.isConnected();
- synchronized (this) {
- if (session == null)
+ if (!wasConnected) {
+ if (session == null) {
session = new SessionClient(this, serverProtocolVersion);
+ }
+ connMgr.connected();
+ connectionNotifier.addConnectEvent();
}
-
- connMgr.connected();
}
}
@@ -1666,6 +1813,28 @@ public TopicDB getTopicDB() {
return this;
}
+ /**
+ * Allows to explicitly set number of transport channels used by the client.
+ * Must be set before connecting to the server.
+ */
+ public void setNumTransportChannels(Integer numTransportChannels) {
+ this.numTransportChannels = numTransportChannels;
+ }
+
+ @VisibleForTesting
+ Integer getNumTransportChannels() {
+ return numTransportChannels;
+ }
+
+ public void setReconnectIntervalAdjuster(ReconnectableImpl.ReconnectIntervalAdjuster adjuster) {
+ connMgr.setAdjuster(adjuster);
+ }
+
+ @VisibleForTesting
+ ReconnectableImpl.ReconnectIntervalAdjuster getReconnectIntervalAdjuster() {
+ return connMgr.getAdjuster();
+ }
+
@Override
public boolean isTopicDBSupported() {
return true;
@@ -1684,4 +1853,110 @@ public Thread newThread(@NotNull Runnable r) {
return thread;
}
}
+
+ private enum ConnectionUpdateEvent {
+ CONNECT,
+ DISCONNECT
+ }
+
+ /**
+ * Enforces sequential processing of connection status change events.
+ * So even if client event handlers are slow/block we still guarantee correct order of events.
+ *
+ * Expected behavior of TickDBClient:
+ *
+ *
+ */
+ private final AtomicReferenceisConnected must be implemented by a client.
+ *
+ */
+ private class ConnectionNotificationTask extends QuickExecutor.QuickTask {
+ private final Queuehelper. After that, the return
+ * value is irrelevant. If unsucessful, this method can either throw
+ * an exception, or return true to reschedule the reconnect,
+ * or, in rare instances, return false to give up.
+ *
+ * @return Whether reconnection should be rescheduled.
+ * @throws Exception
+ * If thrown, reconnect will be rescheduled. Therefore, this
+ * method can freely throw exceptions due to reconnect failure.
+ *
+ */
+ boolean tryReconnect(
+ int numAttempts,
+ long timeSinceDisconnected,
+ TickDBReconnectableImpl helper
+ ) throws Exception;
+ }
+}
diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/UserPrincipalResolver.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/UserPrincipalResolver.java
index 553dde75..701501b8 100644
--- a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/UserPrincipalResolver.java
+++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/UserPrincipalResolver.java
@@ -27,26 +27,37 @@ public class UserPrincipalResolver {
public static final Log LOGGER = LogFactory.getLog("tickdb.client");
private volatile Oauth2Client oauth2Client;
+ private volatile boolean isExternal;
void setOauth2Client(Oauth2Client oauth2Client) {
+ this.setOauth2Client(oauth2Client, false);
+ }
+
+ void setOauth2Client(Oauth2Client oauth2Client, boolean external) {
+ if (this.oauth2Client != null && !isExternal) {
+ throw new RuntimeException("OAuth2 client is not empty.");
+ }
+
this.oauth2Client = oauth2Client;
+ this.isExternal = external;
}
UserPrincipal resolve(UserPrincipal user) {
if (SecretsStorage.isSecretsStorageValue(user.getPass())) {
try {
return new UserPrincipal(
- user.getName(),
- SecretsStorage.INSTANCE.getSecret(user.getPass())
+ user.getName(),
+ SecretsStorage.INSTANCE.getSecret(user.getPass())
);
} catch (Throwable t) {
LOGGER.warn().append("Failed to resolve SecretsStorage value for user ").append(user.getName()).append(t).commit();
}
}
- if (oauth2Client != null) {
- String clientId = oauth2Client.clientId();
- String token = oauth2Client.token();
+ Oauth2Client currentOauth2Client = oauth2Client;
+ if (currentOauth2Client != null) {
+ String clientId = currentOauth2Client.clientId();
+ String token = currentOauth2Client.token();
return new UserPrincipal(clientId, token);
}
@@ -54,8 +65,9 @@ UserPrincipal resolve(UserPrincipal user) {
}
void close() {
- if (oauth2Client != null) {
- oauth2Client.close();
+ Oauth2Client currentOauth2Client = oauth2Client;
+ if (currentOauth2Client != null && !isExternal) {
+ currentOauth2Client.close();
}
}
-}
\ No newline at end of file
+}
diff --git a/java/timebase/commons/src/main/java/com/epam/deltix/util/Version.java b/java/timebase/commons/src/main/java/com/epam/deltix/util/Version.java
index 23084720..4d17e816 100644
--- a/java/timebase/commons/src/main/java/com/epam/deltix/util/Version.java
+++ b/java/timebase/commons/src/main/java/com/epam/deltix/util/Version.java
@@ -17,9 +17,9 @@ public abstract class Version {
public static final int MAJOR = 6;
public static final int MINOR = 2;
public static final String NAME = "18-SNAPSHOT";
- public static final String BUILD = "2c3be72f";
+ public static final String BUILD = "f9c92576";
public static final Integer COMMITS_AFTER_TAG = null;
- public static final String BUILD_DATE = "2026-06-25 21:23:51 +0300";
+ public static final String BUILD_DATE = "2026-07-17 14:24:22 +0300";
public static final String VERSION_STRING;
public static final String MAJOR_VERSION_STRING;
diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/config/TimebaseServiceExecutor.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/config/TimebaseServiceExecutor.java
index 6cd42b1e..40b5fdb5 100644
--- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/config/TimebaseServiceExecutor.java
+++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/config/TimebaseServiceExecutor.java
@@ -183,15 +183,16 @@ public void run() {
String compression = config.getString("compression", VSCompression.AUTO.toString());
int maxConnections = config.getInt("maxConnections", VSServerFramework.MAX_CONNECTIONS);
- short maxSocketsPerConnection = (short) config.getInt("maxSocketsPerConnection", VSServerFramework.MAX_SOCKETS_PER_CONNECTION);
+ short maxChannels = (short) config.getInt("maxChannelsPerConnection", -1);
+ short maxSocketsPerConnection = (short) config.getInt("maxSocketsPerConnection", Runtime.getRuntime().availableProcessors() * 2);
contextContainer.getQuickExecutor().reuseInstance();
VSServerFramework framework = new VSServerFramework(contextContainer.getQuickExecutor(),
(int) interval.toMilliseconds(),
Enum.valueOf(VSCompression.class, compression),
- maxConnections,
- maxSocketsPerConnection, contextContainer, DefaultConnectionAcceptor.INSTANCE);
+ maxConnections, maxSocketsPerConnection, maxChannels,
+ contextContainer, DefaultConnectionAcceptor.INSTANCE);
if (framework.getCompression() == VSCompression.OFF)
LOGGER.info("Timebase communication compression disabled.");
diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/DownloadHandlerFactory.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/DownloadHandlerFactory.java
index 465f56e1..7258b944 100644
--- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/DownloadHandlerFactory.java
+++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/DownloadHandlerFactory.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2024 EPAM Systems, Inc
+ * Copyright 2026 EPAM Systems, Inc
*
* See the NOTICE file distributed with this work for additional information
* regarding copyright ownership. Licensed under the Apache License,
@@ -17,7 +17,6 @@
package com.epam.deltix.qsrv.hf.tickdb.comm.server;
import com.epam.deltix.qsrv.hf.pub.ChannelPerformance;
-import com.epam.deltix.timebase.messages.IdentityKey;
import com.epam.deltix.qsrv.hf.pub.RawMessage;
import com.epam.deltix.qsrv.hf.tickdb.comm.SelectionOptionsCodec;
import com.epam.deltix.qsrv.hf.tickdb.comm.TDBProtocol;
@@ -43,6 +42,8 @@
import java.io.DataOutputStream;
import java.io.IOException;
import java.security.Principal;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import static com.epam.deltix.qsrv.hf.tickdb.comm.TDBProtocol.TRANSPORT_TYPE_AERON;
import static com.epam.deltix.qsrv.hf.tickdb.comm.TDBProtocol.TRANSPORT_TYPE_SOCKET;
@@ -53,9 +54,44 @@
* @author Alexei Osipov
*/
public class DownloadHandlerFactory {
+
+ public static final int CURSORS_PER_CONNECTION = Integer.getInteger("TimeBase.maxCursorsPerConnection", -1);
+
+ private static final ConcurrentHashMap