[
https://issues.apache.org/jira/browse/SPARK-13652?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
huangyu updated SPARK-13652:
----------------------------
Description:
TransportClient is not thread safe and if it is called from multiple threads,
the messages can't be encoded and decoded correctly. Below is my code,and it
will print wrong message.
public static void main(String[] args) throws IOException, InterruptedException
{
TransportServer server = new TransportContext(new TransportConf("test",
new MapConfigProvider(new HashMap<String, String>())), new
RankHandler()).
createServer(8081, new LinkedList<TransportServerBootstrap>());
TransportContext context = new TransportContext(new
TransportConf("test",
new MapConfigProvider(new HashMap<String, String>())), new
NoOpRpcHandler(), true);
final TransportClientFactory clientFactory =
context.createClientFactory();
List<Thread> ts = new ArrayList<>();
for (int i = 0; i < 10; i++) {
ts.add(new Thread(new Runnable() {
@Override
public void run() {
for (int j = 0; j < 1000; j++) {
try {
ByteBuf buf = Unpooled.buffer(8);
buf.writeLong((long) j);
ByteBuffer byteBuffer =
clientFactory.createClient("localhost", 8081).
sendRpcSync(buf.nioBuffer(),
Long.MAX_VALUE);
long response = byteBuffer.getLong();
if (response != j) {
System.err.println("send:" + j + ",response:" +
response);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}));
ts.get(i).start();
}
for (Thread t : ts) {
t.join();
}
server.close();
}
it will print as below
send:221,response:222
send:233,response:234
send:312,response:313
send:358,response:359
...
was:
TransportClient is not thread safe and if it is called from multiple threads,
the messages can't be encoded and decoded correctly. Below is my code,and it
will print wrong message.
public static void main(String[] args) throws IOException, InterruptedException
{
TransportServer server = new TransportContext(new TransportConf("test",
new MapConfigProvider(new HashMap<String, String>())), new
RankHandler()).
createServer(8081, new LinkedList<TransportServerBootstrap>());
TransportContext context = new TransportContext(new
TransportConf("test",
new MapConfigProvider(new HashMap<String, String>())), new
NoOpRpcHandler(), true);
final TransportClientFactory clientFactory =
context.createClientFactory();
List<Thread> ts = new ArrayList<>();
for (int i = 0; i < 10; i++) {
ts.add(new Thread(new Runnable() {
@Override
public void run() {
for (int j = 0; j < 1000; j++) {
try {
ByteBuf buf = Unpooled.buffer(8);
buf.writeLong((long) j);
ByteBuffer byteBuffer =
clientFactory.createClient("localhost", 8081).
sendRpcSync(buf.nioBuffer(),
Long.MAX_VALUE);
long response = byteBuffer.getLong();
if (response != j) {
System.err.println("send:" + j + ",response:" +
response);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}));
ts.get(i).start();
}
for (Thread t : ts) {
t.join();
}
server.close();
}
> spark netty network issu
> ------------------------
>
> Key: SPARK-13652
> URL: https://issues.apache.org/jira/browse/SPARK-13652
> Project: Spark
> Issue Type: Bug
> Affects Versions: 1.5.1, 1.5.2, 1.6.0
> Reporter: huangyu
>
> TransportClient is not thread safe and if it is called from multiple threads,
> the messages can't be encoded and decoded correctly. Below is my code,and it
> will print wrong message.
> public static void main(String[] args) throws IOException,
> InterruptedException {
> TransportServer server = new TransportContext(new
> TransportConf("test",
> new MapConfigProvider(new HashMap<String, String>())), new
> RankHandler()).
> createServer(8081, new
> LinkedList<TransportServerBootstrap>());
> TransportContext context = new TransportContext(new
> TransportConf("test",
> new MapConfigProvider(new HashMap<String, String>())), new
> NoOpRpcHandler(), true);
> final TransportClientFactory clientFactory =
> context.createClientFactory();
> List<Thread> ts = new ArrayList<>();
> for (int i = 0; i < 10; i++) {
> ts.add(new Thread(new Runnable() {
> @Override
> public void run() {
> for (int j = 0; j < 1000; j++) {
> try {
> ByteBuf buf = Unpooled.buffer(8);
> buf.writeLong((long) j);
> ByteBuffer byteBuffer =
> clientFactory.createClient("localhost", 8081).
> sendRpcSync(buf.nioBuffer(),
> Long.MAX_VALUE);
> long response = byteBuffer.getLong();
> if (response != j) {
> System.err.println("send:" + j + ",response:"
> + response);
> }
> } catch (IOException e) {
> e.printStackTrace();
> }
> }
> }
> }));
> ts.get(i).start();
> }
> for (Thread t : ts) {
> t.join();
> }
> server.close();
> }
> it will print as below
> send:221,response:222
> send:233,response:234
> send:312,response:313
> send:358,response:359
> ...
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
