Code review changes

hpcc-systems · Jan 30, 2025 · 38068d9 · 38068d9
1 parent 6a0bcc5
commit 38068d9
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 2 deletions.
diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/BinaryRecordReader.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/BinaryRecordReader.java
@@ -156,6 +156,8 @@ public class BinaryRecordReader implements IRecordReader
                                                          100000000000L, 1000000000000L, 10000000000000L, 100000000000000L, 1000000000000000L };
     private static final int[]   signMap             = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, +1, -1, +1, -1, +1, +1 };
 
+    private static final int     SLEEP_TIME_WARN_MS  = 100;
+    private static final int     SHORT_SLEEP_MS      = 1;
     private static final int     MASK_32_LOWER_HALF  = 0xffff;
     private static final int     BUFFER_GROW_SIZE    = 8192;
     private static final int     OPTIMIZED_STRING_READ_AHEAD = 32;
@@ -726,6 +728,8 @@ private void readIntoScratchBuffer(int offset, int dataLen) throws IOException
         int requiredCapacity = offset + dataLen;
         ensureScratchBufferCapacity(requiredCapacity);
 
+        int totalSleepTimeMS = 0;
+
         int position = offset;
         int bytesConsumed = 0;
         while (bytesConsumed < dataLen)
@@ -742,14 +746,20 @@ private void readIntoScratchBuffer(int offset, int dataLen) throws IOException
             {
                 try
                 {
-                    Thread.sleep(1);
+                    Thread.sleep(SHORT_SLEEP_MS);
+                    totalSleepTimeMS += SHORT_SLEEP_MS;
                 }
                 catch (InterruptedException e) {}
             }
 
             position += bytesRead;
             bytesConsumed += bytesRead;
         }
+
+        if (totalSleepTimeMS > SLEEP_TIME_WARN_MS)
+        {
+            messages.addMessage("Warning: BinaryRecordReader.readIntoScratchBuffer(): slept for more than " + SLEEP_TIME_WARN_MS + "ms");
+        }
     }
 
     /**

diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/CircularByteBuffer.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/CircularByteBuffer.java
@@ -15,6 +15,11 @@
 
 public class CircularByteBuffer
 {
+    // Previous testing found that increasing read size beyond 16MB did not improve performance.
+    // New implementation separates the two read size from the buffer size, but it wouldn't make
+    // sense to go beyond 16MB for the buffer size
+    public static final int MAX_BUFFER_SIZE = 16 * 1024 * 1024;
+
     private final byte[] buffer;
     private int readPos = 0;
     private int writePos = 0;
@@ -35,6 +40,11 @@ public CircularByteBuffer(int bufferSize) throws IllegalArgumentException
             throw new IllegalArgumentException("Buffer size must be greater than 0");
         }
 
+        if (bufferSize > MAX_BUFFER_SIZE)
+        {
+            throw new IllegalArgumentException("Buffer size must be less than " + MAX_BUFFER_SIZE);
+        }
+
         buffer = new byte[bufferSize];
     }
 

diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java
@@ -1447,7 +1447,7 @@ public int available() throws IOException
             {
                 String prefix = "RowServiceInputStream.available(), file "   + dataPart.getFileName() + " part " + dataPart.getThisPart() + " on IP " + getIP() + ":";
 
-                IOException wrappedException = new IOException(prefix + "End of input stream");
+                IOException wrappedException = new IOException(prefix + "End of input stream, streamPos: " + streamPos);
                 throw wrappedException;
             }
         }