Skip to content

Commit 37f37f9

Browse files
belabanbelaban
committed
- Added position(int) to ByteBufferOutputStream
- Added reset() to BaseDataOutputStream - Using reset() instead of position(0) - Added IOException to write methods in BaseDataOutputStream - Moved grow_exponentially into BaseDataOutputStream - Removed use_direct_memory from MessageReader; memory type can be inferred from existing buffer - Added Util.bufferToArray plus unit test - Changed use_direct_memory to true in TCP_NIO2 default config - Changed sendUnicast() to use ByteBuffer instead of byte arrays - NoBundler now reuses output buffer from BaseBundler - UDP now handles ByteBuffer - Removed SimpleTCP - Added use_direct_memory to TPConfig - Util: added doWithByteBuffer() - SHARED_LOOPBACK: implemented sendTo(), sendToAll() and sendUnicast() - LocalTransport: implemented default sendTo(ByteBuffer) and sendToAll(ByteBuffer) - BaseBundler/PerDestinationBundler: changed from byte array to ByteBuffer - TCP/TCP_NIO2: changed from sending byte array to sending ByteBuffer - BasicTCP/TCP/TCP_NIO2: replaced send() with sendUnicast(): one less indirection - TPConfig: added use_direct_memory (moved from TCP_NIO2) - TP: changed from byte arrays to ByteBuffers - GossipRouter: changed from byte arrays to ByteBuffers - FragmentedMessage.writePayload(): cat DataInput to a ByteBufferInputStream rather than a ByteArrayDataInputStream - GossipRouter: added use_direct_memory. Also skip sending to self - FragmentedMessage.writePayload() now downcasts to BaseDataOutputStream - Sanity checks for use_direct_memory
1 parent cfbd540 commit 37f37f9

36 files changed

Lines changed: 441 additions & 606 deletions

‎conf/nio-default.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
sock_conn_timeout="300ms"
1212
tcp_nodelay="${tcp-nodelay:false}"
1313
max_send_buffers="${max_send_buffers:10}"
14+
use_direct_memory="true"
1415

1516
bundler_type="${jgroups.bundler.type:pd}"
1617
bundler.max_size="${jgroups.bundler.max_size:64K}"

‎src/org/jgroups/FragmentedMessage.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package org.jgroups;
22

3-
import org.jgroups.util.ByteArrayDataOutputStream;
3+
import org.jgroups.util.BaseDataOutputStream;
44
import org.jgroups.util.PartialOutputStream;
55

66
import java.io.DataInput;
@@ -47,11 +47,11 @@ protected Message copyPayload(Message copy) {
4747
}
4848

4949
public void writePayload(DataOutput out) throws IOException {
50-
ByteArrayDataOutputStream bos=out instanceof ByteArrayDataOutputStream? (ByteArrayDataOutputStream)out : null;
50+
// don't fail silently (pos would be -1) if wrong type but throw an exception!
51+
BaseDataOutputStream bos=(BaseDataOutputStream)out;
5152
int size_pos=bos != null? bos.position() : -1;
5253
out.writeInt(length);
5354
PartialOutputStream pos=new PartialOutputStream(out, offset, length);
54-
5555
int prev_pos=bos != null? bos.position() : -1;
5656
original_msg.writeTo(pos);
5757
int last_pos=bos != null? bos.position() : -1;
@@ -75,13 +75,11 @@ public void readPayload(DataInput in) throws IOException {
7575
}
7676
}
7777

78-
7978
public String toString() {
8079
return String.format("%s [off=%d len=%d] (original msg: %s)",
8180
getClass().getSimpleName(), offset, length, original_msg);
8281
}
8382

84-
8583
protected <T extends BytesMessage> T createMessage() {
8684
return (T)new FragmentedMessage();
8785
}

