AHeise commented on a change in pull request #13499:
URL: https://github.com/apache/flink/pull/13499#discussion_r496477139
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
##########
@@ -322,23 +350,71 @@ private MemorySegment requestMemorySegment() {
return requestMemorySegment(UNKNOWN_CHANNEL);
}
- @Nullable
- private MemorySegment requestMemorySegmentFromGlobal() {
- assert Thread.holdsLock(availableMemorySegments);
+ private boolean requestMemorySegmentFromGlobal() {
+ if (numberOfRequestedMemorySegments >= currentPoolSize) {
+ return false;
+ }
+
+ MemorySegment segment =
networkBufferPool.requestMemorySegment();
+ if (segment != null) {
+ availableMemorySegments.add(segment);
+ numberOfRequestedMemorySegments++;
+ return true;
+ }
+ return false;
+ }
- if (isDestroyed) {
- throw new IllegalStateException("Buffer pool is
destroyed.");
+ /**
+ * Tries to obtain a buffer from global pool as soon as one pool is
available. Note that multiple
+ * {@link LocalBufferPool}s might wait on the future of the global
pool, hence this method double-check if a new
+ * buffer is really needed at the time it becomes available.
+ */
+ private void eagerlyRequestMemorySegmentFromGlobal() {
+ if (eagerlyRequesting) {
+ return;
}
+ eagerlyRequesting = true;
+ networkBufferPool.getAvailableFuture().thenRun(() -> {
+ eagerlyRequesting = false;
+ if (availabilityHelper.isAvailable()) {
+ // there is currently no benefit for this pool
to obtain buffer from global; give other pools precedent
+ return;
+ }
Review comment:
I was trying to optimize, but I guess it's premature. Segment
reallocation is rather rare anyways afaik.
----------------------------------------------------------------
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]
