Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

HPCC4J-606 DFSClient Allow ReadBuffer size to be set #788

Merged
merged 5 commits into from
Feb 25, 2025
Merged

HPCC4J-606 DFSClient Allow ReadBuffer size to be set #788

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
155c1a4
HPCC4J-606 DFSClient Allow ReadBuffer size to be set
jpmcmu Jan 17, 2025
75a84c5
Code review changes
jpmcmu Jan 27, 2025
6a0bcc5
Code review changes
jpmcmu Jan 27, 2025
38068d9
Code review changes
jpmcmu Jan 30, 2025
ab481f9
Code review changes
jpmcmu Feb 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 22 additions & 18 deletions dfsclient/src/main/java/org/hpccsystems/dfs/client/BinaryRecordReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ public int read() throws IOException
{
streamPos++;
}

return ret;
}

Expand Down Expand Up @@ -155,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;
Expand Down Expand Up @@ -725,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)
Expand All @@ -737,9 +742,24 @@ private void readIntoScratchBuffer(int offset, int dataLen) throws IOException
throw e;
}

if (bytesRead == 0)
{
try
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this just hoping we'll start getting data after the sleep?
Also, is it possible to get stuck in this loop if it doesn't receive any data for a long time?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could get stuck in this loop until the socket timeout kicks in. After which the call to read() will either throw an exception or return -1 which will cause an exception to be thrown. (Which case depends on how the socket was closed.)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To Rodrigo's point should add an indicator / counter if we get stuck in this loop for a while

{
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");
}
}

/**
Expand Down Expand Up @@ -1294,26 +1314,10 @@ else if ((this.scratchBuffer[strByteLen + bytesScanned] & 0xF8) == 0xF0)
// Use the second half of the remaining buffer space as a temp place to read in compressed bytes.
// Beginning of the buffer will be used to construct the string

int bytesToRead = compressedLen;
int availableBytes = 0;
try
{
availableBytes = this.inputStream.available();
}
catch(Exception e)
{
throw new IOException("Error, unexpected EOS while constructing QString.");
}

if (bytesToRead > availableBytes)
{
bytesToRead = availableBytes;
}

// Scratch buffer is divided into two parts. First expandedLen bytes are for the final expanded string
// Remaining bytes are for reading in the compressed string.
int readPos = expandedLen + compressedBytesConsumed;
readIntoScratchBuffer(readPos, bytesToRead);
readIntoScratchBuffer(readPos, compressedLen);

// We want to consume only a whole chunk so round off residual chars
// Below we will handle any residual bytes. (strLen % 4)
Expand All @@ -1334,7 +1338,7 @@ else if ((this.scratchBuffer[strByteLen + bytesScanned] & 0xF8) == 0xF0)
compressedBytesConsumed += QSTR_COMPRESSED_CHUNK_LEN;
}

compressedBytesRead += bytesToRead;
compressedBytesRead += compressedLen;
strByteLen += writePos;
}

Expand Down
Loading
Loading