This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 9754616d [CELEBORN-330] fix deadlock when use the same netty channel
to receive data while other thread wait the response (#1265)
9754616d is described below
commit 9754616d79cb4fa398465310f31724d52e5cbf8e
Author: Shuang <[email protected]>
AuthorDate: Thu Feb 23 17:57:43 2023 +0800
[CELEBORN-330] fix deadlock when use the same netty channel to receive data
while other thread wait the response (#1265)
---
.../plugin/flink/readclient/RssBufferStream.java | 19 +++----------
.../common/network/server/BufferStreamManager.java | 20 ++++++++++++--
.../service/deploy/worker/FetchHandler.scala | 32 ++++++++++++++--------
3 files changed, 42 insertions(+), 29 deletions(-)
diff --git
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
index b853d43d..16cc0341 100644
---
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
+++
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
@@ -19,8 +19,6 @@ package org.apache.celeborn.plugin.flink.readclient;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -75,15 +73,10 @@ public class RssBufferStream {
clientFactory.createClient(
locations[currentLocationIndex].getHost(),
locations[currentLocationIndex].getFetchPort());
+
+ String fileName = locations[currentLocationIndex].getFileName();
OpenStreamWithCredit openBufferStream =
- new OpenStreamWithCredit(
- shuffleKey,
- locations[currentLocationIndex].getFileName(),
- subIndexStart,
- subIndexEnd,
- initialCredit);
- long timeoutMs = conf.fetchTimeoutMs();
- CountDownLatch latch = new CountDownLatch(1);
+ new OpenStreamWithCredit(shuffleKey, fileName, subIndexStart,
subIndexEnd, initialCredit);
client.sendRpc(
openBufferStream.toByteBuffer(),
new RpcResponseCallback() {
@@ -96,7 +89,7 @@ public class RssBufferStream {
mapShuffleClient
.getReadClientHandler()
.registerHandler(streamId, messageConsumer, client);
- latch.countDown();
+ logger.debug("open stream success stream id:{}, fileName: {}",
streamId, fileName);
}
@Override
@@ -104,10 +97,6 @@ public class RssBufferStream {
throw new RuntimeException("OpenStream failed.", e);
}
});
- if (!latch.await(timeoutMs, TimeUnit.MILLISECONDS)) {
- throw new IOException("Send openStream RPC failed to " +
client.getSocketAddress());
- }
- logger.debug("rssbufferstream streamid:{}", streamId);
}
public void addCredit(ReadAddCredit addCredit) {
diff --git
a/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
b/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
index 3cd41eba..ea87dffd 100644
---
a/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
+++
b/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
@@ -21,7 +21,16 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
-import java.util.*;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Random;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -97,7 +106,12 @@ public class BufferStreamManager {
}
public long registerStream(
- Channel channel, int initialCredit, int startSubIndex, int endSubIndex,
FileInfo fileInfo)
+ Consumer<Long> callback,
+ Channel channel,
+ int initialCredit,
+ int startSubIndex,
+ int endSubIndex,
+ FileInfo fileInfo)
throws IOException {
long streamId = nextStreamId.getAndIncrement();
streams.put(streamId, new StreamState(channel, fileInfo.getBufferSize()));
@@ -109,6 +123,8 @@ public class BufferStreamManager {
mapDataPartition.addStream(streamId);
addCredit(initialCredit, streamId);
servingStreams.put(streamId, mapDataPartition);
+ // response streamId to channel first
+ callback.accept(streamId);
mapDataPartition.setupDataPartitionReader(startSubIndex, endSubIndex,
streamId);
}
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
index 1504b400..98e962bb 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
@@ -18,10 +18,10 @@
package org.apache.celeborn.service.deploy.worker
import java.io.{FileNotFoundException, IOException}
-import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets
import java.util
import java.util.concurrent.atomic.AtomicBoolean
+import java.util.function.Consumer
import com.google.common.base.Throwables
import io.netty.util.concurrent.{Future, GenericFutureListener}
@@ -143,17 +143,25 @@ class FetchHandler(val conf: TransportConf) extends
BaseMessageHandler with Logg
val initialCredit =
msg.asInstanceOf[OpenStreamWithCredit].initialCredit
val startIndex = msg.asInstanceOf[OpenStreamWithCredit].startIndex
val endIndex = msg.asInstanceOf[OpenStreamWithCredit].endIndex
- val streamId =
- bufferStreamManager.registerStream(
- client.getChannel,
- initialCredit,
- startIndex,
- endIndex,
- fileInfo)
- val bufferStreamHandle = new StreamHandle(streamId, 0)
- client.getChannel.writeAndFlush(new RpcResponse(
- request.requestId,
- new NioManagedBuffer(bufferStreamHandle.toByteBuffer)))
+
+ val callback = new Consumer[java.lang.Long] {
+ override def accept(streamId: java.lang.Long): Unit = {
+ val bufferStreamHandle = new StreamHandle(streamId, 0)
+ client.getChannel.writeAndFlush(new RpcResponse(
+ request.requestId,
+ new NioManagedBuffer(bufferStreamHandle
+ .toByteBuffer)))
+ }
+ }
+
+ bufferStreamManager.registerStream(
+ callback,
+ client.getChannel,
+ initialCredit,
+ startIndex,
+ endIndex,
+ fileInfo)
+
case PartitionType.MAPGROUP =>
} catch {
case e: IOException =>
