Main 6.2

.github/workflows/build.yml

@@ -38,9 +38,10 @@ 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 }}
+          distribution: 'temurin'
       - name: Build with gradle
         run: ./gradlew build checkLicense
       - name: Build docker
@@ -97,9 +98,10 @@ 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 }}
+          distribution: 'temurin'
       - name: Build with gradle
         run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller
       - name: Archive Linux installer

.github/workflows/release.yml

@@ -60,9 +60,10 @@ 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 }}
+          distribution: 'temurin'
       - name: Build with gradle
         run: ./gradlew build
 
@@ -88,9 +89,10 @@ 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 }}
+          distribution: 'temurin'
       - name: Build with gradle
         run: ./gradlew build windowsTimebaseInstaller linuxTimebaseInstaller
       - name: Archive Linux installer
@@ -145,9 +147,10 @@ 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
+          distribution: 'temurin'
       - name: Publish jars
         run: >
           ./gradlew :java:timebase:aerondirect:publish :java:timebase:commons:publish :java:timebase:s3:publish
@@ -177,9 +180,10 @@ 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
+          distribution: 'temurin'
       - name: Publish docker
         run: ./gradlew :java:timebase:server:dockerPublishImageAll :java:timebase:client:dockerPublishImageAll
         env:

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;

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
      */

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.

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<Entry>  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,65 +90,124 @@ public void                 shutdown () {
     }
 
     public void                 addChannel(VSChannel channel) {
+        assert channel != null;
+
+        if (channel == null)
+            return;
+
         synchronized (channels) {
             channels.linkLast(new Entry(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<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();