rkhachatryan commented on a change in pull request #12120:
URL: https://github.com/apache/flink/pull/12120#discussion_r425073734
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningWrapper.java
##########
@@ -82,187 +94,171 @@ public SpanningWrapper(String[] tempDirs) {
this.buffer = initialBuffer;
}
- void initializeWithPartialRecord(NonSpanningWrapper partial, int
nextRecordLength) throws IOException {
- // set the length and copy what is available to the buffer
- this.recordLength = nextRecordLength;
-
- final int numBytesChunk = partial.remaining();
-
- if (nextRecordLength > THRESHOLD_FOR_SPILLING) {
- // create a spilling channel and put the data there
- this.spillingChannel = createSpillingChannel();
-
- ByteBuffer toWrite =
partial.segment.wrap(partial.position, numBytesChunk);
- FileUtils.writeCompletely(this.spillingChannel,
toWrite);
- }
- else {
- // collect in memory
- ensureBufferCapacity(nextRecordLength);
- partial.segment.get(partial.position, buffer, 0,
numBytesChunk);
- }
-
- this.accumulatedRecordBytes = numBytesChunk;
+ /**
+ * We got the length, but we need the rest from the spanning
deserializer and need to wait for more buffers.
+ */
+ void transferFrom(NonSpanningWrapper partial, int nextRecordLength)
throws IOException {
+ updateLength(nextRecordLength);
+ accumulatedRecordBytes = isAboveSpillingThreshold() ?
spill(partial) : partial.copyTo(buffer);
+ partial.clear();
}
- void initializeWithPartialLength(NonSpanningWrapper partial) throws
IOException {
- // copy what we have to the length buffer
- partial.segment.get(partial.position, this.lengthBuffer,
partial.remaining());
+ private boolean isAboveSpillingThreshold() {
+ return recordLength > THRESHOLD_FOR_SPILLING;
}
void addNextChunkFromMemorySegment(MemorySegment segment, int offset,
int numBytes) throws IOException {
- int segmentPosition = offset;
- int segmentRemaining = numBytes;
- // check where to go. if we have a partial length, we need to
complete it first
- if (this.lengthBuffer.position() > 0) {
- int toPut = Math.min(this.lengthBuffer.remaining(),
segmentRemaining);
- segment.get(segmentPosition, this.lengthBuffer, toPut);
- // did we complete the length?
- if (this.lengthBuffer.hasRemaining()) {
- return;
- } else {
- this.recordLength = this.lengthBuffer.getInt(0);
-
- this.lengthBuffer.clear();
- segmentPosition += toPut;
- segmentRemaining -= toPut;
- if (this.recordLength > THRESHOLD_FOR_SPILLING)
{
- this.spillingChannel =
createSpillingChannel();
- } else {
- ensureBufferCapacity(this.recordLength);
- }
- }
+ Tuple2<Integer, Integer> positionAndRemaining =
readLengthIfNeeded(segment, offset, numBytes);
+ int startPos = positionAndRemaining.f0;
Review comment:
`startPosition` :-1:, but ok :)
> those things are the same
They are not: we need to use old values to compute new `leftOverLimit`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
