zhijiangW commented on a change in pull request #10347: [FLINK-14516][network]
Remove non credit-based flow control code
URL: https://github.com/apache/flink/pull/10347#discussion_r352442350
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CachedBufferStorage.java
##########
@@ -74,49 +102,139 @@ public void add(BufferOrEvent boe) {
}
/**
- * It is never reusing resources and is defaulting to {@link
#rollOverWithoutReusingResources()}.
+ * Starts a new sequence of buffers and event reusing the same
resources and
+ * returns the current sequence of buffers for reading.
+ *
+ * @return The readable sequence of buffers and events, or 'null', if
nothing was added.
*/
- @Override
public BufferOrEventSequence rollOverReusingResources() {
return rollOverWithoutReusingResources();
}
- @Override
+ /**
+ * Starts a new sequence of buffers and event without reusing the same
resources and
+ * returns the current sequence of buffers for reading.
+ *
+ * @return The readable sequence of buffers and events, or 'null', if
nothing was added.
+ */
public BufferOrEventSequence rollOverWithoutReusingResources() {
if (bytesBlocked == 0) {
return null;
}
- CachedBufferOrEventSequence currentSequence = new
CachedBufferOrEventSequence(currentBuffers, bytesBlocked);
+ BufferOrEventSequence currentSequence = new
BufferOrEventSequence(currentBuffers, bytesBlocked);
currentBuffers = new ArrayDeque<>();
bytesBlocked = 0L;
return currentSequence;
}
@Override
- public void close() throws IOException {
+ public void close() {
BufferOrEvent boe;
while ((boe = currentBuffers.poll()) != null) {
if (boe.isBuffer()) {
boe.getBuffer().recycleBuffer();
}
}
- super.close();
+ if (currentBuffered != null) {
+ currentBuffered.cleanup();
+ }
+ for (BufferOrEventSequence seq : queuedBuffered) {
+ seq.cleanup();
+ }
+ queuedBuffered.clear();
+ rolledBytes = 0L;
}
@Override
public long getPendingBytes() {
return bytesBlocked;
}
+ @Override
+ public boolean isFull() {
+ return maxBufferedBytes > 0 && (getRolledBytes() +
getPendingBytes()) > maxBufferedBytes;
+ }
+
+ @Override
+ public void rollOver() {
+ if (currentBuffered == null) {
+ // common case: no more buffered data
+ currentBuffered = rollOverReusingResources();
+ if (currentBuffered != null) {
+ currentBuffered.open();
+ }
+ }
+ else {
+ // uncommon case: buffered data pending
+ // push back the pending data, if we have any
+ LOG.debug("{}: Checkpoint skipped via buffered data:" +
+ "Pushing back current alignment buffers and
feeding back new alignment data first.", taskName);
+
+ // since we did not fully drain the previous sequence,
we need to allocate a new buffer for this one
+ BufferOrEventSequence bufferedNow =
rollOverWithoutReusingResources();
+ if (bufferedNow != null) {
+ bufferedNow.open();
+ queuedBuffered.addFirst(currentBuffered);
+ rolledBytes += currentBuffered.size();
+ currentBuffered = bufferedNow;
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{}: Size of buffered data: {} bytes",
+ taskName,
+ currentBuffered == null ? 0L :
currentBuffered.size());
+ }
+ }
+
+ @Override
+ public long getRolledBytes() {
+ return rolledBytes;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return currentBuffered == null;
+ }
+
+ @Override
+ public Optional<BufferOrEvent> pollNext() {
+ if (currentBuffered == null) {
+ return Optional.empty();
+ }
+ // TODO: FLINK-12536 for non credit-based flow control, getNext
method is blocking
Review comment:
This `TODO` is invalid now and can be removed?
----------------------------------------------------------------
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]
With regards,
Apache Git Services
