You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by nk...@apache.org on 2018/08/30 20:58:16 UTC
[flink] branch release-1.6 updated (3f6e0b7 -> 4cf197e)
This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a change to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 3f6e0b7 [FLINK-9962] allow users to specify TimeZone in DateTimeBucketer
new f574a8f [hotfix][network] simplify moreAvailable/wasEmpty logic
new b3dadd6 [FLINK-10141][network] move notifications outside the bufferQueue and receivedBuffers locks
new 4cf197e [FLINK-10142][network] reduce locking around credit notification
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../io/network/netty/PartitionRequestClient.java | 5 +-
.../partition/consumer/RemoteInputChannel.java | 80 +++++++++++-----------
.../partition/consumer/SingleInputGate.java | 2 +-
.../partition/consumer/RemoteInputChannelTest.java | 66 +++++++++++-------
4 files changed, 84 insertions(+), 69 deletions(-)
[flink] 03/03: [FLINK-10142][network] reduce locking around credit
notification
Posted by nk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4cf197ede67dee9c4fbb41a4c5a8a61b40ddfa5d
Author: Nico Kruber <ni...@data-artisans.com>
AuthorDate: Sun Aug 5 00:41:32 2018 +0200
[FLINK-10142][network] reduce locking around credit notification
This closes #6555.
---
.../io/network/netty/PartitionRequestClient.java | 5 +-
.../partition/consumer/RemoteInputChannel.java | 12 +---
.../partition/consumer/RemoteInputChannelTest.java | 66 ++++++++++++++--------
3 files changed, 43 insertions(+), 40 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
index 27d341a..9c9deaa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClient.java
@@ -171,10 +171,7 @@ public class PartitionRequestClient {
}
public void notifyCreditAvailable(RemoteInputChannel inputChannel) {
- // We should skip the notification if the client is already closed.
- if (!closeReferenceCounter.isDisposed()) {
- clientHandler.notifyCreditAvailable(inputChannel);
- }
+ clientHandler.notifyCreditAvailable(inputChannel);
}
public void close(RemoteInputChannel inputChannel) throws IOException {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index c4954c0..6738abd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -289,10 +289,7 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
private void notifyCreditAvailable() {
checkState(partitionRequestClient != null, "Tried to send task event to producer before requesting a queue.");
- // We should skip the notification if this channel is already released.
- if (!isReleased.get()) {
- partitionRequestClient.notifyCreditAvailable(this);
- }
+ partitionRequestClient.notifyCreditAvailable(this);
}
/**
@@ -354,13 +351,6 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
*/
@Override
public boolean notifyBufferAvailable(Buffer buffer) {
- // Check the isReleased state outside synchronized block first to avoid
- // deadlock with releaseAllResources running in parallel.
- if (isReleased.get()) {
- buffer.recycleBuffer();
- return false;
- }
-
boolean recycleBuffer = true;
try {
boolean needMoreBuffers = false;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
index 6305492..9141b36 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
@@ -106,10 +106,33 @@ public class RemoteInputChannelTest {
@Test
public void testConcurrentOnBufferAndRelease() throws Exception {
- // Config
- // Repeatedly spawn two tasks: one to queue buffers and the other to release the channel
- // concurrently. We do this repeatedly to provoke races.
- final int numberOfRepetitions = 8192;
+ testConcurrentReleaseAndSomething(8192, (inputChannel, buffer, j) -> {
+ inputChannel.onBuffer(buffer, j, -1);
+ return null;
+ });
+ }
+
+ @Test
+ public void testConcurrentNotifyBufferAvailableAndRelease() throws Exception {
+ testConcurrentReleaseAndSomething(1024, (inputChannel, buffer, j) ->
+ inputChannel.notifyBufferAvailable(buffer)
+ );
+ }
+
+ private interface TriFunction<T, U, V, R> {
+ R apply(T t, U u, V v) throws Exception;
+ }
+
+ /**
+ * Repeatedly spawns two tasks: one to call <tt>function</tt> and the other to release the
+ * channel concurrently. We do this repeatedly to provoke races.
+ *
+ * @param numberOfRepetitions how often to repeat the test
+ * @param function function to call concurrently to {@link RemoteInputChannel#releaseAllResources()}
+ */
+ private void testConcurrentReleaseAndSomething(
+ final int numberOfRepetitions,
+ TriFunction<RemoteInputChannel, Buffer, Integer, Object> function) throws Exception {
// Setup
final ExecutorService executor = Executors.newFixedThreadPool(2);
@@ -122,30 +145,23 @@ public class RemoteInputChannelTest {
for (int i = 0; i < numberOfRepetitions; i++) {
final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate);
- final Callable<Void> enqueueTask = new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- while (true) {
- for (int j = 0; j < 128; j++) {
- // this is the same buffer over and over again which will be
- // recycled by the RemoteInputChannel
- inputChannel.onBuffer(buffer.retainBuffer(), j, -1);
- }
+ final Callable<Void> enqueueTask = () -> {
+ while (true) {
+ for (int j = 0; j < 128; j++) {
+ // this is the same buffer over and over again which will be
+ // recycled by the RemoteInputChannel
+ function.apply(inputChannel, buffer.retainBuffer(), j);
+ }
- if (inputChannel.isReleased()) {
- return null;
- }
+ if (inputChannel.isReleased()) {
+ return null;
}
}
};
- final Callable<Void> releaseTask = new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- inputChannel.releaseAllResources();
-
- return null;
- }
+ final Callable<Void> releaseTask = () -> {
+ inputChannel.releaseAllResources();
+ return null;
};
// Submit tasks and wait to finish
@@ -158,8 +174,8 @@ public class RemoteInputChannelTest {
result.get();
}
- assertEquals("Resource leak during concurrent release and enqueue.",
- 0, inputChannel.getNumberOfQueuedBuffers());
+ assertEquals("Resource leak during concurrent release and notifyBufferAvailable.",
+ 0, inputChannel.getNumberOfQueuedBuffers());
}
}
finally {
[flink] 02/03: [FLINK-10141][network] move notifications outside
the bufferQueue and receivedBuffers locks
Posted by nk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b3dadd6fe3fa3e163ec9e25bff52c642e91f683f
Author: Nico Kruber <ni...@data-artisans.com>
AuthorDate: Tue Aug 21 17:34:49 2018 +0200
[FLINK-10141][network] move notifications outside the bufferQueue and receivedBuffers locks
This means, notifyCreditAvailable() and notifyChannelNonEmpty() are called
without any locks being acquired. The change is safe since these callbacks do
their own synchronization already (if needed).
This closes #6553.
---
.../partition/consumer/RemoteInputChannel.java | 62 +++++++++++++---------
1 file changed, 37 insertions(+), 25 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 79d25c6..c4954c0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -306,8 +306,8 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
int numAddedBuffers;
synchronized (bufferQueue) {
- // Important: check the isReleased state inside synchronized block, so there is no
- // race condition when recycle and releaseAllResources running in parallel.
+ // Similar to notifyBufferAvailable(), make sure that we never add a buffer
+ // after releaseAllResources() released all buffers (see below for details).
if (isReleased.get()) {
try {
inputGate.returnExclusiveSegments(Collections.singletonList(segment));
@@ -368,8 +368,13 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
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.
+ // Important: make sure that we never add a buffer after releaseAllResources()
+ // released all buffers. Following scenarios exist:
+ // 1) releaseAllResources() already released buffers inside bufferQueue
+ // -> then isReleased is set correctly
+ // 2) releaseAllResources() did not yet release buffers from bufferQueue
+ // -> we may or may not have set isReleased yet but will always wait for the
+ // lock on bufferQueue to release buffers
if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) {
isWaitingForFloatingBuffers = false;
recycleBuffer = false; // just in case
@@ -385,10 +390,10 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
} else {
needMoreBuffers = true;
}
+ }
- if (unannouncedCredit.getAndAdd(1) == 0) {
- notifyCreditAvailable();
- }
+ if (unannouncedCredit.getAndAdd(1) == 0) {
+ notifyCreditAvailable();
}
return needMoreBuffers;
@@ -484,8 +489,8 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
int numRequestedBuffers = 0;
synchronized (bufferQueue) {
- // Important: check the isReleased state inside synchronized block, so there is no
- // race condition when onSenderBacklog and releaseAllResources running in parallel.
+ // Similar to notifyBufferAvailable(), make sure that we never add a buffer
+ // after releaseAllResources() released all buffers (see above for details).
if (isReleased.get()) {
return;
}
@@ -510,33 +515,40 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
}
public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException {
- boolean success = false;
+ boolean recycleBuffer = true;
try {
+
+ final boolean wasEmpty;
synchronized (receivedBuffers) {
- if (!isReleased.get()) {
- if (expectedSequenceNumber == sequenceNumber) {
- final boolean wasEmpty = receivedBuffers.isEmpty();
+ // Similar to notifyBufferAvailable(), make sure that we never add a buffer
+ // after releaseAllResources() released all buffers from receivedBuffers
+ // (see above for details).
+ if (isReleased.get()) {
+ return;
+ }
- receivedBuffers.add(buffer);
- expectedSequenceNumber++;
+ if (expectedSequenceNumber != sequenceNumber) {
+ onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber));
+ return;
+ }
- if (wasEmpty) {
- notifyChannelNonEmpty();
- }
+ wasEmpty = receivedBuffers.isEmpty();
+ receivedBuffers.add(buffer);
+ recycleBuffer = false;
+ }
- success = true;
- } else {
- onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber));
- }
- }
+ ++expectedSequenceNumber;
+
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
}
- if (success && backlog >= 0) {
+ if (backlog >= 0) {
onSenderBacklog(backlog);
}
} finally {
- if (!success) {
+ if (recycleBuffer) {
buffer.recycleBuffer();
}
}
[flink] 01/03: [hotfix][network] simplify moreAvailable/wasEmpty
logic
Posted by nk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f574a8fc1fafb761983d7a04a72cd6181fe74276
Author: Nico Kruber <ni...@data-artisans.com>
AuthorDate: Tue Aug 21 17:16:29 2018 +0200
[hotfix][network] simplify moreAvailable/wasEmpty logic
If we only need the status of a queue being empty or not, we do not need to
acquire the size.
---
.../io/network/partition/consumer/RemoteInputChannel.java | 10 +++++-----
.../runtime/io/network/partition/consumer/SingleInputGate.java | 2 +-
2 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 28f3020..79d25c6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -191,16 +191,16 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
checkError();
final Buffer next;
- final int remaining;
+ final boolean moreAvailable;
synchronized (receivedBuffers) {
next = receivedBuffers.poll();
- remaining = receivedBuffers.size();
+ moreAvailable = !receivedBuffers.isEmpty();
}
numBytesIn.inc(next.getSizeUnsafe());
numBuffersIn.inc();
- return Optional.of(new BufferAndAvailability(next, remaining > 0, getSenderBacklog()));
+ return Optional.of(new BufferAndAvailability(next, moreAvailable, getSenderBacklog()));
}
// ------------------------------------------------------------------------
@@ -516,12 +516,12 @@ public class RemoteInputChannel extends InputChannel implements BufferRecycler,
synchronized (receivedBuffers) {
if (!isReleased.get()) {
if (expectedSequenceNumber == sequenceNumber) {
- int available = receivedBuffers.size();
+ final boolean wasEmpty = receivedBuffers.isEmpty();
receivedBuffers.add(buffer);
expectedSequenceNumber++;
- if (available == 0) {
+ if (wasEmpty) {
notifyChannelNonEmpty();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index 2e7d076..f51dc74 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -545,7 +545,7 @@ public class SingleInputGate implements InputGate {
currentChannel = inputChannelsWithData.remove();
enqueuedInputChannelsWithData.clear(currentChannel.getChannelIndex());
- moreAvailable = inputChannelsWithData.size() > 0;
+ moreAvailable = !inputChannelsWithData.isEmpty();
}
result = currentChannel.getNextBuffer();