You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by NicoK <gi...@git.apache.org> on 2018/07/06 07:19:48 UTC
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
GitHub user NicoK opened a pull request:
https://github.com/apache/flink/pull/6272
[FLINK-9755][network] forward exceptions in RemoteInputChannel#notifyBufferAvailable() to the responsible thread
## What is the purpose of the change
Exceptions in `RemoteInputChannel#notifyBufferAvailable()`, e.g. state checks, were swallowed inside `LocalBufferPool#recycle()` and neither logged nor otherwise processed and may have lead to stalling processes waiting for a notification that never comes.
Please note that this PR builds upon #6271 which also touched the unit tests which we change here. @tillrohrmann, @zhijiangW can you have a look at this cleanup of `LocalBufferPool#recycle()`?
## Brief change log
- cleanup of `LocalBufferPool#recycle()` also clarifying the contract of `BufferListener#notifyBufferAvailable()` which should recycle the given buffer (one implementation already did that; `RemoteInputChannel` did not)
- forward exceptions in `RemoteInputChannel#notifyBufferAvailable()` to the responsible channel and recycle the given buffer in that case
## Verifying this change
This change added tests and can be verified as follows:
- added `RemoteInputChannelTest#testFailureInNotifyBufferAvailable()`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **JavaDocs**
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/NicoK/flink flink-9755
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6272.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6272
----
commit 0b623b66399915d43f29245da148fed63bf940bf
Author: Nico Kruber <ni...@...>
Date: 2018-07-05T13:49:15Z
[FLINK-9766][network][tests] fix cleanup in RemoteInputChannelTest
If an assertion in the test fails and as a result the cleanup fails, in most
tests the original assertion was swallowed making it hard to debug.
Furthermore, #testConcurrentRecycleAndRelease2() does even not clean up at all
if successful.
commit 38a26a829b047ce1e50794470f979e708b0bb81f
Author: Nico Kruber <ni...@...>
Date: 2018-07-04T22:48:33Z
[FLINK-9755][network] forward exceptions in RemoteInputChannel#notifyBufferAvailable() to the responsible thread
This mainly involves state checks but previously these have only been swallowed
without re-registration or any other logging/handling. This may have lead to
some thread stalling while waiting for the notification that never came.
----
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201969455
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java ---
@@ -277,37 +277,17 @@ public void recycle(MemorySegment segment) {
// We do not know which locks have been acquired before the recycle() or are needed in the
// notification and which other threads also access them.
// -> call notifyBufferAvailable() outside of the synchronized block to avoid a deadlock (FLINK-9676)
- boolean success = false;
- boolean needMoreBuffers = false;
- try {
- needMoreBuffers = listener.notifyBufferAvailable(new NetworkBuffer(segment, this));
- success = true;
- } catch (Throwable ignored) {
- // handled below, under the lock
- }
+ // Note that in case of any exceptions notifyBufferAvailable() should recycle the buffer and
+ // therefore end up in this method again.
+ needMoreBuffers = listener.notifyBufferAvailable(new NetworkBuffer(segment, this));
- if (!success || needMoreBuffers) {
+ if (needMoreBuffers) {
synchronized (availableMemorySegments) {
if (isDestroyed) {
// cleanup tasks how they would have been done if we only had one synchronized block
- if (needMoreBuffers) {
- listener.notifyBufferDestroyed();
- }
- if (!success) {
- returnMemorySegment(segment);
- }
+ listener.notifyBufferDestroyed();
} else {
- if (needMoreBuffers) {
- registeredListeners.add(listener);
- }
- if (!success) {
- if (numberOfRequestedMemorySegments > currentPoolSize) {
- returnMemorySegment(segment);
- } else {
- availableMemorySegments.add(segment);
- availableMemorySegments.notify();
- }
- }
+ registeredListeners.add(listener);
--- End diff --
that's how it always was:
a) if the exception was caught in the `notifyBufferAvailable()` handler, `needMoreBuffers` is `false` and we expect the error handling in that method to fail the logic that the thread waiting on the listener is relying on - otherwise this thread's error handling needs to re-register.
b) if `notifyBufferAvailable()` throws (this is not allowed anymore!), previous logic also did not re-add the listener
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/6272
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201247182
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java ---
@@ -687,7 +690,65 @@ public void testFairDistributionFloatingBuffers() throws Exception {
} catch (Throwable t) {
thrown = t;
} finally {
- cleanup(networkBufferPool, null, thrown, channel1, channel2, channel3);
+ cleanup(networkBufferPool, null, null, thrown, channel1, channel2, channel3);
+ }
+ }
+
+ /**
+ * Tests that failures are propagated correctly if
+ * {@link RemoteInputChannel#notifyBufferAvailable(Buffer)} throws an exception. Also tests that
+ * a second listener will be notified in this case.
+ */
+ @Test
+ public void testFailureInNotifyBufferAvailable() throws Exception {
+ // Setup
+ final int numExclusiveBuffers = 0;
+ final int numFloatingBuffers = 1;
+ final int numTotalBuffers = numExclusiveBuffers + numFloatingBuffers;
+ final NetworkBufferPool networkBufferPool = new NetworkBufferPool(
+ numTotalBuffers, 32);
+
+ final SingleInputGate inputGate = createSingleInputGate();
+ final RemoteInputChannel successfulRemoteIC = createRemoteInputChannel(inputGate);
+ inputGate.setInputChannel(successfulRemoteIC.partitionId.getPartitionId(), successfulRemoteIC);
+
+// inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers);
--- End diff --
remove this annotated code?
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on the issue:
https://github.com/apache/flink/pull/6272
LGTM!
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/6272
rebased to solve the merge conflict (auto-solved by git though)
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201743330
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java ---
@@ -258,12 +258,12 @@ private MemorySegment requestMemorySegment(boolean isBlocking) throws Interrupte
@Override
public void recycle(MemorySegment segment) {
BufferListener listener;
+ boolean needMoreBuffers = false;
--- End diff --
why did you move this declaration up here? Inline it with `needMoreBuffers = listener.notifyBufferAvailable(...)`?
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/6272
actually, I forgot to adapt the unit test which relied on the previous order of the calls and assumed that the buffer got recycled. With the changes from the fixup, the buffer is not recycled by the failure directly but rather during error handling which will eventually release the channel's resources.
-> this is now fixed as well
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on the issue:
https://github.com/apache/flink/pull/6272
Thanks for fixing this potential bug.
It makes sense to handle exception during `notifyBufferAvailable` on listener side. Just some thoughts on my side above. :)
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201750461
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
--- End diff --
nitty nit: this folding doesn't look good and melds with if's body.
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201752234
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ recycleBuffer = false;
+ bufferQueue.addFloatingBuffer(buffer);
- bufferQueue.addFloatingBuffer(buffer);
+ if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ } else {
+ needMoreBuffers = true;
+ }
- if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- } else {
- needMoreBuffers = true;
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
+ }
}
- }
- if (unannouncedCredit.getAndAdd(1) == 0) {
- notifyCreditAvailable();
+ return needMoreBuffers;
+ } catch (Throwable t) {
+ if (recycleBuffer) {
+ buffer.recycleBuffer();
+ }
+ setError(t);
+ return false;
--- End diff --
hmmm, `return false`? Can not it deadlock? Shouldn't we return correct value here? Maybe this `try catch` should be more tight? However I don't fully understand what this try/catch is protecting us from.
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201970654
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ recycleBuffer = false;
--- End diff --
I don't think we should ignore the `checkState` since a failure there in our task is of course an implementation error but should not leak buffers for other tasks still running and working with this buffer pool.
Regarding the `buffer.recycleBuffer()`, you are right: we kind of always assume this cannot throw, but here we should maybe guard against recycling twice.
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/6272
thanks for the review - I changed the code as requested
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201750691
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ recycleBuffer = false;
--- End diff --
I don't understand this `recycleBuffer` flag. How can it ever be `true`?
- If we throw form `checkState` - that case we should just ignore.
- If we throw from `bufferQueue.getAvailableBufferSize()`? Is it possible?
- If we throw from `buffer.recycleBuffer();`? In that case for whatever reason we attempt to recycle second time?
none of those options make sense, so am I missing something?
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201744623
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java ---
@@ -27,6 +27,13 @@
/**
* Notification callback if a buffer is recycled and becomes available in buffer pool.
*
+ * <p>Note: responsibility on recycling the given buffer is transferred to this implementation,
+ * including any errors that lead to exceptions being thrown!
+ *
+ * <p><strong>BEWARE:</strong> since this may be called from outside the thread that relies on
+ * the listener's logic, any exception that occurs with this handler should be forwarded to the
+ * responsible thread for handling (the buffer pool ignores any {@link Throwable} from here!).
--- End diff --
However now `LocalBufferPool` is not ignoring exceptions, but it is forwarding them
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201744892
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java ---
@@ -277,37 +277,17 @@ public void recycle(MemorySegment segment) {
// We do not know which locks have been acquired before the recycle() or are needed in the
// notification and which other threads also access them.
// -> call notifyBufferAvailable() outside of the synchronized block to avoid a deadlock (FLINK-9676)
- boolean success = false;
- boolean needMoreBuffers = false;
- try {
- needMoreBuffers = listener.notifyBufferAvailable(new NetworkBuffer(segment, this));
- success = true;
- } catch (Throwable ignored) {
- // handled below, under the lock
- }
+ // Note that in case of any exceptions notifyBufferAvailable() should recycle the buffer and
+ // therefore end up in this method again.
+ needMoreBuffers = listener.notifyBufferAvailable(new NetworkBuffer(segment, this));
- if (!success || needMoreBuffers) {
+ if (needMoreBuffers) {
synchronized (availableMemorySegments) {
if (isDestroyed) {
// cleanup tasks how they would have been done if we only had one synchronized block
- if (needMoreBuffers) {
- listener.notifyBufferDestroyed();
- }
- if (!success) {
- returnMemorySegment(segment);
- }
+ listener.notifyBufferDestroyed();
} else {
- if (needMoreBuffers) {
- registeredListeners.add(listener);
- }
- if (!success) {
- if (numberOfRequestedMemorySegments > currentPoolSize) {
- returnMemorySegment(segment);
- } else {
- availableMemorySegments.add(segment);
- availableMemorySegments.notify();
- }
- }
+ registeredListeners.add(listener);
--- End diff --
If there is an exception being thrown, this will not be called. Is that how it supposed to be?
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201242662
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,45 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ // note: this call may fail, for better cleanup, increase the counter first
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
--- End diff --
From failure point, `notifyCreditAvailable` should be called before `bufferQueue.addFloatingBuffer`. But from another point, it seems more strict in logic to confirm buffer ready before announcing credit, otherwise the channel may receive new data before queuing this floating buffer, although it can hardly happen based on current implementation.
Another concern is that `notifyCreditAvailable` itself is thread safe. But considering handling failure easily, we place it under `synchronized` part. Currently the process of `notifyCreditAvailable` is very lightweight, so the cost can be ignored.
Maybe the above two concerns are unnecessary, and I can accept the current modifications.
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201260221
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,45 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ // note: this call may fail, for better cleanup, increase the counter first
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
--- End diff --
yes, this one is a tricky one:
Logically it should be: first adding the floating buffer, then enqueuing the channel. However, `notifyCreditAvailable()` is outside our control and could potentially fail. If it fails, we have two options:
1) revert any change done by `notifyBufferAvailable()` and freeing the buffer so that it can be directly re-used by `LocalBufferPool`
2) keep everything done by `notifyBufferAvailable()` and rely on `setError()` to clean up eventually - this means we should not recycle the buffer if it was already added to the `bufferQueue`.
Giving it a second thought, it may indeed be better to go for solution 2 and keep the logical execution as you proposed (infact, now I can't really think of a reason why I did not do that). We should be able to steer the recycling via the `recycleBuffer` as is already done in other cases.
---
[GitHub] flink issue #6272: [FLINK-9755][network] forward exceptions in RemoteInputCh...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/6272
Thanks for the review, merging...
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201292263
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,45 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ // note: this call may fail, for better cleanup, increase the counter first
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
--- End diff --
👍
---
[GitHub] flink pull request #6272: [FLINK-9755][network] forward exceptions in Remote...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/6272#discussion_r201971896
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java ---
@@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
return false;
}
- boolean needMoreBuffers = false;
- synchronized (bufferQueue) {
- checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers.");
+ boolean recycleBuffer = true;
+ try {
+ boolean needMoreBuffers = false;
+ synchronized (bufferQueue) {
+ checkState(isWaitingForFloatingBuffers,
+ "This channel should be waiting for floating buffers.");
+
+ // Important: double check the isReleased state inside synchronized block, so there is no
+ // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
+ if (isReleased.get() ||
+ bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ buffer.recycleBuffer();
+ return false;
+ }
- // Important: double check the isReleased state inside synchronized block, so there is no
- // race condition when notifyBufferAvailable and releaseAllResources running in parallel.
- if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- buffer.recycleBuffer();
- return false;
- }
+ recycleBuffer = false;
+ bufferQueue.addFloatingBuffer(buffer);
- bufferQueue.addFloatingBuffer(buffer);
+ if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) {
+ isWaitingForFloatingBuffers = false;
+ } else {
+ needMoreBuffers = true;
+ }
- if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) {
- isWaitingForFloatingBuffers = false;
- } else {
- needMoreBuffers = true;
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
+ }
}
- }
- if (unannouncedCredit.getAndAdd(1) == 0) {
- notifyCreditAvailable();
+ return needMoreBuffers;
+ } catch (Throwable t) {
+ if (recycleBuffer) {
+ buffer.recycleBuffer();
+ }
+ setError(t);
+ return false;
--- End diff --
The protection (currently) is against the `checkState` here and in `notifyCreditAvailable()` (the latter also has some external calls).
We could return `needMoreBuffers` here but I actually see this exception as if it happened in the responsible thread and therefore not continue to use this listener.
---