From d97c221b86c3d145ca046916e4ba3efdb648ece2 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Thu, 14 Aug 2025 20:37:30 +0300 Subject: [PATCH 1/9] [*] VSocket API merge latest changes --- .../deltix/util/vsocket/ChannelExecutor.java | 130 ++++++-- .../util/vsocket/ChannelOutputStream.java | 168 ++++++---- .../deltix/util/vsocket/ClientConnection.java | 2 +- .../epam/deltix/util/vsocket/VSChannel.java | 24 ++ .../deltix/util/vsocket/VSChannelImpl.java | 152 +++++++--- .../epam/deltix/util/vsocket/VSClient.java | 227 ++++++++++---- .../deltix/util/vsocket/VSDispatcher.java | 185 ++++++++--- .../util/vsocket/VSDispatcherState.java | 7 + .../deltix/util/vsocket/VSOutputStream.java | 5 +- .../util/vsocket/VSServerFramework.java | 44 +-- .../util/vsocket/VSTransportChannel.java | 17 +- .../deltix/util/vsocket/VSocketFactory.java | 7 +- .../epam/deltix/util/vsocket/VSocketImpl.java | 91 +++++- .../util/vsocket/VSocketOutputStream.java | 6 +- .../transport/DatagramInputStream.java | 3 +- .../transport/DatagramOutputStream.java | 4 +- .../vsocket/transport/SocketConnection.java | 5 +- .../deltix/util/vsocket/VSDispatcherTest.java | 5 +- .../hf/tickdb/comm/client/TickDBClient.java | 286 +++++++++--------- .../epam/deltix/qsrv/hf/tickdb/TDBRunner.java | 12 +- .../test/qsrv/hf/tickdb/Test_SSLTomcat.java | 42 ++- .../qsrv/hf/tickdb/Test_TomcatServer.java | 21 -- 22 files changed, 976 insertions(+), 467 deletions(-) create mode 100644 java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcherState.java diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelExecutor.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelExecutor.java index a44e8c74..217be99c 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelExecutor.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelExecutor.java @@ -29,6 +29,9 @@ import java.util.logging.Level; class ChannelExecutor implements Runnable { + //@ApiStatus.Experimental // Temporary option for testing performance effect of using yield on Windows + private static final boolean USE_YIELD_ON_WINDOWS = Boolean.getBoolean("TimeBase.network.executor.windows.useYield"); + private static volatile ChannelExecutor INSTANCE; private static ChannelExecutor createInstance(AffinityConfig affinityConfig) { @@ -37,6 +40,12 @@ private static ChannelExecutor createInstance(AffinityConfig affinityConfig) { return executor; } + @SuppressWarnings("SameParameterValue") + //@VisibleForTesting + static ChannelExecutor createNonSharedTestInstance(AffinityConfig affinityConfig) { + return createInstance(affinityConfig); + } + public static ChannelExecutor getInstance(AffinityConfig affinityConfig) { // "Double checked lock" (via volatile) if (INSTANCE == null) { @@ -51,11 +60,11 @@ public static ChannelExecutor getInstance(AffinityConfig affinityConfig) { private final QuickList channels = new QuickList<>(); private boolean stopped = false; - private final CPUEater cpuEater; + private final CPUEater cpuEater; // Used only for Windows private final int idleTime; private final Thread thread; - private static ChannelExecutor create(AffinityConfig affinityConfig) { + static ChannelExecutor create(AffinityConfig affinityConfig) { ThreadFactory factory = new AffinityThreadFactoryBuilder(affinityConfig) .setNameFormat("ChannelExecutor Thread") .setDaemon(true) @@ -66,7 +75,7 @@ private static ChannelExecutor create(AffinityConfig affinityConfig) { private ChannelExecutor(ThreadFactory factory) { idleTime = VSProtocol.getIdleTime(); - cpuEater = new CPUEater(idleTime); + cpuEater = Util.IS_WINDOWS_OS ? new CPUEater(idleTime) : null; this.thread = factory.newThread(this); } @@ -81,6 +90,11 @@ public void shutdown () { } public void addChannel(VSChannel channel) { + assert channel != null; + + if (channel == null) + return; + synchronized (channels) { channels.linkLast(new Entry(channel)); } @@ -88,58 +102,112 @@ public void addChannel(VSChannel channel) { wakeup(); } + public void removeChannel(VSChannel channel) { + assert channel != null; + + + synchronized (channels) { + Entry entry = channels.getFirst(); + + while (entry != null) { + if (entry.channel.equals(channel)) { + remove(entry); + return; + } else { + entry = entry.next(); + } + } + } + + wakeup(); + } + @Override public void run() { assert Thread.currentThread() == this.thread; while (!stopped) { Entry entry; + boolean isEmpty; + long bytesSent = 0; synchronized (channels) { entry = channels.getFirst(); - } - - if (entry == null) { - LockSupport.park(); - - if (Thread.interrupted ()) { - if (stopped) - break; - } - } + isEmpty = entry == null; - synchronized (channels) { - entry = channels.getFirst(); while (entry != null) { - VSChannel channel = entry.channel; try { - if (channel != null && channel.getNoDelay() && channel.getState() == VSChannelState.Connected) { - VSOutputStream out = channel.getOutputStream(); - out.flushAvailable(); - - entry = entry.next(); - } else if (channel != null) { - if (channel.getState() == VSChannelState.Removed || channel.getState() == VSChannelState.Closed) + switch (channel.getState()) { + case Connected: { + if (channel.getNoDelay()) { + // Flush + VSOutputStream out = channel.getOutputStream(); + // We do not want to send all at once, we will, re-try send shortly + bytesSent += out.flushAvailable(false); + } + break; + } + case Removed: + case Closed: { entry = remove(entry); + continue; + } } } catch (ChannelClosedException e) { // ignore entry = remove(entry); + continue; } catch (IOException e) { VSProtocol.LOGGER.log (Level.WARNING, "Exception while flushing data", e); } + + // Move to the next channel + entry = entry.next(); + } + } + + if (isEmpty) { + // No channels to process => Wait for channels to be added. + LockSupport.park(); + + if (Thread.interrupted ()) { + if (stopped) { + break; + } + } + } else { + // Do not wait if we have sent any data. + // It's very likely that it's time to send more because the sending time is relatively long. + if (bytesSent == 0) { + // Wait till next time to flush channels + idleWait(); } } + } + } - if (!Util.IS_WINDOWS_OS) { - LockSupport.parkNanos (idleTime); + private void idleWait() { + if (!Util.IS_WINDOWS_OS) { + if (USE_YIELD_ON_WINDOWS) { + waitWithYield(idleTime); } else { - if (TimeKeeper.getMode() == TimeKeeper.Mode.HIGH_RESOLUTION_SYNC_BACK) - TimeKeeper.parkNanos(idleTime); - else - cpuEater.run(); + LockSupport.parkNanos(idleTime); } + } else { + if (TimeKeeper.getMode() == TimeKeeper.Mode.HIGH_RESOLUTION_SYNC_BACK) { + TimeKeeper.parkNanos(idleTime); + } else { + cpuEater.run(); + } + } + } + + private static void waitWithYield(int idleTime) { + long start = System.nanoTime(); + long end = start + idleTime; + while (System.nanoTime() < end) { + Thread.yield(); } } @@ -150,7 +218,7 @@ private Entry remove(Entry entry) { } private static class Entry extends QuickList.Entry { - VSChannel channel; + final VSChannel channel; private Entry(VSChannel channel) { this.channel = channel; @@ -162,7 +230,7 @@ private static class CPUEater { private final long cycles; private final MemoryDataOutput out = new MemoryDataOutput(); - private final double value = 345.56787899; + private static final double value = 345.56787899; private CPUEater(long nanos) { this.avgCostOfNanoTimeCall = nanoTimeCost(); diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelOutputStream.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelOutputStream.java index f82384d6..6da705ba 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelOutputStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ChannelOutputStream.java @@ -19,6 +19,7 @@ import com.epam.deltix.util.concurrent.UncheckedInterruptedException; import com.epam.deltix.util.lang.Util; import net.jcip.annotations.GuardedBy; +import org.jetbrains.annotations.NotNull; import java.io.IOException; import java.util.concurrent.atomic.AtomicInteger; @@ -27,19 +28,30 @@ * Date: Mar 25, 2010 */ public class ChannelOutputStream extends VSOutputStream { + //@ApiStatus.Experimental // Temporary option for testing performance effect of flushing single packet + private static final boolean SINGLE_SEND_ON_PARTIAL_FLUSH = Boolean.getBoolean("TimeBase.network.channel.singleSendOnPartialFlush"); + private final int maxCapacity; private final VSChannelImpl channel; + @GuardedBy("this") private boolean closed = false; + @GuardedBy("this") private byte [] buffer; + @GuardedBy("this") private boolean flushDisabled = false; + // Total number of accumulated bytes in the buffer. @GuardedBy("this") private int size = 0; + // Number of bytes available to be sent. + // Always: available <= size. + // Can be less than size if flush is disabled. @GuardedBy("this") private int available = 0; private final AtomicInteger waiting = new AtomicInteger(0); + // TODO: Consider making it volatile instead. All writes are under synchronization private final AtomicInteger remoteCapacityAvailable = new AtomicInteger(-1); private final AtomicInteger capacityIncrement = new AtomicInteger(0); @@ -58,25 +70,32 @@ public synchronized void close() throws IOException { } } - private void doNotify() { - if (waiting.get() > 1) { - notifyAll(); - } else { - notify(); - } - } - @Override public synchronized void enableFlushing() throws IOException { flushDisabled = false; available = size; - if (size >= maxCapacity) + // Previously this check looked like this: "size >= maxCapacity" + // However this is ineffective: the remote capacity is "maxCapacity" at most, + // So attempt to flush more than that almost certainly results in situation + // when we will block on that flush and need to wait for BYTES_AVAILABLE_REPORT from the remote side. + // At the same time we do not want to flush too often (it's costly), + // so we flush only when we have at least half of the buffer filled. + int halfCapacity = maxCapacity >> 1; + if (size >= halfCapacity) { + // If we below of 75% capacity, we can flush buffer partially. + // However, if we are above 75% capacity, we should flush all data + // and block till all accumulated data is sent. + // Otherwise, if the consumer too slow, the buffer will start to grow indefinitely. + // See https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/-/issues/1298 + int buffer75percent = halfCapacity + (halfCapacity >> 1); + boolean partialOk = size < buffer75percent; try { - flushInternal (false); + flushInternal(partialOk, false); } catch (InterruptedException e) { - throw new UncheckedInterruptedException (e); + throw new UncheckedInterruptedException(e); } + } } @Override @@ -95,47 +114,73 @@ synchronized void closeNoFlush() { @Override public synchronized void flush() throws IOException { try { - flushInternal (false); + flushInternal (false, false); } catch (InterruptedException e) { - throw new UncheckedInterruptedException (e); + Thread.currentThread().interrupt(); + throw new UncheckedInterruptedException(e); } } @Override - public synchronized void flushAvailable() throws IOException { + public synchronized int flushAvailable(boolean flushAll) throws IOException { try { // we do not want to wait() here - if (available > 0 && getRemoteCapacity() > VSProtocol.MINSIZE) - flushInternal (true); + + // TODO: Consider adding "capacityIncrement.get()" here + if (available > 0 && getRemoteCapacity() > VSProtocol.MINSIZE) { + boolean stopAfterSingleSend = SINGLE_SEND_ON_PARTIAL_FLUSH && !flushAll; + return flushInternal(true, stopAfterSingleSend); + } else { + return 0; + } } catch (InterruptedException e) { - throw new UncheckedInterruptedException (e); + Thread.currentThread().interrupt(); + throw new UncheckedInterruptedException(e); } } - private void checkCapacity() { - remoteCapacityAvailable.addAndGet(capacityIncrement.getAndSet(0)); + // Must be called by the thread that performs flush under lock + @GuardedBy("this") + private int checkCapacity() { + return remoteCapacityAvailable.addAndGet(capacityIncrement.getAndSet(0)); } private int getRemoteCapacity() { return remoteCapacityAvailable.get(); } - private void flushInternal (boolean partialOk) throws IOException, InterruptedException { + @GuardedBy("this") + private int flushInternal (boolean partialOk, boolean stopAfterSingleSend) throws IOException, InterruptedException { + assert partialOk || !stopAfterSingleSend : "stopAfterSingleSend is only allowed with partialOk==true"; + + // Currently we try to send all "available" bytes at once. + // This may be not the best idea with partialOk==true, because this means that this way we may prevent + // loader thread from adding new data to the buffer and blocking him for a long time. + // At the same time it possible that we will get into a blocking send because socket buffer + // is full. + // TODO: Consider sending only once if partialOk==true. Additionally the flush may return number of bytes sent + // so the ChannelExecutor can decide if it should sleep or not. Because if we sent at least some data + // then it's very likely that we spend on this more time, than the ChannelExecutor sleeps. + + int sent = 0; for (;;) { + int remoteCapacity; for (;;) { - checkCapacity(); + remoteCapacity = checkCapacity(); if (available == 0) - return; + return sent; if (closed) throw new ChannelClosedException(); - if (getRemoteCapacity() >= VSProtocol.MINSIZE) + if (remoteCapacity >= VSProtocol.MINSIZE) { + // The only way to proceed to code after the loop break; + } if (partialOk) - return; + return sent; // "out" could be asynchronously flushed while this thread // is in wait (). Therefore, we have to query the state of @@ -147,28 +192,39 @@ private void flushInternal (boolean partialOk) throws IOException waiting.decrementAndGet(); } - int packetSize = available; - - if (packetSize > getRemoteCapacity()) - packetSize = getRemoteCapacity(); - - if (packetSize > VSProtocol.MAXSIZE) - packetSize = VSProtocol.MAXSIZE; + // No need to get new value of getRemoteCapacity() here, we just got it in the loop above + int packetSize = Math.min(Math.min(available, remoteCapacity), VSProtocol.MAXSIZE); channel.send (buffer, 0, packetSize); - checkCapacity(); remoteCapacityAvailable.addAndGet(-packetSize); size -= packetSize; available -= packetSize; + sent += packetSize; assert size >= 0; - System.arraycopy (buffer, packetSize, buffer, 0, size); + if (size > 0) { + // Shift remaining data to the buffer start. + // Slow on big buffer sizes! + // TODO: Implement cyclic buffer instead + System.arraycopy(buffer, packetSize, buffer, 0, size); + + if (stopAfterSingleSend) { + // We stop after very first send to allow loader thread to add new data. + + // Move capacityIncrement to remoteCapacityAvailable, + // so threads that calls flushAvailable() or addAvailableCapacity() can see updated value. + checkCapacity(); + + return sent; + } + } } } + @GuardedBy("this") private void ensureCapacity (int c) { int cap = buffer.length; @@ -182,7 +238,7 @@ private void ensureCapacity (int c) { } @Override - public synchronized void write (byte [] b, int off, int len) + public synchronized void write (byte @NotNull [] b, int off, int len) throws IOException { if (closed) @@ -201,7 +257,8 @@ else if (newSize <= buffer.length) { } else { try { - flushInternal (false); + // TODO: We do not necessarily need full flush here. We need to get "len" bytes of free space + flushInternal (false, false); assert size == 0; @@ -226,7 +283,7 @@ public synchronized void write(int b) throws IOException { if (flushDisabled) ensureCapacity (size + 1); else if (size >= maxCapacity) - flushInternal (false); + flushInternal (false, false); if (!flushDisabled) available++; @@ -242,47 +299,56 @@ public synchronized void setRemoteCapacity(int capacity) { notify(); } - private synchronized void addCapacity(int value) { - remoteCapacityAvailable.addAndGet(value); - notify(); + private void wakeAfterCapacityAdded() { + synchronized (this) { + // Update remote capacity, so next "addAvailableCapacity" will not need to block + checkCapacity(); + // Wakeup waiting thread + notify(); + } } public void addAvailableCapacity(int capacity) { - if (getRemoteCapacity() < VSProtocol.MINSIZE) - addCapacity(capacity); - else - capacityIncrement.addAndGet(capacity); + capacityIncrement.addAndGet(capacity); + if (getRemoteCapacity() < VSProtocol.MINSIZE) { + wakeAfterCapacityAdded(); + } } - private int send (byte[] data, int offset, int length) + /** + * Sends data immediately, without putting it into the buffer. + * Used when the data does not fit into buffer. + */ + @GuardedBy("this") + private int send (byte @NotNull [] data, int offset, int length) throws IOException { int bytes = 0; try { while (length > 0) { + int remoteCapacity; for (;;) { - checkCapacity(); + remoteCapacity = checkCapacity(); if (closed) throw new ChannelClosedException(); - if (getRemoteCapacity() >= VSProtocol.MINSIZE) + if (remoteCapacity >= VSProtocol.MINSIZE) { + // The only way to proceed to code after the loop break; + } //waiting.incrementAndGet(); wait (); //waiting.decrementAndGet(); } - int packetSize = Math.min(length, getRemoteCapacity()); - - if (packetSize > VSProtocol.MAXSIZE) - packetSize = VSProtocol.MAXSIZE; + // No need to get new value of getRemoteCapacity() here, we just got it in the loop above + int packetSize = Math.min(Math.min(length, remoteCapacity), VSProtocol.MAXSIZE); channel.send(data, offset, packetSize); - checkCapacity(); remoteCapacityAvailable.addAndGet(-packetSize); length -= packetSize; diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ClientConnection.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ClientConnection.java index 4253b68d..55b56146 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ClientConnection.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/ClientConnection.java @@ -37,7 +37,7 @@ public ClientConnection(Socket socket) throws IOException { this.socket = socket; this.in = socket.getInputStream(); this.os = socket.getOutputStream(); - this.bin = new BufferedInputStream(this.in); + this.bin = new BufferedInputStream(this.in, VSocketImpl.INPUT_STREAM_BUFFER_SIZE); } public Socket getSocket() { diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java index d4cc692c..9d5352af 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java @@ -17,6 +17,8 @@ package com.epam.deltix.util.vsocket; import com.epam.deltix.util.lang.Disposable; +import com.epam.deltix.util.lang.DisposableListener; +import org.jetbrains.annotations.Nullable; import java.io.*; @@ -31,6 +33,8 @@ public interface VSChannel extends Disposable { public String getRemoteAddress(); + public String getClientAddress(); + public String getRemoteApplication(); public String getClientId(); @@ -62,4 +66,24 @@ public interface VSChannel extends Disposable { public String encode(String value); public String decode(String value); + + void addDisposableListener(DisposableListener listener); + + void removeDisposableListener(DisposableListener listener); + + + /** + * @return value previously set by {@link #setTag(String)} + * + * @apiNote experimental + */ + @Nullable + String getTag(); + + /** + * Sets an arbitrary tag that can be used for debugging purposes. It is not sent to the remote side. + * + * @apiNote experimental + */ + void setTag(@Nullable String tag); } \ No newline at end of file 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 542f79be..eb89ef03 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 @@ -20,9 +20,12 @@ import com.epam.deltix.util.concurrent.QuickExecutor; import static com.epam.deltix.util.vsocket.VSProtocol.*; import com.epam.deltix.util.io.*; +import com.epam.deltix.util.lang.DisposableListener; import com.epam.deltix.util.lang.Util; import com.epam.deltix.util.memory.DataExchangeUtils; import com.epam.deltix.util.memory.MemoryDataOutput; +import net.jcip.annotations.GuardedBy; +import org.jetbrains.annotations.Nullable; import javax.annotation.CheckReturnValue; import java.io.*; @@ -36,6 +39,9 @@ * */ final class VSChannelImpl implements VSChannel { + //@ApiStatus.Experimental // Configurable option for testing optimal value of notifyThreshold in different setups + private static final int MAX_NOTIFY_THRESHOLD = Integer.getInteger("TimeBase.network.channel.maxNotifyThreshold", VSProtocol.MAXSIZE); + private final ContextContainer contextContainer; private volatile VSChannelState state = VSChannelState.NotConnected; @@ -47,11 +53,11 @@ final class VSChannelImpl implements VSChannel { private final int inCapacity; // = 1 << 15; private final int outCapacity; // = 1 << 14; - private GapQueueInputStream in; - private CountingInputStream cin; + private final GapQueueInputStream in; + private final CountingInputStream cin; private DataInputStream din; - private ChannelOutputStream out; + private final ChannelOutputStream out; private DataOutputStream dout; private boolean autoFlush = false; @@ -67,15 +73,25 @@ final class VSChannelImpl implements VSChannel { private final boolean compressed; + @GuardedBy("inflater") private final Inflater inflater; + @GuardedBy("inflater") private final MemoryDataOutput infOut; + // Previously was protected by "deflater" itself however this was redundant as we always sync on "this" + @GuardedBy("this") private final Deflater deflater; + @GuardedBy("this") private final MemoryDataOutput defOut; private volatile long numBytesSend; // synchronized by "this" private final Counter numBytesRead = new Counter(); + private String tag; // Arbitrary tag for debugging purposes. It is not sent to the remote side. + + @GuardedBy("listeners") + private final HashSet> listeners = new HashSet<>(); + // private final StringBuffer sendLog = new StringBuffer(); // private final StringBuffer recievedLog = new StringBuffer(); @@ -165,7 +181,10 @@ public VSChannelImpl (VSDispatcher dispatcher, int inCapacity, int outCapacity, this.out = new ChannelOutputStream(this, outCapacity); this.in = new GapQueueInputStream (inCapacity); - this.cin = new CountingInputStream(this.in, inCapacity / 4) { + // In case of big buffer we want to notify sender as soon as a full packet can be sent + // or 1/4 of max capacity accumulated + int notifyThreshold = Math.min(inCapacity / 4, MAX_NOTIFY_THRESHOLD); + this.cin = new CountingInputStream(this.in, notifyThreshold) { @Override protected boolean bytesRead(long change) { @@ -206,6 +225,11 @@ public String getRemoteAddress() { return dispatcher != null ? dispatcher.getRemoteAddress() : null; } + @Override + public String getClientAddress() { + return dispatcher != null ? dispatcher.getClientAddress() : null; + } + public String getRemoteApplication() { return dispatcher != null ? dispatcher.getApplicationID() : null; } @@ -238,15 +262,15 @@ public DataOutputStream getDataOutputStream() { return (dout); } - public synchronized VSChannelState getState() { + public VSChannelState getState() { return state; } - public synchronized boolean isClosed () { - return state == VSChannelState.Closed || state == VSChannelState.Removed; - } +// public boolean isClosed () { +// return state == VSChannelState.Closed || state == VSChannelState.Removed; +// } - private synchronized boolean isRemoteConnected() { + private boolean isRemoteConnected() { return state == VSChannelState.Connected; } @@ -298,6 +322,8 @@ public void close (boolean terminate) { state = VSChannelState.Closed; } + + notifyListeners(); } public void processCommand(int cmd, long position) { @@ -352,6 +378,8 @@ void onRemoteClosed() { synchronized (this) { state = VSChannelState.Removed; } + + notifyListeners(); } void onRemoteClosing() { @@ -374,6 +402,8 @@ void onRemoteClosing() { break; } } + + notifyListeners(); } void onDisconnected(IOException error) { @@ -396,6 +426,9 @@ void onDisconnected(IOException error) { // notify availability listener after input close notifyDataAvailable(); + + // notify disposable listeners + notifyListeners(); } // void onRemoteClosing() { @@ -482,6 +515,11 @@ void receive(long position, byte[] data, int offset, int leng if (available) { notifyDataAvailable(); + } + // It's necessary to still process commands when "in" gets closed to properly complete + // the channel closing process. Otherwise, commands may stuck in queue and the channel may leak. + // See https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/-/issues/1264 + if (available || in.isClosed()) { checkCommands(); } } @@ -494,24 +532,26 @@ private void notifyDataAvailable() { void sendBytesRead (long bytes) throws InterruptedException, IOException { - VSChannelState state = this.state; + VSChannelState state = getState(); + if (state != VSChannelState.Connected && state != VSChannelState.RemoteClosed) { - if (LOGGER.isLoggable(Level.FINE)) { + if (LOGGER.isLoggable(Level.FINE)) LOGGER.log(Level.FINE, "Skipping BYTES_AVAILABLE_REPORT report: " + bytes + " because channel state is " + state); - } return; } - if (state == VSChannelState.RemoteClosed && LOGGER.isLoggable(Level.FINE)) { + + state = getState(); + + if (state == VSChannelState.RemoteClosed && LOGGER.isLoggable(Level.FINE)) LOGGER.log(Level.FINE, "Sending BYTES_AVAILABLE_REPORT report: " + bytes + " at state " + state + " with remoteIndex=" + remoteIndex); - } DataExchangeUtils.writeInt (buffer8, 4, (int)bytes); DataExchangeUtils.writeInt (buffer8, 8, remoteIndex); - final VSTransportChannel tc = dispatcher.checkOut (); - if (LOGGER.isLoggable(Level.FINEST)) { + VSTransportChannel tc = dispatcher.checkOut (); + if (LOGGER.isLoggable(Level.FINEST)) LOGGER.log(Level.FINEST, "Sending BYTES_AVAILABLE_REPORT report: " + ((int)bytes) + " from " + tc.getSocketIdStr()); - } + try { tc.write (buffer8, 0, buffer8.length); } finally { @@ -539,6 +579,7 @@ void sendConnect () } } + @GuardedBy("this") void sendClosing() throws InterruptedException, IOException { @@ -554,15 +595,17 @@ void sendClosing() final VSTransportChannel tc = dispatcher.checkOut (); try { tc.write (data, 0, data.length); + //noinspection NonAtomicOperationOnVolatileField This is safe because writes always happen with lock on "this" numBytesSend += 2; } finally { dispatcher.checkIn (tc); } - if (LOGGER.isLoggable(Level.FINEST)) { + + if (LOGGER.isLoggable(Level.FINEST)) LOGGER.log(Level.FINEST, "Sending CLOSING: remoteIndex=" + remoteIndex + " numBytesSend=" + numBytesSend); - } } + @GuardedBy("this") void sendClosed () throws InterruptedException, IOException { @@ -578,13 +621,14 @@ void sendClosed () final VSTransportChannel tc = dispatcher.checkOut (); try { tc.write (data, 0, data.length); + //noinspection NonAtomicOperationOnVolatileField This is safe because writes always happen with lock on "this" numBytesSend += 2; } finally { dispatcher.checkIn (tc); } - if (LOGGER.isLoggable(Level.FINEST)) { + + if (LOGGER.isLoggable(Level.FINEST)) LOGGER.log(Level.FINEST, "Sending CLOSED: remoteIndex=" + remoteIndex + " numBytesSend=" + numBytesSend); - } } synchronized void send(byte [] data, int offset, int length) @@ -595,20 +639,15 @@ synchronized void send(byte [] data, int offset, int length) VSTransportChannel tc = null; try { + tc = dispatcher.checkOut (); if (compressed) { - tc = dispatcher.checkOut (); - - synchronized (deflater) { - int compressed = compress(data, offset, length); - tc.write(remoteId, remoteIndex, numBytesSend, defOut.getBuffer(), 0, compressed, length); - numBytesSend += length; - } + int compressed = compress(data, offset, length); + tc.write(remoteId, remoteIndex, numBytesSend, defOut.getBuffer(), 0, compressed, length); } else { - tc = dispatcher.checkOut (); tc.write(remoteId, remoteIndex, numBytesSend, data, offset, length, length); - numBytesSend += length; - //sendLog.append("sending bytes: ").append(numBytesSend); } + numBytesSend += length; + //sendLog.append("sending bytes: ").append(numBytesSend); } finally { if (tc != null) @@ -701,15 +740,10 @@ void onRemoteConnected(int remoteId, int remoteCapacity, int DataExchangeUtils.writeUnsignedShort (buffer8, 0, remoteId); DataExchangeUtils.writeUnsignedShort (buffer8, 2, BYTES_AVAILABLE_REPORT); - boolean wasClosed; - - synchronized (this) { - //assert state == VSChannelState.NotConnected; //TODO: check this - - wasClosed = state == VSChannelState.Closed; - state = VSChannelState.Connected; - } + boolean wasClosed = state == VSChannelState.Closed; + //assert state == VSChannelState.NotConnected; //TODO: check this + state = VSChannelState.Connected; out.setRemoteCapacity(remoteCapacity); if (wasClosed) { @@ -780,6 +814,7 @@ public String decode(String value) { return new String(output); } + @GuardedBy("this") private int compress(byte[] data, int offset, int length) { deflater.reset(); @@ -802,6 +837,7 @@ private int compress(byte[] data, int offset, int length) { return count; } + @GuardedBy("inflater") private int decompress(byte[] data, int offset, int length) { inflater.reset(); @@ -829,4 +865,42 @@ public String toString () { int getIndex() { return index; } + + @Override + public void addDisposableListener(DisposableListener listener) { + synchronized (listeners) { + listeners.add(listener); + } + } + + @Override + public void removeDisposableListener(DisposableListener listener) { + synchronized (listeners) { + listeners.remove(listener); + } + } + + private void notifyListeners() { + DisposableListener[] list; + + synchronized (listeners) { + //noinspection unchecked,ToArrayCallWithZeroLengthArrayArgument + list = listeners.toArray(new DisposableListener[listeners.size()]); + } + + for (DisposableListener dl : list) { + dl.disposed(this); + } + } + + @Override + @Nullable + public String getTag() { + return tag; + } + + @Override + public void setTag(@Nullable String tag) { + this.tag = tag; + } } \ No newline at end of file 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 ee45299e..a5fcad63 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 @@ -16,7 +16,6 @@ */ package com.epam.deltix.util.vsocket; -import com.google.common.annotations.VisibleForTesting; import com.epam.deltix.util.ContextContainer; import com.epam.deltix.util.io.GUID; import com.epam.deltix.util.io.IOUtil; @@ -29,6 +28,8 @@ import com.epam.deltix.util.time.GlobalTimer; import com.epam.deltix.util.time.TimeKeeper; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.VisibleForTesting; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLSocket; @@ -43,9 +44,14 @@ * */ public class VSClient extends ConnectionStateListener implements Disposable, DisposableListener { + public static final int MIN_ALLOWED_SERVER_VERSION = 1014; + public static final int MIN_COMP_SERVER_VERSION = VSProtocol.VERSION; public static final int MAX_COMP_SERVER_VERSION = VSProtocol.VERSION; + public static final String SSL_TERMINATION_PROPERTY = "TimeBase.network.VSClient.sslTermination"; + 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); @@ -59,8 +65,8 @@ public class VSClient extends ConnectionStateListener implements Disposable, Dis // Protects "dispatcher" field and interactions with quick executor private final Object dispatcherLock = new Object(); - private final String clientId; - private long serverTime = -1; + private String clientId; + private long serverTime = -1; private volatile DisconnectEventListener listener; private int reconnectInterval; @@ -70,13 +76,15 @@ public class VSClient extends ConnectionStateListener implements Disposable, Dis private int timeout = Integer.getInteger("TimeBase.network.VSClient.timeout", 5000); private boolean enableSSL = false; + private final boolean sslTermination; private int sslPort = 0; private SSLContext sslContext; - private boolean aeronEnabled = false; private final ContextContainer contextContainer; + private int protocolVersion = VSProtocol.VERSION; + 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. @@ -146,6 +154,10 @@ public void run() { } else { VSProtocol.LOGGER.log(Level.INFO, "Reconnect failed (no error), connection " + socket.getSocketIdStr() + ", address " + socket.getRemoteAddress() + ", attempt " + attemptNumber); } + } catch (ConnectionRejectedException e) { + // Explicit reject from server. That means that we should not try to reconnect anymore. + transportLost = true; + VSProtocol.LOGGER.log(Level.INFO, "Reconnect rejected by server, connection " + socket.getSocketIdStr() + ", address " + socket.getRemoteAddress() + ", attempt " + attemptNumber); } catch (IOException e) { VSProtocol.LOGGER.log(Level.INFO, "Reconnect failed (" + e.getMessage() + "), connection " + socket.getSocketIdStr() + ", address " + socket.getRemoteAddress() + ", attempt " + attemptNumber); } catch (TransportRecoveryFailre transportRecoveryFailre) { @@ -204,20 +216,27 @@ public void run() { }, new Date(nextAttemptTimestamp)); } - @VisibleForTesting // Should by used in tests ONLY. TODO: Delete? + @org.jetbrains.annotations.VisibleForTesting // Should by used in tests ONLY. TODO: Delete? public VSClient (String host, int port, String ownerID) throws IOException { this(host, port, ownerID, false, ContextContainer.getContextContainerForClientTests()); } - @VisibleForTesting // Should by used in tests ONLY. TODO: Create a factory method with name like "createClientForTests" + @VisibleForTesting + // Should by used in tests ONLY. TODO: Create a factory method with name like "createClientForTests" public VSClient (String host, int port) throws IOException { this(host, port, null, false, ContextContainer.getContextContainerForClientTests()); } public VSClient(String host, int port, String ownerID, boolean enableSSL, ContextContainer contextContainer) throws IOException { + this(host, port, ownerID, enableSSL, SSL_TERMINATION, contextContainer); + } + + public VSClient(String host, int port, String ownerID, boolean enableSSL, boolean sslTermination, + ContextContainer contextContainer) throws IOException { this.host = host; this.port = port; this.enableSSL = enableSSL; + this.sslTermination = sslTermination; this.contextContainer = contextContainer; this.reconnector = createReconnectorTask(contextContainer.getQuickExecutor()); @@ -225,12 +244,10 @@ public VSClient(String host, int port, String ownerID, boolean enableSSL, Contex this.clientId = new GUID().toStringWithPrefix (InetAddress.getLocalHost().getHostAddress() + ":"); else this.clientId = new GUID().toStringWithPrefix(InetAddress.getLocalHost().getHostAddress() + ":" + ownerID + ":"); + } - /* - if (TRANSPORT_RECONNECT_ATTEMPT_INTERVAL < soTimeout) { - VSProtocol.LOGGER.warning("Reconnect interval (" + TRANSPORT_RECONNECT_ATTEMPT_INTERVAL + ") should not be less than socket open timeout (" + soTimeout + ")"); - } - */ + public void setClientAddress(String address, String ownerID) { + this.clientId = new GUID().toStringWithPrefix(address + ":" + ownerID + ":"); } public int getSoTimeout () { @@ -289,6 +306,21 @@ public boolean isConnected() { return dispatcher != null && dispatcher.hasAvailableTransport(); } + /** + * 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 true if connected, false if disconnected + */ + public boolean tryGetConnectionStatus() { + if (dispatcher == null) { + return false; + } else { + return dispatcher.tryGetConnectionStatus(); + } + } + public void connect () throws IOException { if (dispatcher != null) throw new IllegalStateException("Already connected"); @@ -321,46 +353,108 @@ public void increaseNumTransportChannels() throws IOExc dispatcher.addTransportChannel (openTransport ()); } - private Socket processSSLHandshake(Socket socket) throws IOException { - InputStream is = socket.getInputStream(); - OutputStream os = socket.getOutputStream(); + @NotNull + private Socket setupSocket() throws IOException { + // If SSL termination is enabled, then we start with SSL socket and will NOT try to perform upgrade. + // This is necessary because intermediate proxies will get confused + // if we start with non-SSL socket and then upgrade to SSL after negotiation with TB. + boolean startWithSSL = enableSSL && sslTermination; - os.write(0); //first byte of VS protocol - os.write(VSProtocol.getHeader(enableSSL)); - os.flush(); + InetSocketAddress socketAddress = new InetSocketAddress(host, port); - int serverResponse = is.read(); - if (serverResponse == VSProtocol.CONN_RESP_SSL_NOT_SUPPORTED) - throw new IOException("Server not supported SSL."); + Socket socket = null; + boolean success = false; + try { + if (startWithSSL) { + VSProtocol.LOGGER.info("SSL termination enabled: creating SSL socket on [" + host + ":" + port + "]"); + socket = sslContext.getSocketFactory().createSocket(); + } else { + socket = new Socket(); + } - int serverHeader = is.read(); - if (serverHeader == VSProtocol.SSL_HEADER) { - socket = sslContext.getSocketFactory().createSocket( - socket, socket.getInetAddress().getHostAddress(), socket.getPort(), false); - ((SSLSocket) socket).setUseClientMode(true); - ((SSLSocket) socket).startHandshake(); - enableSSL = true; - VSProtocol.LOGGER.info("Socket upgraded to SSL socket! Now connection is secured."); - } else { - if (enableSSL) - VSProtocol.LOGGER.info("Connection isn't secured."); - enableSSL = false; + socket.setSoTimeout(soTimeout); + 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); + + // Connect + socket.connect(socketAddress, timeout); + + InputStream is = socket.getInputStream(); + OutputStream os = socket.getOutputStream(); + + // We should not request SSL from TB server if SSL termination is enabled + boolean requestSSL = enableSSL && !sslTermination; + + os.write(0); //first byte of VS protocol + os.write(VSProtocol.getHeader(requestSSL)); + os.flush(); + + int serverResponse = is.read(); + if (serverResponse == VSProtocol.CONN_RESP_SSL_NOT_SUPPORTED) { + assert !startWithSSL; + throw new IOException("Server not supported SSL."); + } else if (serverResponse != VSProtocol.CONN_RESP_OK) { + throw new RuntimeException("Unexpected server response: " + serverResponse); + } + + int serverHeader = is.read(); + if (serverHeader == VSProtocol.SSL_HEADER) { + + if (startWithSSL) { + throw new IllegalStateException("SSL termination is enabled but server attempts to upgrade to SSL"); + } + + // Upgrade non-SSL socket to SSL + socket = sslContext.getSocketFactory().createSocket( + socket, socket.getInetAddress().getHostAddress(), socket.getPort(), true); + ((SSLSocket) socket).setUseClientMode(true); + ((SSLSocket) socket).startHandshake(); + enableSSL = true; // We now use SSL socket, even if client have not requested it. + VSProtocol.LOGGER.info("Socket upgraded to SSL socket! Now connection is secured."); + } else if (serverHeader == VSProtocol.HEADER) { + if (enableSSL && !startWithSSL) { + // Normally we should not get here: + // 1. If SSL termination is enabled, then we don't request SSL from TB server + // 2. If we have enableSSL = true, then we should request it from TB server and get explicit reject if it's not supported + VSProtocol.LOGGER.info("Connection isn't secured."); + enableSSL = false; + } + } else { + throw new RuntimeException("Unexpected server header: " + serverHeader); + } + + success = true; + return socket; + } finally { + if (!success) { + IOUtil.close(socket); + } } + } - return socket; + public void setProtocolVersion(int version) { + if (version < VSClient.MIN_ALLOWED_SERVER_VERSION || version > VSClient.MAX_COMP_SERVER_VERSION) + throw new IllegalArgumentException("Protocol version should be in range [" + + VSClient.MIN_ALLOWED_SERVER_VERSION + ", " + VSClient.MAX_COMP_SERVER_VERSION + "]"); + + this.protocolVersion = version; } + /** Used for re-connecting existing VSocket */ @SuppressFBWarnings(value = "UNENCRYPTED_SOCKET", justification = "Timebase ports should be protected from public access by SSL-terminating NLB") private VSocket openTransport (VSocket stopped) throws IOException, TransportRecoveryFailre { - Socket s = new Socket(); + Socket s = null; boolean ok = false; try { - s.setSoTimeout(soTimeout); - s.setTcpNoDelay(true); - s.connect(new InetSocketAddress(host, port), timeout); - - s = processSSLHandshake(s); + s = setupSocket(); ClientConnection cc = new ClientConnection(s); @@ -368,18 +462,20 @@ private VSocket openTransport (VSocket stopped) throws I DataInputStream dis = new DataInputStream (cc.getBufferedInputStream()); //check version compatibility - dos.writeInt (VSProtocol.VERSION); + dos.writeInt (protocolVersion); dos.writeUTF(clientId); dos.flush(); int spv = dis.readInt (); String sid = dis.readUTF (); - if (spv < MIN_COMP_SERVER_VERSION || spv > MAX_COMP_SERVER_VERSION) + if (spv != protocolVersion && (spv < MIN_COMP_SERVER_VERSION || spv > MAX_COMP_SERVER_VERSION)) throw new IncompatibleClientException (sid, spv); sslPort = dis.readInt(); - processTransportHandshake(dis); + + if (protocolVersion > 1014) + processTransportHandshake(dis); //send other sync data dos.writeBoolean(false); // restore @@ -430,33 +526,31 @@ private VSocket openTransport (VSocket stopped) throws I @SuppressFBWarnings(value = "UNENCRYPTED_SOCKET", justification = "Timebase ports should be protected from public access by SSL-terminating NLB") VSocket openTransport () throws IOException { - Socket s = new Socket(); + Socket s = null; boolean ok = false; TransportType transportType; ClientConnection cc; try { - s.setSoTimeout(soTimeout); - s.setTcpNoDelay(true); - s.connect(new InetSocketAddress(host, port), timeout); - - s = processSSLHandshake(s); + s = setupSocket(); cc = new ClientConnection(s); DataOutputStream dos = new DataOutputStream (cc.getOutputStream()); DataInputStream dis = new DataInputStream (cc.getBufferedInputStream()); //check version compatibility - dos.writeInt (VSProtocol.VERSION); + dos.writeInt (protocolVersion); dos.writeUTF(clientId); dos.flush (); int spv = dis.readInt (); String sid = dis.readUTF (); - if (spv < MIN_COMP_SERVER_VERSION || spv > MAX_COMP_SERVER_VERSION) + + if (spv != protocolVersion && (spv < MIN_COMP_SERVER_VERSION || spv > MAX_COMP_SERVER_VERSION)) throw new IncompatibleClientException (sid, spv); sslPort = dis.readInt(); - transportType = processTransportHandshake(dis); + + transportType = (protocolVersion > 1014) ? processTransportHandshake(dis) : TransportType.SOCKET_TCP; //send other sync data dos.writeBoolean(true); @@ -484,7 +578,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) @@ -497,11 +591,7 @@ VSocket openTransport () throws IOException { private TransportType processTransportHandshake(DataInputStream dis) throws IOException { TransportType transportType = TransportType.values()[dis.readInt()]; if (transportType == TransportType.AERON_IPC) { - String aeronDir = dis.readUTF(); - if (!aeronEnabled) { - DXAeron.start(aeronDir, false); - aeronEnabled = true; - } + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); } else if (transportType == TransportType.OFFHEAP_IPC) { OffHeap.start(dis.readUTF(), false); } @@ -536,14 +626,15 @@ public void close () { } private void close(boolean waitForChannelsToFinish) { + boolean triggerDisconnectEvent; synchronized (dispatcherLock) { closed = true; - if (aeronEnabled) - DXAeron.shutdown(); - VSDispatcher d = dispatcher; + // If dispatcher is null, then we already disconnected or even never were connected. + triggerDisconnectEvent = d != null; + if (d != null) { d.setStateListener(null); d.removeDisposableListener(this); @@ -554,6 +645,15 @@ 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. + if (triggerDisconnectEvent) { + DisconnectEventListener listenerRef = listener; + if (listenerRef != null) { + listenerRef.onDisconnected(); + } + } } public void setDisconnectedListener(DisconnectEventListener listener) { @@ -624,15 +724,14 @@ public void disposed (VSDispatcher d) { synchronized (dispatcherLock) { closed = true; - if (aeronEnabled) - DXAeron.shutdown(); - if (d == dispatcher) { d.setStateListener(null); d.removeDisposableListener(this); contextContainer.getQuickExecutor().shutdownInstance(); dispatcher = null; + + onDisconnected(); } } } @@ -649,4 +748,4 @@ public int getSSLPort() { public String toString () { return ("VSClient (" + host + ":" + port + ")"); } -} \ No newline at end of file +} 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 34c5cc6f..32f599e4 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 @@ -23,16 +23,19 @@ import com.epam.deltix.util.concurrent.QuickExecutor; import com.epam.deltix.util.lang.*; +import com.epam.deltix.util.time.TimeKeeper; import com.epam.deltix.util.time.TimerRunner; import com.epam.deltix.util.memory.DataExchangeUtils; +import net.jcip.annotations.GuardedBy; +import java.io.EOFException; +import java.io.IOException; +import java.net.SocketException; import java.net.SocketTimeoutException; import java.util.*; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; -import java.io.*; -import java.net.SocketException; /** * @@ -46,15 +49,24 @@ public final class VSDispatcher implements Disposable { private final Date creationDate = new Date (); private Timer timer; + @GuardedBy("transportChannels") private final ObjectHashSet transportChannels = new ObjectHashSet<> (); + /** + * Set to non-null value during transport channel recovery. + * If multiple transport channels are being recovered, this future will be completed with value "true" if all of + * them are recovered successfully, or "false" if at least one of them failed to recover. + */ + + @GuardedBy("freeChannels") + // TODO: Replace by Deque private final Stack freeChannels = new Stack <> (); private final ArrayList channels = new ArrayList <> (10); - private volatile boolean hasAvailableTransport = false; + //private volatile boolean hasAvailableTransport = false; volatile VSConnectionListener connectionListener = null; volatile ConnectionStateListener stateListener; @@ -63,6 +75,7 @@ public final class VSDispatcher implements Disposable { private int reconnectInterval; private String address; + private final String clientAddress; private String applicationID; // State of Dispatcher on remote side @@ -71,6 +84,10 @@ public final class VSDispatcher implements Disposable { private volatile long totalBytes = 0; // number of bytes sent private final EMA average = new EMA(1000 * 60); // 1 minute + private volatile VSDispatcherState state = VSDispatcherState.DISCONNECTED; + + // Set to "true" once all operations related to closing the dispatcher are completed, + // just before calling notifyListeners() private final AtomicBoolean disposed = new AtomicBoolean(false); private TimerTask flusher = new TimerRunner() { @@ -91,35 +108,44 @@ protected void runInternal() { VSChannelImpl channel = list[i]; try { if (channel != null && channel.isAutoflush()) { - channel.getOutputStream().flushAvailable(); + // For low latency channels (noDelay==true) we do not want to flush all + // the accumulated data at once because the remaining data will be sent + // by ChannelExecutor shortly. This allows to get more steady rate. + // For regular channels (noDelay==false) we want to send all the data + // (there is nobody else to do that). + channel.getOutputStream().flushAvailable(!channel.getNoDelay()); } } catch (ChannelClosedException e) { // ignore } catch (ConnectionAbortedException e) { VSProtocol.LOGGER.log (Level.WARNING, "Client unexpectedly drop connection. Remote address: " + channel.getRemoteAddress()); - } catch (com.epam.deltix.util.io.UncheckedIOException | IOException e ) { + } catch (Exception e) { VSProtocol.LOGGER.log (Level.WARNING, "Exception while flushing data. Remote address: " + channel.getRemoteAddress(), e); } } - // do keep-alive assuming that this task runs every millisecond - if (runs++ % VSProtocol.KEEP_ALIVE_INTERVAL == 0) { - synchronized (transportChannels) { - transports = transportChannels.toArray(transports); - size = transportChannels.size(); - } + try { + // do keep-alive assuming that this task runs every millisecond + if (runs++ % VSProtocol.KEEP_ALIVE_INTERVAL == 0) { + synchronized (transportChannels) { + transports = transportChannels.toArray(transports); + size = transportChannels.size(); + } - long bytes = 0; - for (int i = 0; i < size; i++) { - VSTransportChannel transport = transports[i]; - transport.keepAlive(); - bytes += transport.socket.getOutputStream().getBytesWritten(); - bytes += transport.socket.getInputStream().getBytesRead(); - } + long bytes = 0; + for (int i = 0; i < size; i++) { + VSTransportChannel transport = transports[i]; + transport.keepAlive(); + bytes += transport.socket.getOutputStream().getBytesWritten(); + bytes += transport.socket.getInputStream().getBytesRead(); + } - throughput = (bytes - totalBytes) / VSProtocol.KEEP_ALIVE_INTERVAL * 1000; - average.register(throughput); - totalBytes = bytes; + throughput = (bytes - totalBytes) / VSProtocol.KEEP_ALIVE_INTERVAL * 1000; + average.register(throughput); + totalBytes = bytes; + } + } catch (Exception ex) { + VSProtocol.LOGGER.log (Level.WARNING, "Exception while sending transport keep-alive.", ex); } } }; @@ -136,11 +162,14 @@ protected void runInternal() { */ public VSDispatcher(String clientId, boolean isClient, ContextContainer contextContainer) { this.clientId = clientId; + + String[] parts = clientId.split(":"); + this.clientAddress = parts.length > 1 ? parts[0] : null; + this.isClient = isClient; this.contextContainer = contextContainer; timer = new Timer ("Flush Timer (" + this + ")", true); - timer.scheduleAtFixedRate (flusher, 1L, 1L); this.transportChannelThreadFactory = new AffinityThreadFactoryBuilder(contextContainer.getAffinityConfig()) @@ -172,12 +201,16 @@ public int getReconnectInterval() { public String getApplicationID() { if (applicationID == null) { - String[] parts = clientId.split(":"); - applicationID = parts.length == 4 ? parts[2] : ""; + applicationID = getApplicationID(clientId); } return applicationID; } + public static String getApplicationID(String clientId) { + String[] parts = clientId.split(":"); + return parts.length == 4 ? parts[2] : ""; + } + public void setApplicationID(String applicationID) { this.applicationID = applicationID; } @@ -202,25 +235,25 @@ public int getNumTransportChannels () { public boolean hasTransportChannels() { synchronized (transportChannels) { - return transportChannels.size() > 0; + return !transportChannels.isEmpty(); } } public boolean hasAvailableTransport() { - return hasAvailableTransport; + return state == VSDispatcherState.CONNECTED; } public void addTransportChannel (VSocket socket) throws IOException { - boolean hasTransport = hasAvailableTransport; + boolean hasTransport = state == VSDispatcherState.CONNECTED; VSTransportChannel tc = new VSTransportChannel(this, socket, transportChannelThreadFactory); tc.checkedOut = true; // Initially this channel is not in "freeChannels" so it is effectively "checked out" // set that we have transport before starting transport channel thread if (!hasTransport) - hasAvailableTransport = true; + state = VSDispatcherState.CONNECTED; // start transport tc.start (); @@ -259,17 +292,15 @@ void transportStopped (VSTransportChannel channel, Throwa long startTime = System.currentTimeMillis(); long endTime = startTime + reconnectInterval; - boolean transportIsUnrecoverablyBroken = false; - boolean wasCheckedIn; + synchronized (transportChannels) { if (transportChannels.isEmpty()) // already closed return; - if (!transportChannels.remove (channel)) // check that channel already removed + if (!transportChannels.remove(channel)) // check that channel already removed return; - synchronized (freeChannels) { wasCheckedIn = freeChannels.remove(channel); assert wasCheckedIn == !channel.checkedOut; @@ -286,14 +317,17 @@ void transportStopped (VSTransportChannel channel, Throwa } } - hasAvailableTransport = transportChannels.size() > 0; + state = VSDispatcherState.CONNECTING; } + boolean transportIsUnrecoverablyBroken = false; + + // trying to recover transport VSocketRecoveryInfo recoveryInfo = new VSocketRecoveryInfo(channel.socket, startTime); long now = System.currentTimeMillis(); if (wasCheckedIn && (now < endTime)) { - // notify state listener that transport lost + if (stateListener != null) { if (stateListener.onTransportStopped(recoveryInfo)) { transportIsUnrecoverablyBroken = true; @@ -304,8 +338,8 @@ void transportStopped (VSTransportChannel channel, Throwa // System.out.println("WAITED: remoteConnected=" + remoteConnected + " transportIsUnrecoverablyBroken=" + transportIsUnrecoverablyBroken); try { // Try to wait for connection restore + state = VSDispatcherState.CONNECTING; - //noinspection SynchronizationOnLocalVariableOrMethodParameter synchronized (recoveryInfo) { long timeToWait; while ((timeToWait = endTime - now) > 0 && recoveryInfo.isWaitingForRecovery() && remoteConnected) { @@ -340,20 +374,25 @@ void transportStopped (VSTransportChannel channel, Throwa } 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 - hasAvailableTransport = false; + state = VSDispatcherState.DISCONNECTED; // 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); + 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); + 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 { @@ -372,23 +411,62 @@ void transportStopped (VSTransportChannel channel, Throwa stateListener.onDisconnected(); close(); + } else { + state = VSDispatcherState.CONNECTED; } } /** - * Waits for the specified channed to become checked in. + * 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 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) + return false; + Thread.sleep(period); + } + } catch (InterruptedException e) { + } + + if (state == VSDispatcherState.CONNECTED) + return true; + else if (state == VSDispatcherState.DISCONNECTED) + return false; + + return false; + } + + /** + * Waits for the specified channel to become checked in. * * @param channel channel to wait for * @param now current time * @param endTime completion deadline (will stop after this time even if channel still checked out) * @return true if channel was checked in */ + @GuardedBy("transportChannels") private boolean waitForTransportCheckIn(VSTransportChannel channel, long now, long endTime) { assert Thread.holdsLock(transportChannels); boolean checkedIn = false; try { - while (now < endTime && !checkedIn) { + while (now < endTime && !checkedIn && !disposed.get()) { transportChannels.wait(endTime - now); now = System.currentTimeMillis(); synchronized (freeChannels) { @@ -439,7 +517,7 @@ VSTransportChannel checkOut () { synchronized (freeChannels) { for (;;) { - if (!hasAvailableTransport && !remoteConnected) + if (state != VSDispatcherState.CONNECTED && !remoteConnected) throw new ConnectionAbortedException("Connection aborted from remote side [" + getRemoteAddress() + "]"); if (!freeChannels.isEmpty ()) { @@ -495,7 +573,7 @@ void onRemoteClosed() { } private void sendClosing() { - if (!remoteConnected || !hasAvailableTransport) + if (!remoteConnected || state != VSDispatcherState.CONNECTED) return; VSTransportChannel channel = null; @@ -524,7 +602,8 @@ public void close () { transportChannels.notify(); } - remoteConnected = hasAvailableTransport = false; + state = VSDispatcherState.DISCONNECTED; + remoteConnected = false; // disable free channels to prevent locking on code below synchronized (freeChannels) { @@ -558,6 +637,7 @@ public void close () { VSChannelImpl newChannel (int inCapacity, int outCapacity, boolean compressed) { VSChannelImpl vsc; + if (!remoteConnected) { throw new IllegalStateException("Attempt to create new channel after disconnect"); } @@ -600,6 +680,10 @@ public String getRemoteAddress() { return address; } + public String getClientAddress() { + return "/" + clientAddress + ":"; + } + VSChannelImpl getChannel (int id) { synchronized (channels) { if (id >= channels.size() ) { @@ -632,17 +716,18 @@ void channelClosed (VSChannelImpl vsc) { int id = vsc.getLocalId(); synchronized (channels) { - if (vsc.equals(channels.get(id))) - channels.set (id, null); - else - VSProtocol.LOGGER.log (Level.SEVERE, "Trying to remove wrong channel."); + if (vsc.equals(channels.get(id))) { + channels.set(id, null); - activeChannels--; - channels.notify(); + activeChannels--; + channels.notify(); + } else { + VSProtocol.LOGGER.log(Level.SEVERE, "Trying to remove wrong channel."); + } } } - public void addDisposableListener(DisposableListener listener) { + public void addDisposableListener(DisposableListener listener) { synchronized (listeners) { if (!listeners.contains(listener)) listeners.add(listener); diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcherState.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcherState.java new file mode 100644 index 00000000..cccd66dd --- /dev/null +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSDispatcherState.java @@ -0,0 +1,7 @@ +package com.epam.deltix.util.vsocket; + +public enum VSDispatcherState { + CONNECTED, + CONNECTING, + DISCONNECTED +} diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSOutputStream.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSOutputStream.java index 92240493..47ad392c 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSOutputStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSOutputStream.java @@ -35,8 +35,11 @@ public abstract class VSOutputStream extends OutputStream { /* * Flushes portion of data that can recieved on the remote side immediately. + * + * @param flushAll attempts to flush all available data, even if it requires multiple send operations + * and blocking for longer time. */ - public abstract void flushAvailable() throws IOException; + public abstract int flushAvailable(boolean flushAll) throws IOException; //public abstract int available(); } \ No newline at end of file diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSServerFramework.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSServerFramework.java index 9b0bc12b..2548e88f 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSServerFramework.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSServerFramework.java @@ -44,7 +44,7 @@ public class VSServerFramework implements ConnectionHandshakeHandler, Disposable public final static short MAX_SOCKETS_PER_CONNECTION = 8; private final Map dispatchers = - new HashMap <> (); + new HashMap <> (); private final QuickExecutor executor; private final ContextContainer contextContainer; @@ -53,7 +53,7 @@ public class VSServerFramework implements ConnectionHandshakeHandler, Disposable private final int connectionsLimit; private final short transportsLimit; private final long time; - private int reconnectInterval; + private final int reconnectInterval; private final VSCompression compression; private TLSContext tlsContext; @@ -149,7 +149,7 @@ public void initTransport(TransportProperties transportPropertie if (transportProperties != null) { transportType = transportProperties.transportType; if (transportType == TransportType.AERON_IPC) { - DXAeron.start(transportProperties.transportDir, false); + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); } else if (transportType == TransportType.OFFHEAP_IPC) { OffHeap.start(transportProperties.transportDir, true); } @@ -161,9 +161,9 @@ public boolean handleHandshake (Socket s) throws IOException { s.setTcpNoDelay(true); s.setKeepAlive(true); + BufferedInputStream bis = new BufferedInputStream(s.getInputStream(), VSocketImpl.INPUT_STREAM_BUFFER_SIZE); return handleHandshake( - SocketConnectionFactory.createConnection( - s, new BufferedInputStream(s.getInputStream()), s.getOutputStream()) + SocketConnectionFactory.createConnection(s, bis, s.getOutputStream()) ); } @@ -175,7 +175,7 @@ public boolean handleHandshake(Socket s, BufferedInputStream is, OutputStream os s.setKeepAlive(true); return handleHandshake( - SocketConnectionFactory.createConnection(s, is, os) + SocketConnectionFactory.createConnection(s, is, os) ); } @@ -224,10 +224,10 @@ private boolean handleHandshakeInternal (Connection c) throws IOExc if (!isCompatible) { VSProtocol.LOGGER.severe ( - "Connection from " + clientId + " rejected due to incompatible protocol version #" + - clientVersion + " (accepted: " + - MIN_COMPATIBLE_CLIENT_VERSION + " .. " + - MAX_COMPATIBLE_CLIENT_VERSION + ")" + "Connection from " + clientId + " rejected due to incompatible protocol version #" + + clientVersion + " (accepted: " + + MIN_COMPATIBLE_CLIENT_VERSION + " .. " + + MAX_COMPATIBLE_CLIENT_VERSION + ")" ); dout.writeByte (VSProtocol.CONN_RESP_INCOMPATIBLE_CLIENT); @@ -248,7 +248,7 @@ private boolean handleHandshakeInternal (Connection c) throws IOExc boolean isNew = dis.readBoolean(); int sCode = dis.readInt(); - long recieved = dis.readLong(); + long received = dis.readLong(); Connector connector = process(clientId); if (connector == null) { @@ -272,6 +272,17 @@ private boolean handleHandshakeInternal (Connection c) throws IOExc VSProtocol.LOGGER.fine("Recovery attempt failed because another attempt for this thread in progress"); } } + } else { + if (!isNew) { + // Client attempts to recover a connection but there are no connection with such "sCode" on the server side. + // That may happen if server was restarted. In such case we should reject the connection + // and force a client to do a full reconnect. + VSProtocol.LOGGER.warning("Connection restore failed for transport (" + sCode + ") for " + clientId + " because server side transport is not found"); + + dout.writeByte(VSProtocol.CONN_RESP_CONNECTION_REJECTED); + dout.flush(); + return false; + } } boolean success = false; @@ -280,7 +291,7 @@ private boolean handleHandshakeInternal (Connection c) throws IOExc String transportTag = sCode + " / " + Integer.toHexString(sCode); if (broken != null) { VSProtocol.LOGGER.info("Restoring connection (" + transportTag + ") for " + clientId); - broken.getOutputStream().confirm(recieved); + broken.getOutputStream().confirm(received); } else { if (!isNew) { VSProtocol.LOGGER.warning("Connection restore failed for transport (" + transportTag + ") for " + clientId + " because server side transport is not found"); @@ -361,6 +372,7 @@ private void processSSLHandshake(Connection c) throws IOExcep int clientHeader = is.read(); if (clientHeader == VSProtocol.SSL_HEADER && !enableSSL) { os.write(VSProtocol.CONN_RESP_SSL_NOT_SUPPORTED); + os.flush(); throw new IOException("Client wants SSL but server have not prepared for handshake."); } os.write(VSProtocol.CONN_RESP_OK); @@ -392,7 +404,7 @@ private void processTransportHandshake(Connection c) throws I c.setTransportType(type); if (type == TransportType.AERON_IPC) - dout.writeUTF(DXAeron.getAeronDir()); + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); else if (type == TransportType.OFFHEAP_IPC) dout.writeUTF(OffHeap.getOffHeapDir()); } @@ -413,7 +425,7 @@ public void disposed(VSDispatcher resource) { @Override public void close() { if (transportType == TransportType.AERON_IPC) - DXAeron.shutdown(); + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); } static class Connector extends ConnectionStateListener implements Closeable { @@ -573,9 +585,7 @@ private static class FakeRecoveryInfo extends VSocketRecoveryInfo { @Override public String toString() { - return "FakeVSocket{" + - "" + label + '\'' + - '}'; + return "FakeVSocket{" + label + '}'; } } } \ No newline at end of file diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSTransportChannel.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSTransportChannel.java index f0a6bb1a..b7c4d0d3 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSTransportChannel.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSTransportChannel.java @@ -53,7 +53,9 @@ class VSTransportChannel implements Runnable, Disposable { private volatile boolean closed = false; volatile long latency = Long.MAX_VALUE; - private volatile long reported; + // Value at which the "completeTask" should be triggered next time. + // Checked and updated by transport thread. + private long nextReportValue = VSocketOutputStream.REPORT_THRESHOLD; private final Thread thread; @@ -62,6 +64,9 @@ class VSTransportChannel implements Runnable, Disposable { @Nonnull private QuickExecutor.QuickTask createCompleteTask(QuickExecutor quickExecutor) { return new QuickExecutor.QuickTask(quickExecutor) { + // Bytes that already reported. Checked and updated by completeTask. + private long reported = 0; + @Override public void run() throws InterruptedException { long bytesRead; @@ -71,8 +76,9 @@ public void run() throws InterruptedException { // We already reported this value return; } - DataExchangeUtils.writeLong(bytesReport, 2, (reported = bytesRead)); + DataExchangeUtils.writeLong(bytesReport, 2, bytesRead); out.write(bytesReport, 0, bytesReport.length); + reported = bytesRead; } if (LOGGER.isLoggable(Level.FINEST)) { LOGGER.log(Level.FINEST, "Sent BYTES_RECIEVED report: " + bytesRead + " from " + socket.getSocketIdStr()); @@ -166,9 +172,12 @@ public void run () { if (currentThread.isInterrupted()) throw new InterruptedException(); - if (vin.getBytesRead() - reported > VSocketOutputStream.CAPACITY / 4) + long bytesRead = vin.getBytesRead(); + if (bytesRead >= nextReportValue) { + nextReportValue = bytesRead + VSocketOutputStream.REPORT_THRESHOLD; completeTask.submit(); - + } + int destId = din.readUnsignedShort (); //System.out.println(this.socket + ": signal = " + destId); diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketFactory.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketFactory.java index 1e80de2d..e1400813 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketFactory.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketFactory.java @@ -32,9 +32,6 @@ public enum Transport { Memory } - private final static HashMap cache = - new HashMap(); - public static volatile Transport transport; public static VSocket get(ClientConnection cc, TransportType transportType) throws IOException { @@ -56,7 +53,7 @@ public static VSocket get(ClientConnection cc, TransportType transportType) thro Socket s = cc.getSocket(); if (transportType == TransportType.AERON_IPC) { - return new AeronIpcSocket(s, s.hashCode(), false, socketNumber); + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); } else if (transportType == TransportType.OFFHEAP_IPC) return new OffHeapIpcSocket(s, s.hashCode(), false, socketNumber); @@ -70,8 +67,6 @@ public static VSocket get(ClientConnection cc, VSocket stopped) throws IOExcepti VSocket socket; if (stopped instanceof VSocketImpl) { socket = new VSocketImpl(cc, socketNumber); - } else if (stopped instanceof AeronIpcSocket) { - socket = new AeronIpcSocket(s, stopped.getCode(), false, socketNumber); } else if (stopped instanceof OffHeapIpcSocket) { socket = new OffHeapIpcSocket(s, stopped.getCode(), false, socketNumber); } else { diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketImpl.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketImpl.java index 0be2fc92..4b2e6415 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketImpl.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketImpl.java @@ -23,52 +23,103 @@ import java.net.Socket; import java.io.*; import java.net.SocketAddress; +import java.net.SocketException; import java.util.logging.Level; /** * Date: Mar 5, 2010 */ public class VSocketImpl implements VSocket { + // Buffer size should be at least big enough to fit maximum packet size (VSProtocol.MAX_SIZE). // Controls buffer sized. "Default buffer size" lets you set both send and receive buffer size using single argument. - private static final int SOCKET_DEFAULT_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.bufferSize", 1 << 16); - private static final int SOCKET_RECEIVE_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.receiveBufferSize", SOCKET_DEFAULT_BUFFER_SIZE); - private static final int SOCKET_SEND_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.sendBufferSize", SOCKET_DEFAULT_BUFFER_SIZE); + public static final int SOCKET_DEFAULT_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.bufferSize", 1 << 16); + public static final int SOCKET_RECEIVE_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.receiveBufferSize", SOCKET_DEFAULT_BUFFER_SIZE); + public static final int SOCKET_SEND_BUFFER_SIZE = Integer.getInteger("TimeBase.network.socket.sendBufferSize", SOCKET_DEFAULT_BUFFER_SIZE); + + //@ApiStatus.Experimental // Temporary option for testing performance effect of using buffered reader of different size + // 8kb size matches to the previous value. However it's very likely that we need 64k or 128k buffer size to match the maximum "logical packet" size (VSProtocol.MAXSIZE). + // TODO: Consider increasing default value to 64kb. + public static final int INPUT_STREAM_BUFFER_SIZE = Integer.getInteger("TimeBase.network.streamBufferSize", 8 * 1024); + + public static final boolean PRINT_VSOCKET_SETTINGS = Boolean.getBoolean("TimeBase.network.printSettings"); + + static { + if (PRINT_VSOCKET_SETTINGS) { + System.out.println("SOCKET_RECEIVE_BUFFER_SIZE: " + SOCKET_RECEIVE_BUFFER_SIZE); + System.out.println("SOCKET_SEND_BUFFER_SIZE: " + SOCKET_SEND_BUFFER_SIZE); + System.out.println("INPUT_STREAM_BUFFER_SIZE: " + INPUT_STREAM_BUFFER_SIZE); + } + } private static final int IPTOS_THROUGHPUT = 0x08; + private static final int DEFAULT_TRAFFIC_CLASS = IPTOS_THROUGHPUT; + /** Value for {@link Socket#setTrafficClass(int)} */ + public static final int SOCKET_TRAFFIC_CLASS = validateTrafficClassValue(Integer.getInteger("TimeBase.network.socket.tos", DEFAULT_TRAFFIC_CLASS)); - private Socket socket; - private InputStream in; - private BufferedInputStream bin; + private final Socket socket; + private final InputStream in; + private final BufferedInputStream bin; - private OutputStream out; - private VSocketOutputStream vout; - private VSocketInputStream vin; + private final OutputStream out; + private final VSocketOutputStream vout; + private final VSocketInputStream vin; private String remoteAddress; private int code; private final int socketNumber; + /** + * Configures socket. + * + *

Current implementation of {@link VSClient} executes this on a connected socket. + * While this is allowed to change socket buffer sizes after the connection is established, + * it may have different effects on different platforms.

+ * + *

Most importantly, TCP Window size may be limited by 64k if receive buffer size + * is set after the connection is established.

+ */ private void setUpSocket () { try { socket.setTcpNoDelay (true); socket.setSoTimeout (0); socket.setKeepAlive (true); - socket.setTrafficClass(IPTOS_THROUGHPUT); + + // This is likely to have no effect as corresponding TOS field is deprecated. + // See https://en.wikipedia.org/wiki/Type_of_service + // and https://en.wikipedia.org/wiki/Differentiated_services + socket.setTrafficClass(SOCKET_TRAFFIC_CLASS); SocketAddress address = socket.getRemoteSocketAddress(); remoteAddress = address != null ? address.toString() : null; - // TODO: temp fix for dead-lock in VSockets - // TODO: Clarify TODO above. - - socket.setReceiveBufferSize(SOCKET_RECEIVE_BUFFER_SIZE); - socket.setSendBufferSize(SOCKET_SEND_BUFFER_SIZE); + // We do not change socket buffer sizes here because + // we expect that they are already set using configureBufferSizes(...) method or manually. } catch (IOException x) { VSProtocol.LOGGER.log (Level.WARNING, null, x); } } - public VSocketImpl(ClientConnection cc, int socketNumber) throws IOException { + /** + * Configures buffer sizes for a socket. + * + *

It's important to configure decent buffer sizes. + * It should be at least big enough to fit maximum packet size ({@link VSProtocol#MAXSIZE}). + * Otherwise, there is a possible situation when TimeBase client and server may run into a deadlock, + * when transport thread gets stuck in blocking write into socket.

+ * + *

Please note that if socket is already connected, then OS may ignore these values.

+ * + * Also please note that in case of server-side socket, the "receive" buffer size should be set on + * {@link java.net.ServerSocket} using {@link java.net.ServerSocket#setReceiveBufferSize(int)} method. + * On the client socket, the "receive" buffer size should be set before the connection is established. + * Otherwise, TCP window size will be limited by 64k. + */ + public static void configureBufferSizes(Socket socket) throws SocketException { + socket.setReceiveBufferSize(SOCKET_RECEIVE_BUFFER_SIZE); + socket.setSendBufferSize(SOCKET_SEND_BUFFER_SIZE); + } + + public VSocketImpl(ClientConnection cc, int socketNumber) { this.socket = cc.getSocket(); this.in = cc.getInputStream (); this.bin = cc.getBufferedInputStream(); @@ -101,6 +152,7 @@ public VSocketImpl(Connection c, int code, int socketNumber) { String socketIdStr = getSocketIdStr(); this.vout = new VSocketOutputStream(out, socketIdStr); this.vin = new VSocketInputStream(bin, socketIdStr); + this.socket = null; //setUpSocket (); } @@ -145,4 +197,11 @@ public String toString() { public int getSocketNumber() { return socketNumber; } + + private static int validateTrafficClassValue(int value) { + if (value < 0 || value > 255) { + throw new IllegalArgumentException("Invalid value for TimeBase.network.socket.tos: " + value); + } + return value; + } } \ No newline at end of file diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketOutputStream.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketOutputStream.java index 62154f9f..0b3c20a9 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketOutputStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSocketOutputStream.java @@ -28,8 +28,12 @@ public class VSocketOutputStream extends OutputStream { private final String socketIdStr; - public static int CAPACITY = 1024 * 512; + //@ApiStatus.Experimental + public static int CAPACITY = Integer.getInteger("TimeBase.network.socketOutputStream.bufferCapacity", 1024 * 512); public static int INCREMENT = CAPACITY / 4; + /** Controls how often {@link VSProtocol#BYTES_RECIEVED} message will be sent from {@link VSTransportChannel} */ + //@ApiStatus.Experimental + public static int REPORT_THRESHOLD = Integer.getInteger("TimeBase.network.socketOutputStream.reportThreshold", CAPACITY / 4); private final ByteQueue buffer; private final OutputStream out; diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramInputStream.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramInputStream.java index 342238e5..b3994162 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramInputStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramInputStream.java @@ -50,11 +50,12 @@ private boolean fillBuffer() { connection.remotePort = pack.getPort(); } catch (Exception e) { - VSProtocol.LOGGER.log (Level.WARNING, null, e); + e.printStackTrace(System.out); return false; } int s = DataExchangeUtils.readInt(pack.getData(), 0); + //System.out.println("sequence: " + s); buffer.offer(pack.getData(), 4, pack.getLength() - 4); return true; diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramOutputStream.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramOutputStream.java index f24504aa..47d155f4 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramOutputStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/DatagramOutputStream.java @@ -55,7 +55,7 @@ public synchronized void write(int b) { try { ds.send(packet); } catch (Exception e) { - VSProtocol.LOGGER.log (Level.WARNING, null, e); + e.printStackTrace(System.out); } } @@ -78,7 +78,7 @@ public synchronized void write(byte buf[], int pos, int len) { ds.send(packet); //System.out.println("Send packet, size = " + out.getSize()); } catch (Exception e) { - VSProtocol.LOGGER.log (Level.WARNING, null, e); + e.printStackTrace(System.out); } } } \ No newline at end of file diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/SocketConnection.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/SocketConnection.java index 75282aa7..72a658c1 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/SocketConnection.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/transport/SocketConnection.java @@ -72,7 +72,7 @@ public VSocket create(VSocket stopped) throws IOException { private VSocket create(int code, TransportType type) throws IOException { int socketNumber = VSocketFactory.nextSocketNumber(); if (type == TransportType.AERON_IPC) - return new AeronIpcSocket(socket, code, true, socketNumber); + throw new RuntimeException("Legacy version of Aeron IPC is not supported"); else if (type == TransportType.OFFHEAP_IPC) return new OffHeapIpcSocket(socket, code, true, socketNumber); else @@ -115,7 +115,8 @@ public void upgradeToSSL(SSLSocketFactory sslSocketFactory) ((SSLSocket) socket).startHandshake(); //upgrade streams - in = new BufferedInputStream(socket.getInputStream()); + + in = new BufferedInputStream(socket.getInputStream(), VSocketImpl.INPUT_STREAM_BUFFER_SIZE); out = socket.getOutputStream(); } } \ No newline at end of file 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 7fca4184..089c9ed8 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,9 +33,6 @@ */ public class VSDispatcherTest { - /** - * Test for https://gitlab.deltixhub.com/Deltix/QuantServer/QuantServer/issues/43 - */ @Test (timeout = 5_000) // Note: test timeout must be greater than reconnectInterval + 2000ms public void testNoHangsOnConcurrentDisconnects() throws IOException, InterruptedException { int reconnectInterval = 2000; @@ -132,7 +129,7 @@ boolean onTransportBroken(VSocketRecoveryInfo recoveryInfo) { boolean gotChanelClosedException = false; try { // Note: test may hang here if Dispatcher bug still present - vsChannel.getOutputStream().flushAvailable(); + vsChannel.getOutputStream().flushAvailable(true); } catch (ConnectionAbortedException e) { gotChanelClosedException = true; System.out.println("Got ConnectionAbortedException as expected"); diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickDBClient.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickDBClient.java index b8a66fdc..89589fcc 100644 --- a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickDBClient.java +++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickDBClient.java @@ -21,6 +21,7 @@ import com.epam.deltix.qsrv.hf.tickdb.comm.*; import com.epam.deltix.qsrv.hf.tickdb.pub.*; import com.epam.deltix.qsrv.hf.topic.DirectProtocol; +import com.epam.deltix.timebase.messages.InstrumentKey; import com.epam.deltix.util.io.SSLClientContextProvider; import com.epam.deltix.data.stream.DXChannel; import com.epam.deltix.streaming.MessageChannel; @@ -84,6 +85,7 @@ import com.epam.deltix.util.vsocket.VSClient; import com.epam.deltix.util.vsocket.VSProtocol; import io.aeron.Aeron; +import org.jetbrains.annotations.ApiStatus; import org.jetbrains.annotations.NotNull; import javax.annotation.Nonnull; @@ -116,6 +118,14 @@ public class TickDBClient implements DXRemoteDB, DBStateNotifier, RemoteTickDB, public static final Log LOGGER = LogFactory.getLog("tickdb.client"); + @ApiStatus.Experimental + private static final int MAX_REVERSE_BUFFER_SIZE = Integer.getInteger("TimeBase.transport.channel.maxReverseBufferSize", 64 * 1024); + @ApiStatus.Experimental + private static final int DEFAULT_LOCAL_CHANNEL_SIZE = Integer.getInteger("TimeBase.transport.channel.local.defaultSize", VSProtocol.CHANNEL_BUFFER_SIZE); + @ApiStatus.Experimental + private static final int DEFAULT_REMOTE_CHANNEL_SIZE = Integer.getInteger("TimeBase.transport.channel.remote.defaultSize", VSProtocol.CHANNEL_MAX_BUFFER_SIZE); + + private static int getConnectionsNumber(boolean isRemote) { int value = 2; @@ -127,28 +137,22 @@ private static int getConnectionsNumber(boolean isRemo } } - private UserPrincipal user; + private volatile UserPrincipal user; private final String host; private final int port; private int timeout; - private SSLContext sslContext; private boolean isOpen = false; private boolean isReadOnly = false; - private int serverProtocolVersion; + private int serverProtocolVersion = -1; private String serverVersion = ""; private long[] latency; private long availableBandwidth = 0; - private final ReconnectableImpl connMgr = new ReconnectableImpl("TickDBClient"); - private final Runnable updater = - new Runnable () { - public void run () { - sendMetaDataUpdate (); - } - }; + private final ReconnectableImpl connMgr; + private final Runnable updater = this::sendMetaDataUpdate; private VSClient connection; // @@ -167,10 +171,10 @@ public void run () { private boolean secured = false; private final CodecFactory intpCodecFactory = - CodecFactory.newInterpretingCachingFactory(); + CodecFactory.newInterpretingCachingFactory(); private final CodecFactory compCodecFactory = - CodecFactory.newCompiledCachingFactory (); + CodecFactory.newCompiledCachingFactory (); private boolean useCompression = false; private boolean isRemoteConnection = false; @@ -179,10 +183,10 @@ public void run () { protected final boolean enableSSL; - protected Boolean sslTermination; // null stands for default value from VSClient.SSL_TERMINATION + protected Boolean sslTermination; // null stands for default value from VSClient.SSL_TERMINATION - private final ContextContainer contextContainer; - private final DXClientAeronContext aeronContext; + private final ContextContainer contextContainer; + private final DXClientAeronContext aeronContext; private static final ThreadFactory topicNoAffinityConsumerThreadFactory = new TopicConsumerThreadFactory(); private ThreadFactory topicConsumerThreadFactory; @@ -218,6 +222,7 @@ protected TickDBClient (String host, int port, boolean enableSSL, UserPrincipal this.timeout = isRemoteConnection ? 5000 : 1000; + connMgr = new ReconnectableImpl("TickDBClient", this); //connMgr.setLazyLogger(LOGGER); // connMgr.setLogger(LOGGER); // connMgr.setLogLevel (Level.INFO); @@ -282,19 +287,16 @@ public void setSslTermination(Boolean sslTermination) { } public boolean getSslTermination() { - return false; - // TODO: @MERGE - -// if (this.sslTermination != null) { -// return this.sslTermination; -// } else { -// return VSClient.SSL_TERMINATION; -// } + if (this.sslTermination != null) { + return this.sslTermination; + } else { + return VSClient.SSL_TERMINATION; + } } /* - Tests round-trip latency (in nanoseconds) - */ + * Tests round-trip latency (in nanoseconds) + */ public long[] testConnectLatency(int iterations) throws IOException { long[] times = new long[iterations]; @@ -323,7 +325,7 @@ VSChannel connect (ChannelType type, boolean autoCommit, boolean n } public VSChannel connect(ChannelType type, boolean autoCommit, boolean noDelay, ChannelCompression c, int channelBufferSize) - throws IOException + throws IOException { UserPrincipal user = userPrincipalResolver.resolve(getUser()); VSChannel channel = createChannel(type, autoCommit, noDelay, c, channelBufferSize); @@ -335,58 +337,93 @@ protected UserPrincipal getUser() { return user; } - protected VSChannel createChannel(ChannelType type, boolean autoCommit, boolean noDelay, ChannelCompression c, int channelBufferSize) - throws IOException - { - synchronized (this) { - if (connection == null || !connection.isConnected()) { - Util.close(connection); + /** + * Returns a VSClient instance, ensuring that it is connected. + *

+ * Will wait for transport recovery to complete, if necessary. + * Will re-create a VSClient, if previous is not connected. + * Will throw an exception if fails to connect. + */ + private VSClient getConnectedVSClient() throws IOException { + // This loop is expected to eventually stop as soon as recovery succeeds or fails. + + while (true) { + // Connection instance may be changed from other thread. We have to ensure that we use the correct connection instance. + VSClient connectionRef; + synchronized (this) { + connectionRef = connection; + } - //int connectionPort = port; - String idd = (applicationId != null ? id + ":" + applicationId : id); + boolean isConnected = connectionRef != null && connectionRef.tryGetConnectionStatus(); - if (sslTermination == null) { - connection = new VSClient(host, port, idd, enableSSL, contextContainer); - } else { - connection = new VSClient(host, port, idd, enableSSL, contextContainer); + synchronized (this) { + if (connection != connectionRef) { + // Connection instance was changed from other thread. Retry. + continue; } -// if (address != null) -// connection.setClientAddress(address, idd); - - connection.setNumTransportChannels(isRemoteConnection ? 1 : getConnectionsNumber(isRemoteConnection)); - connection.setTimeout(timeout); - connection.setDisconnectedListener(listener); - connection.setSslContext(SSLClientContextProvider.getSSLContext()); - connection.connect(); - } else if (isRemoteConnection) { - // lazy initialization of additional sockets transports - int number = getConnectionsNumber(true); - if (connection.getNumTransportChannels() < number) - connection.increaseNumTransportChannels(); + if (!isConnected) { + // It's necessary to create a new instance + Util.close(connection); + + String idd = (applicationId != null ? id + ":" + applicationId : id); + + if (sslTermination == null) + connection = new VSClient(host, port, idd, enableSSL, contextContainer); + else + connection = new VSClient(host, port, idd, enableSSL, sslTermination, contextContainer); + + if (address != null) + connection.setClientAddress(address, idd); + + connection.setNumTransportChannels(isRemoteConnection ? 1 : getConnectionsNumber(isRemoteConnection)); + connection.setTimeout(timeout); + connection.setDisconnectedListener(listener); + connection.setSslContext(SSLClientContextProvider.getSSLContext()); + connection.connect(); + } else if (isRemoteConnection) { + // lazy initialization of additional sockets transports + int number = getConnectionsNumber(true); + if (connection.getNumTransportChannels() < number) + connection.increaseNumTransportChannels(); + } + return connection; } } + } + + protected VSChannel createChannel(ChannelType type, boolean autoCommit, boolean noDelay, ChannelCompression c, int channelBufferSize) + throws IOException + { + VSClient vsClient = getConnectedVSClient(); boolean compressed = c == ChannelCompression.AUTO ? useCompression : (c == ChannelCompression.ON); int inCapacity; int outCapacity; - int capacity = isRemoteConnection ? VSProtocol.CHANNEL_MAX_BUFFER_SIZE : VSProtocol.CHANNEL_BUFFER_SIZE; - - if (channelBufferSize > 0) { - // Override ChannelType - inCapacity = channelBufferSize; - outCapacity = channelBufferSize; - } else if (type == ChannelType.Input) { - inCapacity = capacity; - outCapacity = capacity / 4; - } else if (type == ChannelType.Output) { - inCapacity = capacity / 4; - outCapacity = capacity; - } else { - inCapacity = capacity; - outCapacity = capacity / 2; + int defaultCapacity = isRemoteConnection ? DEFAULT_REMOTE_CHANNEL_SIZE : DEFAULT_LOCAL_CHANNEL_SIZE; + + int configuredCapacity = channelBufferSize > 0 ? channelBufferSize : defaultCapacity; + + switch (type) { + case Input: + inCapacity = configuredCapacity; + outCapacity = Math.max(configuredCapacity / 4, MAX_REVERSE_BUFFER_SIZE); + break; + case Output: + inCapacity = Math.max(configuredCapacity / 4, MAX_REVERSE_BUFFER_SIZE); + outCapacity = configuredCapacity; + break; + default: + inCapacity = configuredCapacity; + // By default, set output capacity as half of input capacity. + // However, if explicit channel size is provided, we use it "as is", without changes. + if (channelBufferSize > 0) { + outCapacity = configuredCapacity; + } else { + outCapacity = configuredCapacity / 2; + } } // int inCapacity = isRemoteConnection ? @@ -397,7 +434,7 @@ protected VSChannel createChannel(ChannelType type, boolean autoCommit, // (type == ChannelType.Output ? VSProtocol.CHANNEL_MAX_BUFFER_SIZE : VSProtocol.CHANNEL_BUFFER_SIZE) : // (type == ChannelType.Output ? VSProtocol.CHANNEL_BUFFER_SIZE: VSProtocol.CHANNEL_BUFFER_SIZE / 4); - VSChannel channel = connection.openChannel(inCapacity, outCapacity, compressed); + VSChannel channel = vsClient.openChannel(inCapacity, outCapacity, compressed); channel.setAutoflush(autoCommit); channel.setNoDelay(noDelay); channel.getDataOutputStream().writeInt (TDBProtocol.VERSION); @@ -422,9 +459,9 @@ public String getConnectionString() { public CodecFactory getCodecFactory (ChannelQualityOfService qos) { return ( - CodecFactory.useInterpretedCodecs (qos == ChannelQualityOfService.MIN_INIT_TIME) ? - intpCodecFactory : - compCodecFactory + CodecFactory.useInterpretedCodecs (qos == ChannelQualityOfService.MIN_INIT_TIME) ? + intpCodecFactory : + compCodecFactory ); } @@ -452,32 +489,8 @@ public void setTimeout(int timeout) { this.timeout = timeout; } - public SSLContext getSslContext() { - return sslContext; - } - - public void setSslContext(SSLContext sslContext) { - this.sslContext = sslContext; - } - - protected SSLContext getOrCreateContext() { - if (enableSSL) { - if (sslContext == null) - sslContext = SSLClientContextProvider.getSSLContext(); - return sslContext; - } - - return null; - } - - /** - * Returns server protocol version if client is already connected. - * @return server version - * @throws IllegalStateException if is not open - */ @Override public int getServerProtocolVersion() { - assertOpen(); return serverProtocolVersion; } @@ -498,7 +511,7 @@ public void open(boolean readOnly) { private synchronized boolean syncOpen (boolean readOnly) { - if (isOpen && connMgr.isConnected()) + if (isOpen && isConnected()) throw new IllegalStateException("Database already opened & connected."); if (!connMgr.isConnected()) { @@ -580,9 +593,9 @@ public QuickExecutor getQuickExecutor() { } void checkResponse (VSChannel ds) - throws IOException + throws IOException { - checkResponse(ds, serverProtocolVersion); + checkResponse(ds, serverProtocolVersion); } static void checkResponse (VSChannel ds, int protocolVersion) @@ -715,7 +728,7 @@ public void coolDown () { } public DXTickStream createAnonymousStream ( - StreamOptions options + StreamOptions options ) { throw new UnsupportedOperationException(); @@ -725,23 +738,23 @@ public DXTickStream createAnonymousStream ( } public TickStreamClient createStream ( - String key, - String name, - String description, - int distributionFactor + String key, + String name, + String description, + int distributionFactor ) { return ( - createStream ( - key, - new StreamOptions (StreamScope.DURABLE, name, description, distributionFactor) - ) + createStream ( + key, + new StreamOptions (StreamScope.DURABLE, name, description, distributionFactor) + ) ); } public synchronized TickStreamClient createStream ( - String key, - StreamOptions options + String key, + StreamOptions options ) { assertOpen(); @@ -958,6 +971,7 @@ public MessageChannel createPublisher(@Nonnull String topicKe //Direct Aeron aeron = aeronContext.getAeronInstance(response.getAeronDir(), response.getTransferType()); + return loaderFactory.create( aeron, pref.raw, response.getPublisherChannel(), response.getDataStreamId(), response.getTypes(), @@ -985,10 +999,7 @@ public Disposable createConsumerWorker( DirectReaderFactory factory = new DirectReaderFactory(compCodecFactory, pref.getTypeLoader()); Aeron aeron = aeronContext.getAeronInstance(response.getAeronDir(), response.getTransferType()); - SubscriptionWorker subscriptionWorker = factory.createListener(aeron, pref.raw, response.getChannel(), - response.getDataStreamId(), response.getTypes(), processor, - pref.getEffectiveIdleStrategy(idleStrategy), pref.getTopicDataLossHandler()); - + SubscriptionWorker subscriptionWorker = factory.createListener(aeron, pref.raw, response.getChannel(), response.getDataStreamId(), response.getTypes(), processor, pref.getEffectiveIdleStrategy(idleStrategy), pref.getTopicDataLossHandler()); if (threadFactory == null) { threadFactory = topicConsumerThreadFactory; } @@ -1040,8 +1051,8 @@ private AddTopicSubscriberResponse executeSubscribeRequest(@Nonnull String topic @Override public TickCursor createCursor ( - SelectionOptions options, - TickStream ... streams + SelectionOptions options, + TickStream ... streams ) { assertOpen(); @@ -1117,13 +1128,12 @@ public void close () { synchronized (this) { // free locks - if (session != null) { + if (session != null && connection.tryGetConnectionStatus()) { TickStreamClient[] streams = session.getStreams(); for (TickStreamClient stream : streams) { try { stream.unlock(); } catch (Throwable e) { - LOGGER.warn("Cannot unlock stream [%s]. Error: %s").with(stream.getKey()).with(e); } } @@ -1189,13 +1199,13 @@ private long getLongProperty (int req) { public long getMetaDataVersion () { assertOpen(); - + return (getLongProperty (TDBProtocol.REQ_GET_MD_VERSION)); } private void sendMetaDataUpdate () { assertOpen(); - + VSChannel ds = null; try { @@ -1245,7 +1255,7 @@ private void refreshMetaData () { public synchronized MetaData getMetaData () { refreshMetaData (); return (md); - } + } // DisconnectableImpl.Reconnector impl. @Override @@ -1402,33 +1412,33 @@ public ClassSet describeQuery(String qql, SelectionOptions opti } public InstrumentMessageSource executeQuery ( - String qql, - Parameter ... params + String qql, + Parameter ... params ) - throws CompilationException + throws CompilationException { return (executeQuery (qql, null, null, params)); } @Override public InstrumentMessageSource executeQuery ( - String qql, - SelectionOptions options, - Parameter ... params + String qql, + SelectionOptions options, + Parameter ... params ) - throws CompilationException + throws CompilationException { return (executeQuery (qql, options, null, params)); } @Override public InstrumentMessageSource executeQuery ( - String qql, - SelectionOptions options, - CharSequence [] ids, - Parameter ... params + String qql, + SelectionOptions options, + CharSequence[] ids, + Parameter ... params ) - throws CompilationException + throws CompilationException { return (executeQuery (qql, options, null, ids, Long.MIN_VALUE, params)); } @@ -1467,14 +1477,14 @@ public TickCursor select(long time, SelectionOptions options, String[] types, Id } public InstrumentMessageSource executeQuery ( - String qql, - SelectionOptions options, - TickStream [] streams, - CharSequence [] ids, - long time, - Parameter ... params + String qql, + SelectionOptions options, + TickStream [] streams, + CharSequence [] ids, + long time, + Parameter ... params ) - throws CompilationException + throws CompilationException { assertOpen(); return TickCursorClientFactory.create(this, options, time, Long.MAX_VALUE, qql, params, ids, null, getAeronContext(), streams); diff --git a/java/timebase/test/src/main/java/com/epam/deltix/qsrv/hf/tickdb/TDBRunner.java b/java/timebase/test/src/main/java/com/epam/deltix/qsrv/hf/tickdb/TDBRunner.java index 907a079c..84bf02d5 100644 --- a/java/timebase/test/src/main/java/com/epam/deltix/qsrv/hf/tickdb/TDBRunner.java +++ b/java/timebase/test/src/main/java/com/epam/deltix/qsrv/hf/tickdb/TDBRunner.java @@ -17,6 +17,7 @@ package com.epam.deltix.qsrv.hf.tickdb; import com.epam.deltix.qsrv.hf.tickdb.comm.client.TickDBClient; +import com.epam.deltix.qsrv.servlet.HomeServlet; import com.epam.deltix.qsrv.test.messages.BarMessage; import com.epam.deltix.qsrv.test.messages.BestBidOfferMessage; import com.epam.deltix.qsrv.test.messages.TradeMessage; @@ -51,6 +52,8 @@ import java.util.Iterator; import java.util.Random; +import static junit.framework.Assert.assertEquals; + public class TDBRunner { static { @@ -74,7 +77,7 @@ public class TDBRunner { public String pass = null; public TransportProperties transportProperties; public boolean useSSL; - public SSLContext sslContext; + //public SSLContext sslContext; private int port = 0; @@ -147,7 +150,7 @@ public void startup() throws Exception { this.port = server.start(); TickDBClient connection = createClient(); - connection.setSslContext(sslContext); + //connection.setSslContext(sslContext); client = connection; // // TODO: Find out what kind of workers we want to start. @@ -177,6 +180,11 @@ public TickDBClient createClient() { return result; } + public static void testHome(String host, int port, String home) throws IOException { + String value = HomeServlet.get(host, port); + assertEquals(home, value); + } + public void shutdown() throws Exception { close(); diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SSLTomcat.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SSLTomcat.java index 8207f57b..7262917a 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SSLTomcat.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SSLTomcat.java @@ -24,20 +24,21 @@ import com.epam.deltix.qsrv.hf.tickdb.comm.server.TomcatServer; import com.epam.deltix.qsrv.hf.tickdb.pub.DXTickDB; import com.epam.deltix.qsrv.hf.tickdb.pub.TickDBFactory; +import com.epam.deltix.qsrv.servlet.HomeServlet; +import com.epam.deltix.util.io.Home; import com.epam.deltix.util.io.IOUtil; import com.epam.deltix.util.io.SSLClientContextProvider; import com.epam.deltix.util.net.SSLContextProvider; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Ignore; -import org.junit.Test; +import org.junit.*; +import static com.epam.deltix.qsrv.hf.tickdb.TDBRunner.testHome; import static junit.framework.Assert.assertEquals; import org.junit.experimental.categories.Category; import com.epam.deltix.util.JUnitCategories.TickDBFast; import java.io.File; +import java.io.IOException; /** * @@ -49,6 +50,8 @@ public class Test_SSLTomcat { @BeforeClass public static void start() throws Throwable { + SSLClientContextProvider.useDynamicKeystore(true); + File tb = new File(TDBRunner.getTemporaryLocation()); QSHome.set(tb.getParent()); @@ -58,33 +61,40 @@ public static void start() throws Throwable { StartConfiguration config = StartConfiguration.create(true, false, false); SSLProperties ssl = new SSLProperties(true, false); ssl.keystoreFile = certificate.getAbsolutePath(); + ssl.keystorePass = "deltix"; config.tb.setSSLConfig(ssl); + + System.setProperty(SSLClientContextProvider.CLIENT_KEYSTORE_PROPNAME, certificate.getAbsolutePath()); + System.setProperty(SSLClientContextProvider.CLIENT_KEYSTORE_PASS_PROPNAME, "deltix"); + runner = new TDBRunner(true, true, tb.getAbsolutePath(), new TomcatServer(config)); - runner.sslContext = SSLContextProvider.createSSLContext(ssl.keystoreFile, ssl.keystorePass, false); - runner.useSSL = true; runner.startup(); } + @Test + public void testHomeServletSSL() throws Throwable { + testHome("localhost", runner.getWebPort(), new File(runner.getLocation()).getParent()); + } + @AfterClass public static void stop() throws Throwable { + SSLClientContextProvider.useDynamicKeystore(false); runner.shutdown(); runner = null; } @Test - @Ignore // TODO: 2/11/2025 @AK public void testConnectionToSSLTomcat() throws Throwable { - try (TickDBClient client = (TickDBClient) TickDBFactory.connect("localhost", runner.getPort(), false)) { - client.open(false); - assertEquals(client.isSSLEnabled(), false); - } + DXTickDB client = TickDBFactory.connect("localhost", runner.getPort(), false); + client.open(false); + Assert.assertEquals(((TickDBClient) client).isSSLEnabled(), false); + client.close(); //connect with ssl - try (TickDBClient sslClient = (TickDBClient) TickDBFactory.connect("localhost", runner.getPort(), true)) { - sslClient.setSslContext(runner.sslContext); - sslClient.open(false); - assertEquals(sslClient.isSSLEnabled(), true); - } + DXTickDB sslClient = TickDBFactory.connect("localhost", runner.getPort(), true); + sslClient.open(false); + Assert.assertEquals(((TickDBClient) sslClient).isSSLEnabled(), true); + sslClient.close(); } } \ No newline at end of file diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TomcatServer.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TomcatServer.java index e9220e1c..2020e032 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TomcatServer.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TomcatServer.java @@ -47,27 +47,6 @@ public void testHomeServlet() throws Throwable { runner.shutdown(); } - @Test - public void testHomeServletSSL() throws Throwable { - File tb = new File(TDBRunner.getTemporaryLocation()); - QSHome.set(tb.getParent()); - - File certificate = new File(tb.getParent(), "selfsigned.jks"); - IOUtil.extractResource("com/epam/deltix/cert/selfsigned.jks", certificate); - - StartConfiguration config = StartConfiguration.create(true, false, false); - SSLProperties ssl = new SSLProperties(true, false); - ssl.keystoreFile = certificate.getAbsolutePath(); - config.tb.setSSLConfig(ssl); - TDBRunner runner = new TDBRunner(true, true, new TomcatServer(config)); - runner.sslContext = SSLContextProvider.createSSLContext(ssl.keystoreFile, ssl.keystorePass, false); - runner.startup(); - - testHome("localhost", runner.getWebPort(), new File(runner.getLocation()).getParent()); - - runner.shutdown(); - } - public static void testHome(String host, int port, String home) throws IOException { String value = HomeServlet.get(host, port); assertEquals(home, value); From d06a9c7a327d775ef157abc07a44fd4656a120d8 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Thu, 14 Aug 2025 22:01:28 +0300 Subject: [PATCH 2/9] [*] comments fixed --- .../src/main/java/com/epam/deltix/util/vsocket/VSChannel.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java index 9d5352af..e4f836ee 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/util/vsocket/VSChannel.java @@ -75,7 +75,7 @@ public interface VSChannel extends Disposable { /** * @return value previously set by {@link #setTag(String)} * - * @apiNote experimental + * experimental */ @Nullable String getTag(); @@ -83,7 +83,7 @@ public interface VSChannel extends Disposable { /** * Sets an arbitrary tag that can be used for debugging purposes. It is not sent to the remote side. * - * @apiNote experimental + * experimental */ void setTag(@Nullable String tag); } \ No newline at end of file From 9fead034bb3361118dfa721387ad63e300921903 Mon Sep 17 00:00:00 2001 From: Artsiom Chmutau Date: Wed, 26 Nov 2025 13:43:13 +0300 Subject: [PATCH 3/9] Qql describe compilation error --- .../hf/tickdb/comm/server/RequestHandler.java | 7 +- .../test/qsrv/hf/tickdb/Test_TDBServer.java | 67 +++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java index 026047f6..01dd38de 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java @@ -539,7 +539,12 @@ private void doDescribeQQL(VSChannel ds) throws IOException { SelectionOptionsCodec.read(is, options, clientVersion); Parameter[] parameters = TDBProtocol.readParameters(is, clientVersion); - ClassSet set = db.describeQuery(query, options, parameters); + ClassSet set; + try { + set = db.describeQuery(query, options, parameters); + } catch (CompilationException e) { + throw new CompilationException(e.diag, e.location); + } out.writeInt(TDBProtocol.RESP_OK); TDBProtocol.writeClassSet(ds.getDataOutputStream(), set, clientVersion); diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java index d913c804..9dbbce62 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java @@ -1160,6 +1160,73 @@ public void testQueriesCache() { } } + @Test + public void testQuery() { + DXTickStream bars = getBars(); + StreamOptions options = bars.getStreamOptions(); + String name = options.name = "mybars"; + + //get schema of query + DXTickDB tickDb = runner.getTickDb(); + + ClassSet classSet = tickDb.describeQuery("select open, close from bars", new SelectionOptions()); + RecordClassDescriptor[] descriptors = Arrays.stream(classSet.getContentClasses()) + .filter(RecordClassDescriptor.class::isInstance) + .map(RecordClassDescriptor.class::cast) + .toArray(RecordClassDescriptor[]::new); + + //get or create stream + options = new StreamOptions(StreamScope.DURABLE, "testQuery", "", 1); + options.setPolymorphic(descriptors); + DXTickStream stream = tickDb.createStream(options.name, options); + + // get changes between stream and query + StreamMetaDataChange change = new SchemaAnalyzer(new SchemaMapping()).getChanges( + stream.getStreamOptions().getMetaData(), + MetaDataChange.ContentType.Polymorphic, + new RecordClassSet(descriptors), + MetaDataChange.ContentType.Polymorphic + ); + + // NOTE: query was changed, but schema of query is the same + try (InstrumentMessageSource cursor = tickDb.executeQuery("select open, close from bars where symbol == 'GOOG'", new SelectionOptions(true, false)); + TickLoader loader = stream.createLoader(new LoadingOptions(true))) { + + while (cursor.next()) { + RawMessage message = (RawMessage) cursor.getMessage(); + message.type = descriptors[0]; + loader.send(message); + } + } + + } + + @Test + public void testDescribeExceptions() { + DXTickDB tickDb = runner.getTickDb(); + try { + tickDb.describeQuery("select open, close from bars where hello", new SelectionOptions()); + Assert.fail("Exception expected"); + } catch (CompilationException e) { + // expected + } catch(Throwable t) { + Assert.fail("CompilationException expected"); + } + } + + @Test + public void testDescribeExceptions2() { + DXTickDB tickDb = runner.getTickDb(); + try { + tickDb.describeQuery("selec t open, close from bars", new SelectionOptions()); + Assert.fail("Exception expected"); + } catch (CompilationException e) { + // expected + } catch(Throwable t) { + Assert.fail("CompilationException expected"); + } + } + @Test public void test1SelectAPI() { DXTickDB db = getTickDb(); From 6b353cf556b4c453737d9ae57c3adaa3bc964148 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Wed, 26 Nov 2025 15:15:19 +0300 Subject: [PATCH 4/9] [*] fix build --- .../com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java index 9dbbce62..a2f066e5 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_TDBServer.java @@ -16,6 +16,7 @@ */ package com.epam.deltix.test.qsrv.hf.tickdb; +import com.epam.deltix.qsrv.hf.pub.RawMessage; import com.epam.deltix.qsrv.hf.pub.md.*; import com.epam.deltix.qsrv.hf.tickdb.StreamConfigurationHelper; import com.epam.deltix.qsrv.hf.tickdb.TDBRunner; @@ -30,6 +31,7 @@ import com.epam.deltix.qsrv.hf.tickdb.pub.task.StreamCopyTask; import com.epam.deltix.qsrv.hf.tickdb.schema.MetaDataChange; import com.epam.deltix.qsrv.hf.tickdb.schema.SchemaAnalyzer; +import com.epam.deltix.qsrv.hf.tickdb.schema.SchemaMapping; import com.epam.deltix.qsrv.hf.tickdb.schema.StreamMetaDataChange; import com.epam.deltix.qsrv.test.messages.AggressorSide; import com.epam.deltix.qsrv.test.messages.BarMessage; From e55ba0bfeae6ba8548061eaea40ccf44150370f3 Mon Sep 17 00:00:00 2001 From: Roman Kisel Date: Mon, 11 May 2026 14:29:42 +0300 Subject: [PATCH 5/9] validate qql for dxapi --- .../deltix/qsrv/hf/tickdb/http/QQLState.java | 4 ++++ .../qsrv/hf/tickdb/http/TimebaseServlet.java | 17 ++++++++++++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/QQLState.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/QQLState.java index c2da64ad..177f5a48 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/QQLState.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/QQLState.java @@ -26,4 +26,8 @@ public class QQLState { @XmlElement() public long errorLocation = -1; // -1 if qql is valid + + @XmlElement() + public String errorText; + } \ No newline at end of file diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/TimebaseServlet.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/TimebaseServlet.java index d00f1843..e68c7de3 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/TimebaseServlet.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/http/TimebaseServlet.java @@ -19,6 +19,7 @@ import com.epam.deltix.qsrv.hf.pub.md.Introspector; import com.epam.deltix.qsrv.hf.pub.md.RecordClassDescriptor; import com.epam.deltix.qsrv.hf.pub.md.RecordClassSet; +import com.epam.deltix.qsrv.hf.pub.md.StandardTypes; import com.epam.deltix.qsrv.hf.pub.md.json.SchemaBuilder; import com.epam.deltix.qsrv.hf.tickdb.comm.UnknownStreamException; import com.epam.deltix.qsrv.hf.tickdb.http.download.*; @@ -28,6 +29,9 @@ import com.epam.deltix.qsrv.hf.tickdb.impl.topic.TopicTransferType; import com.epam.deltix.qsrv.hf.tickdb.impl.topic.topicregistry.LoaderSubscriptionResult; import com.epam.deltix.qsrv.hf.tickdb.lang.parser.QQLParser; +import com.epam.deltix.qsrv.hf.tickdb.lang.pub.CompilerUtil; +import com.epam.deltix.qsrv.hf.tickdb.lang.pub.Expression; +import com.epam.deltix.qsrv.hf.tickdb.lang.pub.QuantQueryCompiler; import com.epam.deltix.qsrv.hf.tickdb.lang.pub.TextMap; import com.epam.deltix.qsrv.hf.tickdb.pub.DXTickDB; import com.epam.deltix.qsrv.hf.tickdb.pub.DXTickStream; @@ -119,7 +123,7 @@ protected void service(HttpServletRequest req, HttpServletResponse resp) throws HTTPProtocol.validateVersion(((XmlRequest) body).version); if (body instanceof ValidateQQLRequest) { - validateQQL((ValidateQQLRequest)body, resp); + validateQQL(db, (ValidateQQLRequest)body, resp); } else if (body instanceof CreateStreamRequest) { StreamHandler.createStream(db, (CreateStreamRequest) body, resp); } else if (body instanceof ListStreamsRequest) { @@ -367,12 +371,18 @@ private void processCreateTopicPublisher(CreateTopicPublisherRequest body, } } - static void validateQQL(ValidateQQLRequest request, HttpServletResponse response) throws IOException { + static void validateQQL(DXTickDB db, ValidateQQLRequest request, HttpServletResponse response) throws IOException { TextMap map = QQLParser.createTextMap(); QQLState state = new QQLState(); try { - QQLParser.parse(request.qql, map); + Object sx = QQLParser.parse(request.qql, map); + if (sx instanceof Expression) { + // TODO: Optimize - creating compiler takes a lot of time + QuantQueryCompiler compiler = CompilerUtil.createCompiler(db); + compiler.compile((Expression) sx, StandardTypes.CLEAN_QUERY); + } + state.tokens = new ArrayList<>(); for (com.epam.deltix.qsrv.hf.tickdb.lang.pub.Token token : map.getTokens()) @@ -380,6 +390,7 @@ static void validateQQL(ValidateQQLRequest request, HttpServletResponse resp } catch (CompilationException e) { state.errorLocation = e.location; + state.errorText = e.diag; } final ValidateQQLResponse r = new ValidateQQLResponse(state); From aeacb42a639e75939042fd78bf72042fd63365b3 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Mon, 25 May 2026 14:31:47 +0300 Subject: [PATCH 6/9] [*] support "spaces" for the "truncate" and "purge" stream operations --- .../qsrv/hf/tickdb/comm/TDBProtocol.java | 1 + .../qsrv/hf/tickdb/pub/DXTickStream.java | 2 +- .../hf/tickdb/pub/WritableTickStream.java | 8 +++ .../tickdb/comm/client/TickStreamClient.java | 32 +++++++++- .../qsrv/hf/tickdb/ui/tbshell/Selector.java | 34 ++++++++++ .../hf/tickdb/ui/tbshell/TickDBShell.java | 35 ++++++++-- .../hf/tickdb/comm/server/RequestHandler.java | 18 +++++- .../deltix/qsrv/hf/tickdb/impl/PDStream.java | 20 ++++++ .../hf/tickdb/impl/ServerStreamWrapper.java | 7 ++ .../qsrv/hf/tickdb/impl/StubTimeStream.java | 7 +- .../hf/tickdb/impl/TransientStreamImpl.java | 14 ++-- .../qsrv/hf/tickdb/Test_SpaceReading.java | 64 +++++++++++++++++-- 12 files changed, 222 insertions(+), 20 deletions(-) diff --git a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/TDBProtocol.java b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/TDBProtocol.java index a0e4b3a1..3f8e28c2 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/TDBProtocol.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/TDBProtocol.java @@ -176,6 +176,7 @@ public abstract class TDBProtocol extends SerializationUtils { public static final int REQ_DELETE_SPACES = 147; public static final int REQ_RENAME_SPACES = 148; public static final int REQ_DESCRIBE_QUERY = 149; + public static final int REQ_TRUNCATE_SPACE = 150; public static final int REQ_CREATE_STREAM = 200; public static final int REQ_DELETE_STREAM = 201; diff --git a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/DXTickStream.java b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/DXTickStream.java index 084bbaf4..368b5312 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/DXTickStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/DXTickStream.java @@ -121,7 +121,7 @@ void setFixedType ( void purge(long time); /** - * Deletes stream data in specific space that is older than a specified time + * Deletes stream data in specific space/partition that is older than a specified time * @param time Purge time in milliseconds * @param space Space to be purged */ diff --git a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/WritableTickStream.java b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/WritableTickStream.java index 07ba7aa4..7fa263c8 100644 --- a/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/WritableTickStream.java +++ b/java/timebase/api/src/main/java/com/epam/deltix/qsrv/hf/tickdb/pub/WritableTickStream.java @@ -50,6 +50,14 @@ public interface WritableTickStream extends TickStream { */ public void truncate(long time, IdentityKey... ids); + /** + * Truncates stream data for the given entities from given time uder given space + * @param time Timestamp. If time less than stream start time, then all stream data will be deleted. + * @param space Partition/space name. + * @param ids A list of entities. If unknown, all stream entities will be used. + */ + public void truncate(long time, String space, IdentityKey... ids); + /** * Deletes stream data for the given entities using specified time range * @param from start timestamp (inclusive). Time is measured in milliseconds or nanoseconds that passed since January 1, 1970 UTC. diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickStreamClient.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickStreamClient.java index cfc6b794..506f660f 100644 --- a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickStreamClient.java +++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/client/TickStreamClient.java @@ -885,6 +885,36 @@ public void truncate(long time, IdentityKey... ids) { } } + public void truncate(long time, String space, IdentityKey... ids) { + assertWritable(); + + VSChannel ds = null; + + try { + ds = connect(); + + final DataOutputStream out = ds.getDataOutputStream (); + + out.writeInt (TDBProtocol.REQ_TRUNCATE_SPACE); + out.writeUTF (key); + out.writeLong (time); + out.writeUTF (space); + TDBProtocol.writeInstrumentIdentities (ids, out); + writeLock(out); + out.flush (); + + checkResponse(ds); + + setWriteMode(false); + invalidateProperties(TickStreamProperties.TIME_RANGE); + + } catch (IOException iox) { + throw new com.epam.deltix.util.io.UncheckedIOException(iox); + } finally { + Util.close (ds); + } + } + public void clear(IdentityKey... ids) { assertWritable(); @@ -1010,7 +1040,7 @@ public void purge(long time) { } @Override - public void purge(long time, String space) { + public void purge(long time, String space) { assertSupportsStreamSpaces(); assertWritable(); diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/Selector.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/Selector.java index 6b71eab3..665c6d3e 100644 --- a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/Selector.java +++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/Selector.java @@ -102,6 +102,7 @@ public class Selector { private CharSequenceSet types = null; private boolean printJson = false; + private String[] spaces; public Selector (TickDBShell shell) { this.shell = shell; @@ -160,6 +161,18 @@ public long getEndtime (long startTime) { return endtime == TimeConstants.TIMESTAMP_UNKNOWN ? Long.MAX_VALUE : endtime; } + public Interval getTimeOffset() { + return timeOffset; + } + + public Interval getEndTimeOffset() { + return endTimeOffset; + } + + public ChannelQualityOfService getQos() { + return qos; + } + public void setDecodeRaw (boolean decodeRaw) { this.decodeRaw = decodeRaw; } @@ -270,6 +283,18 @@ boolean doSet (String option, String value) throws Exception return true; } + if (option.equalsIgnoreCase("spaces")) { + spaces = value.split ("\\s", 3); + + if (spaces.length == 0) { + spaces = null; + shell.confirm("Selecting all stream spaces"); + } else { + shell.confirm("Select next spaces: " + Arrays.toString(spaces)); + } + return true; + } + return (false); } @@ -724,6 +749,7 @@ public boolean doCommand (String key, String args, String fileId, L return (true); } + return (false); } @@ -924,6 +950,14 @@ public SelectionOptions getSelectionOptions () { return (opts); } + + String[] listSpaces() { + return spaces; + } + + String[] listSpaces(DXTickStream stream) { + return spaces != null ? spaces : stream.listSpaces(); + } public TypeLoader getTypeLoader() { MappingTypeLoader typeLoader = new MappingTypeLoader(TypeLoaderImpl.DEFAULT_INSTANCE); diff --git a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/TickDBShell.java b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/TickDBShell.java index d066cf19..17385ba3 100644 --- a/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/TickDBShell.java +++ b/java/timebase/client/src/main/java/com/epam/deltix/qsrv/hf/tickdb/ui/tbshell/TickDBShell.java @@ -42,6 +42,7 @@ import com.epam.deltix.util.lang.StringUtils; import com.epam.deltix.util.lang.Util; import com.epam.deltix.util.progress.ConsoleProgressIndicator; +import com.epam.deltix.util.time.GMT; import com.epam.deltix.util.time.TimeKeeper; import com.epam.deltix.util.time.Interval; @@ -541,9 +542,22 @@ public void send (InstrumentMessage msg) { final long ts = TimeKeeper.currentTime; - for (TickStream stream : dbmgr.getStreams ()) { - ((DXTickStream)stream).truncate(argTime); - } + String[] spaces = selector.listSpaces(); + + for (DXTickStream stream : dbmgr.getStreams ()) { + if (spaces != null) { + for (String space : spaces) { + if (space != null) { + System.out.println("Truncating stream: [" + stream.getKey() + "] + using space = (" + space + ")" + + " with time= " + GMT.formatDateTimeMillis(argTime)); + stream.truncate(argTime, space); + } + } + } else { + System.out.println("Truncating stream: [" + stream.getKey() + "] with time= " + GMT.formatDateTimeMillis(argTime)); + stream.truncate(argTime); + } + } if (!Util.QUIET) System.out.println("total time (ms): " + (TimeKeeper.currentTime - ts)); @@ -562,8 +576,21 @@ public void send (InstrumentMessage msg) { final long ts = TimeKeeper.currentTime; + String[] spaces = selector.listSpaces(); + for (DXTickStream stream : dbmgr.getStreams ()) { - stream.purge(argTime); // now synchronous + if (spaces != null) { + for (String space : spaces) { + if (space != null) { + System.out.println("Purge stream: [" + stream.getKey() + "] + using space = (" + space + ")" + + " with time= " + GMT.formatDateTimeMillis(argTime)); + stream.purge(argTime, space); // now synchronous + } + } + } else { + System.out.println("Purge stream: [" + stream.getKey() + "] with time= " + GMT.formatDateTimeMillis(argTime)); + stream.purge(argTime); // now synchronous + } } if (!Util.QUIET) diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java index 01dd38de..453646c1 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/comm/server/RequestHandler.java @@ -239,6 +239,7 @@ public void run () throws InterruptedException { case TDBProtocol.REQ_LIST_ENTITIES: doListEntities (ds); break; case TDBProtocol.REQ_CLEAR_DATA: doClearData(ds); break; case TDBProtocol.REQ_TRUNCATE_DATA: doTruncateData(ds); break; + case TDBProtocol.REQ_TRUNCATE_SPACE: doTruncateSpace(ds); break; case TDBProtocol.REQ_PURGE_STREAM: doPurge(ds); break; case TDBProtocol.REQ_PURGE_STREAM_SPACE: doPurgeSpace(ds); break; case TDBProtocol.REQ_DELETE_RANGE: doDeleteStreamRange(ds); break; @@ -922,6 +923,22 @@ private void doTruncateData(VSChannel ds) throws IOException { out.flush(); } + private void doTruncateSpace(VSChannel ds) throws IOException { + final DXTickStream stream = getStream (ds); + + DataInputStream din = ds.getDataInputStream(); + + long time = din.readLong(); + String space = din.readUTF(); + final IdentityKey [] ids = TDBProtocol.readInstrumentIdentities (din); + + ((LockVerifier) stream).checkExclusiveWrite(readLock(ds)); + + stream.truncate(time, space, ids); + out.writeInt (TDBProtocol.RESP_OK); + out.flush(); + } + private void doRunTransformation(VSChannel ds) throws IOException { @@ -952,7 +969,6 @@ private void doPurgeSpace(VSChannel ds) throws IOException { long time = dis.readLong(); String space = dis.readUTF(); - ServerLock lock = readLock(ds); // do not verify lock for the purge //stream.verify(readLock(ds), LockType.WRITE); diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/PDStream.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/PDStream.java index f660b30c..0c0cae6a 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/PDStream.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/PDStream.java @@ -608,6 +608,7 @@ public void purge(long time, String space) { time = time - 1; String key = getKey(); + long startTime = System.currentTimeMillis(); LOGGER.debug("Purge for stream %s space %s using time=%s started ... ").with(key).with(space).with(GMT.formatDateTimeMillis(time)); try { @@ -619,6 +620,25 @@ public void purge(long time, String space) { } } + @Override + public void truncate(long timestamp, String space, IdentityKey... ids) { + long time = TimeStamp.getNanoTime(timestamp); + + long startTime = System.currentTimeMillis(); + boolean changed = false; + try { + TSRoot tsr = getRootBySpaceName(space); + if (deleteInternal(tsr, time, Long.MAX_VALUE, ids)) + changed = true; + } finally { + long endTime = System.currentTimeMillis(); + LOGGER.debug("Truncate for stream %s[space= %s] was finished in %s ms").with(getKey()).with(space).with(endTime - startTime); + } + + if (changed) + onStreamTruncated(time, ids.length != 0 ? ids : listEntities()); + } + @Override MessageChannel createChannel(InstrumentMessage msg, LoadingOptions options) { diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/ServerStreamWrapper.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/ServerStreamWrapper.java index 372dfd5b..90eb3de0 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/ServerStreamWrapper.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/ServerStreamWrapper.java @@ -76,6 +76,13 @@ public void truncate(long time, IdentityKey... ids) { delegate.truncate(time, ids); } + @Override + public void truncate(long time, String space, IdentityKey... ids) { + context.checkWritable(this); + + delegate.truncate(time, space, ids); + } + @Override public void delete(TimeStamp from, TimeStamp to, IdentityKey... ids) { context.checkWritable(this); diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/StubTimeStream.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/StubTimeStream.java index 0b670cf4..7ec02e3a 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/StubTimeStream.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/StubTimeStream.java @@ -77,7 +77,12 @@ public TickLoader createLoader(LoadingOptions options) { @Override public void truncate(long time, IdentityKey... ids) { - throw new UnsupportedOperationException(); + throw new UnsupportedOperationException("truncate"); + } + + @Override + public void truncate(long time, String space, IdentityKey... ids) { + throw new UnsupportedOperationException("truncate"); } @Override diff --git a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/TransientStreamImpl.java b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/TransientStreamImpl.java index 6042c21e..ce0e9224 100644 --- a/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/TransientStreamImpl.java +++ b/java/timebase/server/src/main/java/com/epam/deltix/qsrv/hf/tickdb/impl/TransientStreamImpl.java @@ -245,22 +245,26 @@ MessageChannel createChannel( @Override public void truncate(long time, IdentityKey... ids) { - notSupported(); + notSupported("truncate"); } + @Override + public void truncate(long time, String space, IdentityKey... ids) { + notSupported("truncate"); + } @Override public void deleteSpaces(String... names) { - notSupported(); + notSupported("deleteSpaces"); } @Override public void renameSpace(String newName, String oldName) { - notSupported(); + notSupported("renameSpace"); } - private void notSupported() { - throw new UnsupportedOperationException("Not supported for TRANSIENT streams"); + private void notSupported(String operation) { + throw new UnsupportedOperationException(operation + " is not supported for TRANSIENT streams"); } @Override diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java index 839bb8e2..e7174c4a 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java @@ -94,11 +94,11 @@ public void testSpaceReadingAndPurging() throws Exception { Assert.assertArrayEquals(new String[]{""}, stream.listSpaces()); String space1 = "spaceOne"; - writeBars(space1, stream, 100); + writeBars(space1, stream, 100, "MSFT", "IBM"); assertEquals(asSet("", space1), asSet(stream.listSpaces())); String space2 = "space2_tWo"; - writeBars(space2, stream, 200); + writeBars(space2, stream, 200, "ORCL", "GOOG"); assertEquals(asSet("", space1, space2), asSet(stream.listSpaces())); int count1 = countMessages(stream, space1); @@ -114,8 +114,8 @@ public void testSpaceReadingAndPurging() throws Exception { Assert.assertArrayEquals(new IdentityKey[] { - new ConstantIdentityKey("MSFT"), - new ConstantIdentityKey("IBM") + new ConstantIdentityKey("GOOG"), + new ConstantIdentityKey("ORCL") }, stream.listEntities(space2)); @@ -127,7 +127,57 @@ public void testSpaceReadingAndPurging() throws Exception { stream.purge(Long.MAX_VALUE, space); } - stream.delete(); + assertEquals(0, countMessages(stream, null)); + } + + @Test + public void testSpaceReadingAndTruncate() throws Exception { + String name = "testSpacesStream1"; + + StreamOptions options = new StreamOptions(StreamScope.DURABLE, name, null, 0); + options.version = "5.0"; + options.setFixedType(StreamConfigurationHelper.mkUniversalBarMessageDescriptor()); + + final DXTickStream stream = runner.getTickDb().createStream(name, options); + + Assert.assertArrayEquals(new String[]{""}, stream.listSpaces()); + + String space1 = "space1"; + writeBars(space1, stream, 100, "MSFT", "IBM"); + assertEquals(asSet("", space1), asSet(stream.listSpaces())); + + String space2 = "space2"; + writeBars(space2, stream, 200, "ORCL", "GOOG"); + assertEquals(asSet("", space1, space2), asSet(stream.listSpaces())); + + int count1 = countMessages(stream, space1); + assertEquals(100, count1); + + int count2 = countMessages(stream, space2); + assertEquals(200, count2); + + Assert.assertArrayEquals(new IdentityKey[] + { + new ConstantIdentityKey("MSFT"), + new ConstantIdentityKey("IBM") + }, stream.listEntities(space1)); + + Assert.assertArrayEquals(new IdentityKey[] + { + new ConstantIdentityKey("GOOG"), + new ConstantIdentityKey("ORCL") + }, stream.listEntities(space2)); + + + Assert.assertNotNull(stream.getTimeRange(space1)); + Assert.assertNotNull(stream.getTimeRange(space2)); + Assert.assertNull(stream.getTimeRange("")); + + stream.truncate(Long.MIN_VALUE, space2); + + int result = countMessages(stream, null); + + assertEquals(100, result); } @NotNull @@ -135,7 +185,7 @@ private HashSet asSet(String... spaces) { return new HashSet<>(Arrays.asList(spaces)); } - private void writeBars(String space, DXTickStream stream, int count) { + private void writeBars(String space, DXTickStream stream, int count, String ... symbols) { LoadingOptions o1 = new LoadingOptions(); o1.space = space; o1.writeMode = LoadingOptions.WriteMode.APPEND; @@ -151,7 +201,7 @@ public void onError(LoadingError e) { loader.addEventListener(listener); TDBRunner.BarsGenerator gn = - new TDBRunner.BarsGenerator(null, (int) BarMessage.BAR_MINUTE, count, "MSFT", "IBM"); + new TDBRunner.BarsGenerator(null, (int) BarMessage.BAR_MINUTE, count, symbols); while (gn.next()) loader.send(gn.getMessage()); From 8141c0fb9801fbfa8f8d2c92ff3b882f493461bb Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Mon, 25 May 2026 15:31:46 +0300 Subject: [PATCH 7/9] [*] tests --- .github/workflows/build.yml | 4 ++-- .github/workflows/release.yml | 8 ++++---- .../deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java | 2 +- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 952a83ad..e61299c7 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -38,7 +38,7 @@ jobs: path: ~/.gradle/wrapper key: ${{ runner.os }}-gradle-wrapper-${{ hashFiles('**/gradle/wrapper/gradle-wrapper.properties') }} - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - name: Build with gradle @@ -97,7 +97,7 @@ jobs: path: ~/.gradle/wrapper key: ${{ runner.os }}-gradle-wrapper-${{ hashFiles('**/gradle/wrapper/gradle-wrapper.properties') }} - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - name: Build with gradle diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 646c14fb..d182ca2f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -60,7 +60,7 @@ jobs: git fetch git checkout -b workflow-$GITHUB_RUN_ID origin/workflow-$GITHUB_RUN_ID~1 - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - name: Build with gradle @@ -88,7 +88,7 @@ jobs: git fetch git checkout -b workflow-$env:GITHUB_RUN_ID origin/workflow-$env:GITHUB_RUN_ID~1 - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - name: Build with gradle @@ -145,7 +145,7 @@ jobs: git fetch git checkout -b workflow-$GITHUB_RUN_ID origin/workflow-$GITHUB_RUN_ID~1 - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: 11 - name: Publish jars @@ -177,7 +177,7 @@ jobs: git fetch git checkout -b workflow-$GITHUB_RUN_ID origin/workflow-$GITHUB_RUN_ID~1 - name: Setup java - uses: actions/setup-java@v1 + uses: actions/setup-java@v2 with: java-version: 11 - name: Publish docker diff --git a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java index e7174c4a..4884dda7 100644 --- a/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java +++ b/java/timebase/test/src/test/java/com/epam/deltix/test/qsrv/hf/tickdb/Test_SpaceReading.java @@ -132,7 +132,7 @@ public void testSpaceReadingAndPurging() throws Exception { @Test public void testSpaceReadingAndTruncate() throws Exception { - String name = "testSpacesStream1"; + String name = "test-spaces-truncate"; StreamOptions options = new StreamOptions(StreamScope.DURABLE, name, null, 0); options.version = "5.0"; From ce33e2f0e075b20e337881dc19dc5a9f330a6427 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Mon, 25 May 2026 15:38:37 +0300 Subject: [PATCH 8/9] [*] CI --- .github/workflows/build.yml | 2 ++ .github/workflows/release.yml | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index e61299c7..a6d4cde6 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -41,6 +41,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} + distribution: "corretto" - name: Build with gradle run: ./gradlew build checkLicense - name: Build docker @@ -100,6 +101,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} + distribution: "corretto" - name: Build with gradle run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller - name: Archive Linux installer diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d182ca2f..6de4331c 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -63,6 +63,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} + distribution: "corretto" - name: Build with gradle run: ./gradlew build @@ -91,6 +92,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} + distribution: "corretto" - name: Build with gradle run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller - name: Archive Linux installer @@ -148,6 +150,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: 11 + distribution: "corretto" - name: Publish jars run: > ./gradlew :java:timebase:aerondirect:publish :java:timebase:commons:publish :java:timebase:s3:publish @@ -180,6 +183,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: 11 + distribution: "corretto" - name: Publish docker run: ./gradlew :java:timebase:server:dockerPublishImageAll :java:timebase:client:dockerPublishImageAll env: From 98747164765c4dd5500cd1b32a1cc6be1fd3aa52 Mon Sep 17 00:00:00 2001 From: Alex Karpovich Date: Mon, 25 May 2026 15:42:13 +0300 Subject: [PATCH 9/9] [*] CI --- .github/workflows/build.yml | 4 ++-- .github/workflows/release.yml | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index a6d4cde6..6626f071 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -41,7 +41,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - distribution: "corretto" + distribution: 'temurin' - name: Build with gradle run: ./gradlew build checkLicense - name: Build docker @@ -101,7 +101,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - distribution: "corretto" + distribution: 'temurin' - name: Build with gradle run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller - name: Archive Linux installer diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 6de4331c..53a3bacf 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -63,7 +63,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - distribution: "corretto" + distribution: 'temurin' - name: Build with gradle run: ./gradlew build @@ -92,7 +92,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: ${{ matrix.java }} - distribution: "corretto" + distribution: 'temurin' - name: Build with gradle run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller - name: Archive Linux installer @@ -150,7 +150,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: 11 - distribution: "corretto" + distribution: 'temurin' - name: Publish jars run: > ./gradlew :java:timebase:aerondirect:publish :java:timebase:commons:publish :java:timebase:s3:publish @@ -183,7 +183,7 @@ jobs: uses: actions/setup-java@v2 with: java-version: 11 - distribution: "corretto" + distribution: 'temurin' - name: Publish docker run: ./gradlew :java:timebase:server:dockerPublishImageAll :java:timebase:client:dockerPublishImageAll env: