[*] VSocket API merge latest changes

Author: alex-karpovich

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();