HPCC4J-579 prefetch thread hot loop fix

- Changed prefetch thread behavior to short sleep after each request - Added short sleep on main thread if blocked by prefetch thread for more than 10us Signed-off-by: James McMullan [email protected]
hpcc-systems · Feb 23, 2024 · 1c3b1e4 · 1c3b1e4
1 parent 961b1e1
commit 1c3b1e4
Showing 1 changed file with 41 additions and 27 deletions.
diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java
@@ -117,8 +117,9 @@ public class RowServiceInputStream extends InputStream implements IProfilable
 
     // Note: The platform may respond with more data than this if records are larger than this limit.
     private static final int         DEFAULT_MAX_READ_SIZE_KB = 4096;
-    private static final int         PREFETCH_SLEEP_MS        = 1;
+    private static final int         SHORT_SLEEP_MS           = 1;
     private static final int         LONG_WAIT_THRESHOLD_US   = 100;
+    private static final int         MAX_HOT_LOOP_NS          = 10000;
 
     // This is used to prevent the prefetch thread from hot looping when
     // the network connection is slow. The read on the socket will block until
@@ -406,7 +407,7 @@ public RowServiceInputStream(DataPartition dp, FieldDef rd, FieldDef pRd, int co
                         this.skip(avail);
 
                         // Sleep 1ms to prevent hot loop from eating CPU resources
-                        Thread.sleep(1);
+                        Thread.sleep(SHORT_SLEEP_MS);
                     }
                 }
                 catch (IOException e)
@@ -435,27 +436,24 @@ public void run()
                 {
                     while (inputStream.isClosed() == false)
                     {
-                        // If we don't have room in the buffer to fetch more data sleep
-                        if (inputStream.getRemainingBufferCapacity() <= (inputStream.bufferPrefetchThresholdKB * 1024))
+                        inputStream.prefetchData();
+
+                        // Sleep after each prefetch to prevent hot loop from eating CPU resources
+                        try
                         {
-                            try
+                            if (CompileTimeConstants.PROFILE_CODE)
+                            {
+                                long sleepTime = System.nanoTime();
+                                Thread.sleep(SHORT_SLEEP_MS);
+                                sleepTime = System.nanoTime() - sleepTime;
+                                sleepTimeNS += sleepTime;
+                            }
+                            else
                             {
-                                if (CompileTimeConstants.PROFILE_CODE)
-                                {
-                                    long sleepTime = System.nanoTime();
-                                    Thread.sleep(PREFETCH_SLEEP_MS);
-                                    sleepTime = System.nanoTime() - sleepTime;
-                                    sleepTimeNS += sleepTime;
-                                }
-                                else
-                                {
-                                    Thread.sleep(PREFETCH_SLEEP_MS);
-                                }
+                                Thread.sleep(SHORT_SLEEP_MS);
                             }
-                            catch(Exception e){}
                         }
-
-                        inputStream.prefetchData();
+                        catch(Exception e){}
                     }
                 }
             };
@@ -1340,14 +1338,34 @@ public int read() throws IOException
         }
 
         // We are waiting on a single byte so hot loop
-        long waitNS = 0;
+        long waitNS = System.nanoTime();
         try
         {
-            // Available will throw an exception when it reaches EOS and available bytes == 0
+            while (this.available() < 1)
+            {
+                long currentWaitNS = System.nanoTime() - waitNS;
+                if (currentWaitNS >= MAX_HOT_LOOP_NS)
+                {
+                    try
+                    {
+                        if (CompileTimeConstants.PROFILE_CODE)
+                        {
+                            long sleepTime = System.nanoTime();
+                            Thread.sleep(SHORT_SLEEP_MS);
+                            sleepTime = System.nanoTime() - sleepTime;
+                            sleepTimeNS += sleepTime;
+                        }
+                        else
+                        {
+                            Thread.sleep(SHORT_SLEEP_MS);
+                        }
+                    }
+                    catch(Exception e){}
+                }
+            }
+
             if (CompileTimeConstants.PROFILE_CODE)
             {
-                waitNS = System.nanoTime();
-                while (this.available() < 1) {}
                 waitNS = System.nanoTime() - waitNS;
                 waitTimeNS += waitNS;
 
@@ -1357,10 +1375,6 @@ public int read() throws IOException
                     numLongWaits++;
                 }
             }
-            else
-            {
-                while (this.available() < 1) {}
-            }
         }
         catch (IOException e)
         {