‎src/org/jgroups/blocks/cs/BaseServer.java‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,16 @@
2222
import java.util.Map.Entry;
2323
import java.util.concurrent.ConcurrentHashMap;
2424
import java.util.concurrent.CopyOnWriteArrayList;
25-
import java.util.concurrent.TimeUnit;
2625
import java.util.concurrent.atomic.AtomicBoolean;
2726
import java.util.concurrent.locks.Lock;
2827
import java.util.concurrent.locks.ReentrantLock;
2928
import java.util.function.BiConsumer;
3029
import java.util.function.Predicate;
3130
import java.util.stream.Collectors;
3231

32+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
33+
import static java.util.concurrent.TimeUnit.NANOSECONDS;
34+
3335
/**
3436
* Abstract class for a server handling sending, receiving and connection management.
3537
* @since 3.6.5
@@ -110,7 +112,7 @@ protected BaseServer(ThreadFactory f, SocketFactory sf, int recv_buf_size) {
110112
public int socketConnectionTimeout() {return sock_conn_timeout;}
111113
public BaseServer socketConnectionTimeout(int timeout) {this.sock_conn_timeout = timeout; return this;}
112114
public long connExpireTime() {return conn_expire_time;}
113-
public BaseServer connExpireTimeout(long t) {conn_expire_time=TimeUnit.NANOSECONDS.convert(t, TimeUnit.MILLISECONDS); return this;}
115+
public BaseServer connExpireTimeout(long t) {conn_expire_time=NANOSECONDS.convert(t, MILLISECONDS); return this;}
114116
public TimeService timeService() {return time_service;}
115117
public BaseServer timeService(TimeService ts) {this.time_service=ts; return this;}
116118
public int receiveBufferSize() {return recv_buf_size;}
@@ -141,9 +143,6 @@ public int getNumOpenConnections() {
141143
return retval;
142144
}
143145

144-
/**
145-
* Starts accepting connections. Typically, socket handler or selectors thread are started here.
146-
*/
147146
public void start() throws Exception {
148147
if(reaperInterval > 0 && (reaper == null || !reaper.isAlive())) {
149148
reaper=new Reaper();
@@ -539,10 +538,17 @@ public void sendToAll(byte[] data, int offset, int length) {
539538
}
540539

541540
public void sendToAll(ByteBuffer data) {
541+
int pos=data.position(), limit=data.limit();
542+
boolean first=true;
542543
for(Map.Entry<Address,Connection> entry: conns.entrySet()) {
543544
Connection conn=entry.getValue();
544545
try {
545-
conn.send(data.duplicate());
546+
if(!first)
547+
data.position(pos).limit(limit);
548+
else
549+
first=false;
550+
conn.send(data);
551+
first=false;
546552
}
547553
catch(Throwable ex) {
548554
Address dest=entry.getKey();

‎src/org/jgroups/blocks/cs/TcpConnection.java‎

Lines changed: 8 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,7 @@
44
import org.jgroups.Version;
55
import org.jgroups.annotations.GuardedBy;
66
import org.jgroups.stack.IpAddress;
7-
import org.jgroups.util.Bits;
8-
import org.jgroups.util.ByteArrayDataOutputStream;
9-
import org.jgroups.util.ThreadFactory;
10-
import org.jgroups.util.Util;
7+
import org.jgroups.util.*;
118

129
import javax.net.ssl.SSLException;
1310
import javax.net.ssl.SSLHandshakeException;
@@ -138,24 +135,17 @@ public void send(byte[] data, int offset, int length) throws Exception {
138135
}
139136
}
140137

141-
public void locklessSend(byte[] data, int offset, int length) throws Exception {
142-
if(out == null)
138+
public void send(ByteBuffer buf) throws Exception {
139+
if(buf == null)
143140
return;
144-
doSend(data, offset, length, true);
141+
ByteArray ba=Util.bufferToByteArray(buf);
142+
send(ba.array(), ba.offset(), ba.length());
145143
}
146144

147-
public void send(ByteBuffer buf) throws Exception {
148-
if(buf == null)
145+
public void locklessSend(byte[] data, int offset, int length) throws Exception {
146+
if(out == null)
149147
return;
150-
int offset=buf.hasArray()? buf.arrayOffset() + buf.position() : buf.position(),
151-
len=buf.remaining();
152-
if(buf.hasArray())
153-
send(buf.array(), offset, len);
154-
else {
155-
byte[] tmp=new byte[len];
156-
buf.get(tmp, 0, len);
157-
send(tmp, 0, len); // will get copied again if send-queues are enabled
158-
}
148+
doSend(data, offset, length, true);
159149
}
160150

161151
@GuardedBy("send_lock")

‎src/org/jgroups/nio/MessageReader.java‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,10 @@
1919
* @since 5.5.3
2020
*/
2121
public class MessageReader {
22-
23-
private final SocketChannel channel;
24-
private ByteBuffer buffer;
25-
private int readerIndex;
26-
protected int max_length; // max number of bytes to read (JGRP-2523)
27-
protected boolean use_direct_buffers;
22+
private final SocketChannel channel;
23+
private ByteBuffer buffer;
24+
private int readerIndex;
25+
protected int max_length; // max number of bytes to read (JGRP-2523)
2826
protected final NioConnection conn;
2927

3028

@@ -39,7 +37,6 @@ public MessageReader(NioConnection c, SocketChannel channel) {
3937
public MessageReader(NioConnection c, SocketChannel channel, int initial_buf_size, boolean use_direct_buffers) {
4038
this.conn=c;
4139
this.channel = channel;
42-
this.use_direct_buffers = use_direct_buffers;
4340
buffer = use_direct_buffers? ByteBuffer.allocateDirect(initial_buf_size) : ByteBuffer.allocate(initial_buf_size);
4441
}
4542

@@ -85,7 +82,6 @@ public ByteBuffer readMessage() throws IOException {
8582
}
8683

8784
// Not enough data or buffer too small
88-
8985
if (readerIndex + 4 + length > buffer.capacity()) {
9086
// Ensure buffer fits entire message
9187
makeSpace(4 + length);
@@ -150,7 +146,7 @@ private void makeSpace(int totalSpace) {
150146
buffer.compact();
151147
} else {
152148
int newCapacity = Math.max(totalSpace, buffer.capacity() * 2);
153-
ByteBuffer newBuffer = use_direct_buffers? ByteBuffer.allocateDirect(newCapacity) : ByteBuffer.allocate(newCapacity);
149+
ByteBuffer newBuffer = buffer.isDirect()? ByteBuffer.allocateDirect(newCapacity) : ByteBuffer.allocate(newCapacity);
154150
newBuffer.put(buffer);
155151
buffer = newBuffer;
156152
}
@@ -163,7 +159,7 @@ public MessageReader maxLength(int max_length) {
163159

164160
@Override
165161
public String toString() {
166-
return String.format("[readerIndex=%d writerIndex=%d capacity=%d remaining=%d]", readerIndex, buffer.position(), buffer.capacity(),
167-
buffer.position() - readerIndex);
162+
return String.format("[readerIndex=%d writerIndex=%d capacity=%d remaining=%d]",
163+
readerIndex, buffer.position(), buffer.capacity(), buffer.position() - readerIndex);
168164
}
169165
}

‎src/org/jgroups/protocols/ASYM_ENCRYPT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -409,7 +409,7 @@ protected void serializeKeys(ByteArrayDataOutputStream out, boolean serialize_sh
409409
num++;
410410
}
411411
int curr_pos=out.position();
412-
out.position(0).writeInt(num);
412+
out.reset().writeInt(num);
413413
out.position(curr_pos);
414414
}
415415

‎src/org/jgroups/protocols/BaseBundler.java‎

Lines changed: 11 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ public abstract class BaseBundler implements Bundler {
4545
protected MessageProcessingPolicy msg_processing_policy;
4646
protected final ReentrantLock lock=new ReentrantLock();
4747
protected @GuardedBy("lock") long count; // current number of bytes accumulated
48-
protected ByteArrayDataOutputStream output;
48+
protected ByteBufferOutputStream output;
4949
protected MsgStats msg_stats;
5050
protected Log log;
5151
protected SuppressLog<Address> suppress_log;
@@ -156,7 +156,7 @@ public void init(TP transport) {
156156
msg_stats=transport.getMessageStats();
157157
log=transport.getLog();
158158
suppress_log=new SuppressLog<>(log);
159-
output=new ByteArrayDataOutputStream(max_size + MSG_OVERHEAD);
159+
output=new ByteBufferOutputStream(max_size + MSG_OVERHEAD, false, transport.useDirectMemory());
160160
}
161161

162162
public void resetStats() {
@@ -204,7 +204,7 @@ public int getQueueSize() {
204204
if(list.isEmpty())
205205
continue;
206206
Address dst=entry.getKey();
207-
output.position(0);
207+
output.reset();
208208
try {
209209
if(list.size() == 1)
210210
sendSingle(dst, list.get(0), output);
@@ -227,7 +227,7 @@ public int getQueueSize() {
227227
}
228228
}
229229

230-
protected void sendSingle(Address dst, Message msg, ByteArrayDataOutputStream out) throws Exception {
230+
protected void sendSingle(Address dst, Message msg, ByteBufferOutputStream out) throws Exception {
231231
if(dst == null) { // multicast
232232
sendSingleMessage(dst, msg, out);
233233
loopbackUnlessDontLoopbackIsSet(msg);
@@ -242,7 +242,7 @@ protected void sendSingle(Address dst, Message msg, ByteArrayDataOutputStream ou
242242
}
243243
}
244244

245-
protected void sendMultiple(Address dst, Address sender, List<Message> list, ByteArrayDataOutputStream out) throws Exception {
245+
protected void sendMultiple(Address dst, Address sender, List<Message> list, ByteBufferOutputStream out) throws Exception {
246246
if(dst == null) { // multicast
247247
sendMessageList(dst, sender, list, out);
248248
loopback(dst, transport.getAddress(), list, list.size());
@@ -257,21 +257,6 @@ protected void sendMultiple(Address dst, Address sender, List<Message> list, Byt
257257
}
258258
}
259259

260-
protected void sendMultiple(Address dst, Address sender, Message[] list, int len, ByteArrayDataOutputStream out) {
261-
if(dst == null) { // multicast
262-
sendMessageListArray(dst, sender, list, len, out);
263-
loopback(dst, transport.getAddress(), list, len);
264-
}
265-
else { // unicast
266-
boolean send_to_self=Objects.equals(transport.getAddress(), dst)
267-
|| dst instanceof PhysicalAddress && dst.equals(transport.localPhysicalAddress());
268-
if(send_to_self)
269-
loopback(dst, transport.getAddress(), list, len);
270-
else
271-
sendMessageListArray(dst, sender, list, len, out);
272-
}
273-
}
274-
275260
protected void loopbackUnlessDontLoopbackIsSet(Message msg) {
276261
if(msg.isFlagSet(DONT_LOOPBACK))
277262
return;
@@ -312,31 +297,22 @@ protected void loopback(Address dest, Address sender, Message[] list, int len) {
312297
loopback(dest, sender, fa, fa.size());
313298
}
314299

315-
protected void sendSingleMessage(final Address dest, final Message msg, ByteArrayDataOutputStream out) throws Exception {
300+
protected void sendSingleMessage(final Address dest, final Message msg, ByteBufferOutputStream out) throws Exception {
316301
Util.writeMessage(msg, out, dest == null);
317-
transport.doSend(out.buffer(), 0, out.position(), dest);
302+
out.buf().flip();
303+
transport.doSend(out.buf(), dest);
318304
transport.getMessageStats().incrNumSingleMsgsSent();
319305
num_single_msgs_sent.increment();
320306
}
321307

322-
protected void sendMessageList(Address dest, Address src, List<Message> list, ByteArrayDataOutputStream out) throws Exception {
308+
protected void sendMessageList(Address dest, Address src, List<Message> list, ByteBufferOutputStream out) throws Exception {
323309
Util.writeMessageList(dest, src, transport.cluster_name.val(), list, out, dest == null);
324-
transport.doSend(out.buffer(), 0, out.position(), dest);
310+
out.buf().flip();
311+
transport.doSend(out.buf(), dest);
325312
transport.getMessageStats().incrNumBatchesSent();
326313
num_batches_sent.increment();
327314
}
328315

329-
protected void sendMessageListArray(final Address dest, final Address src, Message[] list, int len, ByteArrayDataOutputStream out) {
330-
try {
331-
Util.writeMessageList(dest, src, transport.cluster_name.val(), list, 0, len, out, dest == null);
332-
transport.doSend(out.buffer(), 0, out.position(), dest);
333-
transport.getMessageStats().incrNumBatchesSent();
334-
}
335-
catch(Throwable e) {
336-
log.trace(Util.getMessage("FailureSendingMsgBundle"), transport.getAddress(), e);
337-
}
338-
}
339-
340316
@GuardedBy("lock") protected void addMessage(Message msg, int size) {
341317
Address dest=msg.getDest();
342318
List<Message> tmp=msgs.computeIfAbsent(dest, FUNC);

‎src/org/jgroups/protocols/BasicTCP.java‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,7 @@
1515
import org.jgroups.protocols.pbcast.GMS;
1616

1717
import java.net.InetAddress;
18-
import java.util.Collection;
19-
import java.util.HashSet;
20-
import java.util.List;
21-
import java.util.Optional;
22-
import java.util.Set;
18+
import java.util.*;
2319
import java.util.concurrent.atomic.LongAdder;
2420
import java.util.function.Predicate;
2521

@@ -179,17 +175,11 @@ else if(conn_expire_time > 0 && reaper_interval == 0) {
179175
}
180176
}
181177

182-
public void sendUnicast(PhysicalAddress dest, byte[] data, int offset, int length) throws Exception {
183-
send(dest, data, offset, length);
184-
}
185-
186178
public abstract String printConnections();
187179

188180
@ManagedOperation(description="Clears all connections (they will get re-established). For testing only, don't use !")
189181
public abstract BasicTCP clearConnections(boolean graceful);
190182

191-
public abstract void send(Address dest, byte[] data, int offset, int length) throws Exception;
192-
193183
public abstract void retainAll(Collection<Address> members);
194184

195185
@Override

‎src/org/jgroups/protocols/LocalTransport.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,10 @@
22

33
import org.jgroups.Address;
44
import org.jgroups.View;
5+
import org.jgroups.util.ByteArray;
6+
import org.jgroups.util.Util;
7+
8+
import java.nio.ByteBuffer;
59

610
/**
711
* A local transport is used for sending messages only to single (or all) members of the same host.
@@ -36,6 +40,12 @@ public interface LocalTransport {
3640
*/
3741
void sendTo(Address dest, byte[] buf, int offset, int length) throws Exception;
3842

43+
default void sendTo(Address dest, ByteBuffer buf) throws Exception {
44+
ByteArray ba=Util.bufferToByteArray(buf);
45+
if(ba != null)
46+
sendTo(dest, ba.array(), ba.offset(), ba.length());
47+
}
48+
3949
/**
4050
* Sends a message to all local members.
4151
* @param buf The buffer to send
@@ -44,4 +54,10 @@ public interface LocalTransport {
4454
* @exception Exception Thrown when the send failed.
4555
*/
4656
void sendToAll(byte[] buf, int offset, int length) throws Exception;
57+
58+
default void sendToAll(ByteBuffer buf) throws Exception {
59+
ByteArray ba=Util.bufferToByteArray(buf);
60+
if(ba != null)
61+
sendToAll(ba.array(), ba.offset(), ba.length());
62+
}
4763
}

0 commit comments

Comments
 (0)