Skip to content

Commit 4d59b9d

Browse files
patrickstuedipatrickstuedi
committed
Read from client socket when stuck on a write to avoid blocking server
- #2
1 parent e358ec7 commit 4d59b9d

8 files changed

+136
-137
lines changed

‎pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
<groupId>com.ibm.narpc</groupId>
55
<artifactId>narpc</artifactId>
66
<packaging>jar</packaging>
7-
<version>1.4</version>
7+
<version>1.5</version>
88
<name>narpc</name>
99

1010
<url>https://github.com/patrickstuedi/narpc</url>

‎src/main/java/com/ibm/narpc/NaRPCChannel.java

-77
This file was deleted.

‎src/main/java/com/ibm/narpc/NaRPCDispatcher.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -81,10 +81,10 @@ public void run() {
8181
}
8282
if (key.isReadable()) {
8383
NaRPCServerChannel channel = (NaRPCServerChannel) key.attachment();
84-
long ticket = channel.fetch(request);
84+
long ticket = channel.receiveMessage(request);
8585
if(ticket > 0){
8686
T response = service.processRequest(request);
87-
channel.transmit(ticket, response);
87+
channel.transmitMessage(ticket, response);
8888
} else if (ticket < 0){
8989
LOG.info("closing channel " + channel.address());
9090
this.service.removeEndpoint(channel);

‎src/main/java/com/ibm/narpc/NaRPCEndpoint.java

+68-43
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323

2424
import java.io.IOException;
2525
import java.net.InetSocketAddress;
26+
import java.net.StandardSocketOptions;
2627
import java.nio.ByteBuffer;
2728
import java.nio.channels.SocketChannel;
2829
import java.util.concurrent.ArrayBlockingQueue;
@@ -31,7 +32,7 @@
3132
import java.util.concurrent.atomic.AtomicLong;
3233
import java.util.concurrent.locks.ReentrantLock;
3334

34-
public class NaRPCEndpoint<R extends NaRPCMessage, T extends NaRPCMessage> extends NaRPCChannel {
35+
public class NaRPCEndpoint<R extends NaRPCMessage, T extends NaRPCMessage> {
3536
private NaRPCGroup group;
3637
private ConcurrentHashMap<Long, NaRPCFuture<R,T>> pendingRPCs;
3738
private ArrayBlockingQueue<ByteBuffer> bufferQueue;
@@ -43,81 +44,105 @@ public class NaRPCEndpoint<R extends NaRPCMessage, T extends NaRPCMessage> exten
4344
public NaRPCEndpoint(NaRPCGroup group, SocketChannel channel) throws Exception {
4445
this.group = group;
4546
this.channel = channel;
47+
this.channel.setOption(StandardSocketOptions.TCP_NODELAY, group.isNodelay());
48+
this.channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
4649
this.pendingRPCs = new ConcurrentHashMap<Long, NaRPCFuture<R,T>>();
47-
this.readLock = new ReentrantLock();
48-
this.writeLock = new ReentrantLock();
4950
this.bufferQueue = new ArrayBlockingQueue<ByteBuffer>(group.getQueueDepth());
5051
for (int i = 0; i < group.getQueueDepth(); i++){
5152
ByteBuffer buffer = ByteBuffer.allocate(group.getMessageSize());
5253
bufferQueue.put(buffer);
5354
}
5455
this.sequencer = new AtomicLong(1);
56+
this.readLock = new ReentrantLock();
57+
this.writeLock = new ReentrantLock();
58+
}
59+
60+
public void connect(InetSocketAddress address) throws IOException {
61+
this.channel.connect(address);
62+
this.channel.configureBlocking(false);
63+
}
64+
65+
public void close() throws IOException{
66+
this.channel.close();
5567
}
5668

5769
public NaRPCFuture<R,T> issueRequest(R request, T response) throws IOException {
5870
ByteBuffer buffer = getBuffer();
71+
while(buffer == null){
72+
buffer = getBuffer();
73+
}
5974
long ticket = sequencer.getAndIncrement();
60-
makeMessage(ticket, request, buffer);
75+
NaRPCProtocol.makeMessage(ticket, request, buffer);
6176
NaRPCFuture<R,T> future = new NaRPCFuture<R,T>(this, request, response, ticket);
6277
pendingRPCs.put(ticket, future);
63-
while(!tryTransmitting(buffer)){
78+
79+
boolean wlocked = writeLock.tryLock();
80+
if (wlocked){
81+
channel.write(buffer);
82+
while(buffer.hasRemaining()){
83+
pollResponse();
84+
channel.write(buffer);
85+
}
86+
writeLock.unlock();
6487
}
88+
6589
putBuffer(buffer);
6690
return future;
6791
}
68-
69-
public void pollResponse(AtomicBoolean done) throws IOException {
92+
93+
void pollResponse() throws IOException {
7094
ByteBuffer buffer = getBuffer();
71-
boolean locked = readLock.tryLock();
72-
if (locked) {
73-
if (!done.get()){
74-
long ticket = fetchBuffer(channel, buffer);
75-
if (ticket < 0){
76-
throw new IOException("Got invalid ticket, connection closed? " + ticket);
77-
}
78-
NaRPCFuture<R,T> future = pendingRPCs.remove(ticket);
79-
future.getResponse().update(buffer);
80-
future.signal();
81-
}
95+
if (buffer == null){
96+
return;
97+
}
98+
boolean rlocked = readLock.tryLock();
99+
long ticket = 0;
100+
if (rlocked){
101+
ticket = NaRPCProtocol.fetchBuffer(channel, buffer);
82102
readLock.unlock();
83-
}
103+
}
104+
if (ticket > 0){
105+
NaRPCFuture<R, T> future = pendingRPCs.remove(ticket);
106+
future.getResponse().update(buffer);
107+
future.signal();
108+
}
84109
putBuffer(buffer);
85110
}
86-
87-
public void connect(InetSocketAddress address) throws IOException {
88-
this.channel.connect(address);
89-
this.channel.socket().setTcpNoDelay(group.isNodelay());
90-
this.channel.socket().setReuseAddress(true);
91-
this.channel.configureBlocking(false);
92-
}
93-
94-
public void close() throws IOException{
95-
this.channel.close();
96-
}
97111

98112
public String address() throws IOException {
99113
return channel.getRemoteAddress().toString();
100114
}
101115

102-
private boolean tryTransmitting(ByteBuffer buffer) throws IOException{
103-
boolean locked = writeLock.tryLock();
104-
if (locked) {
105-
transmitMessage(channel, buffer);
106-
writeLock.unlock();
107-
return true;
108-
}
109-
return false;
110-
}
111-
112116
private ByteBuffer getBuffer(){
113117
ByteBuffer buffer = bufferQueue.poll();
114-
while(buffer == null){
115-
buffer = bufferQueue.poll();
116-
}
117118
return buffer;
118119
}
119120

120121
private void putBuffer(ByteBuffer buffer){
121122
bufferQueue.add(buffer);
122123
}
124+
125+
//----------- Old stuff
126+
127+
// public NaRPCFuture<R,T> issueRequestOld(R request, T response) throws IOException {
128+
// ByteBuffer buffer = getBuffer();
129+
// long ticket = sequencer.getAndIncrement();
130+
// makeMessage(ticket, request, buffer);
131+
// NaRPCFuture<R,T> future = new NaRPCFuture<R,T>(this, request, response, ticket);
132+
// pendingRPCs.put(ticket, future);
133+
// while(!tryTransmittingOld(buffer)){
134+
// }
135+
// putBuffer(buffer);
136+
// return future;
137+
// }
138+
139+
// private boolean tryTransmittingOld(ByteBuffer buffer) throws IOException{
140+
// boolean locked = writeLock.tryLock();
141+
// if (locked) {
142+
// transmitMessageOld(channel, buffer);
143+
// writeLock.unlock();
144+
// return true;
145+
// }
146+
// return false;
147+
// }
123148
}

‎src/main/java/com/ibm/narpc/NaRPCFuture.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ public boolean isCancelled() {
5656
public boolean isDone() {
5757
try {
5858
if (!done.get()){
59-
endpoint.pollResponse(done);
59+
endpoint.pollResponse();
6060
}
6161
} catch(Exception e){
6262
}
@@ -67,7 +67,7 @@ public boolean isDone() {
6767
public T get() throws InterruptedException, ExecutionException {
6868
try {
6969
while (!done.get()){
70-
endpoint.pollResponse(done);
70+
endpoint.pollResponse();
7171
}
7272
} catch(Exception e){
7373
throw new ExecutionException(e);
@@ -80,7 +80,7 @@ public T get(long timeout, TimeUnit unit) throws InterruptedException,
8080
ExecutionException, TimeoutException {
8181
try {
8282
while (!done.get()){
83-
endpoint.pollResponse(done);
83+
endpoint.pollResponse();
8484
}
8585
} catch(Exception e){
8686
throw new ExecutionException(e);

‎src/main/java/com/ibm/narpc/NaRPCProtocol.java

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
package com.ibm.narpc;
2+
3+
import org.slf4j.Logger;
4+
5+
import java.io.IOException;
6+
import java.nio.ByteBuffer;
7+
import java.nio.channels.SocketChannel;
8+
9+
public class NaRPCProtocol {
10+
public static final int HEADERSIZE = Integer.BYTES + Long.BYTES;
11+
private static final Logger LOG = NaRPCUtils.getLogger();
12+
13+
public static void makeMessage(long ticket, NaRPCMessage message, ByteBuffer buffer) throws IOException {
14+
buffer.clear().position(HEADERSIZE);
15+
int size = message.write(buffer);
16+
buffer.flip();
17+
if ((size + HEADERSIZE) != buffer.remaining()) {
18+
throw new IOException("Error in serialization");
19+
}
20+
21+
buffer.clear();
22+
buffer.putInt(size);
23+
buffer.putLong(ticket);
24+
buffer.clear().limit(HEADERSIZE + size);
25+
}
26+
27+
public static long fetchBuffer(SocketChannel channel, ByteBuffer buffer) throws IOException{
28+
buffer.clear().limit(HEADERSIZE);
29+
while (buffer.hasRemaining()) {
30+
if (channel.read(buffer) < 0){
31+
return -1;
32+
}
33+
}
34+
buffer.flip();
35+
int size = buffer.getInt();
36+
long ticket = buffer.getLong();
37+
buffer.clear().limit(size);
38+
while (buffer.hasRemaining()) {
39+
if (channel.read(buffer) < 0) {
40+
throw new IOException("error when reading header from socket");
41+
}
42+
43+
}
44+
buffer.flip();
45+
// LOG.info("fetching message with ticket " + ticket + ", threadid " + Thread.currentThread().getName());
46+
return ticket;
47+
}
48+
}

‎src/main/java/com/ibm/narpc/NaRPCServerChannel.java

+12-10
Original file line numberDiff line numberDiff line change
@@ -25,36 +25,38 @@
2525
import java.nio.ByteBuffer;
2626
import java.nio.channels.SocketChannel;
2727

28-
public class NaRPCServerChannel extends NaRPCChannel {
28+
public class NaRPCServerChannel {
2929
private SocketChannel channel;
3030
private ByteBuffer buffer;
3131

3232
public NaRPCServerChannel(NaRPCGroup group, SocketChannel channel){
3333
this.channel = channel;
3434
this.buffer = ByteBuffer.allocate(group.getMessageSize());
3535
}
36-
37-
public SocketChannel getSocketChannel() {
38-
return channel;
39-
}
4036

41-
public long fetch(NaRPCMessage message) throws IOException {
42-
long ticket = fetchBuffer(channel, buffer);
37+
public long receiveMessage(NaRPCMessage message) throws IOException {
38+
long ticket = NaRPCProtocol.fetchBuffer(channel, buffer);
4339
if (ticket > 0){
4440
message.update(buffer);
4541
}
4642
return ticket;
4743
}
4844

49-
public void transmit(long ticket, NaRPCMessage message) throws IOException {
50-
makeMessage(ticket, message, buffer);
51-
transmitMessage(channel, buffer);
45+
public void transmitMessage(long ticket, NaRPCMessage message) throws IOException {
46+
NaRPCProtocol.makeMessage(ticket, message, buffer);
47+
while(buffer.hasRemaining()){
48+
channel.write(buffer);
49+
}
5250
}
5351

5452
public void close() throws IOException{
5553
this.channel.close();
5654
}
5755

56+
public SocketChannel getSocketChannel() {
57+
return channel;
58+
}
59+
5860
public String address() throws IOException {
5961
return channel.getRemoteAddress().toString();
6062
}

0 commit comments

Comments
 (0)