Fixes, improved tests and other improvements
This commit is contained in:
parent
56ebb7b8fa
commit
334a510743
@ -41,19 +41,19 @@ public class NetworkAddress implements Streamable {
|
|||||||
/**
|
/**
|
||||||
* Stream number for this node
|
* Stream number for this node
|
||||||
*/
|
*/
|
||||||
private long stream;
|
private final long stream;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* same service(s) listed in version
|
* same service(s) listed in version
|
||||||
*/
|
*/
|
||||||
private long services;
|
private final long services;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* IPv6 address. IPv4 addresses are written into the message as a 16 byte IPv4-mapped IPv6 address
|
* IPv6 address. IPv4 addresses are written into the message as a 16 byte IPv4-mapped IPv6 address
|
||||||
* (12 bytes 00 00 00 00 00 00 00 00 00 00 FF FF, followed by the 4 bytes of the IPv4 address).
|
* (12 bytes 00 00 00 00 00 00 00 00 00 00 FF FF, followed by the 4 bytes of the IPv4 address).
|
||||||
*/
|
*/
|
||||||
private byte[] ipv6;
|
private final byte[] ipv6;
|
||||||
private int port;
|
private final int port;
|
||||||
|
|
||||||
private NetworkAddress(Builder builder) {
|
private NetworkAddress(Builder builder) {
|
||||||
time = builder.time;
|
time = builder.time;
|
||||||
|
@ -16,12 +16,16 @@
|
|||||||
|
|
||||||
package ch.dissem.bitmessage.factory;
|
package ch.dissem.bitmessage.factory;
|
||||||
|
|
||||||
import ch.dissem.bitmessage.ports.NetworkHandler;
|
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
import java.nio.ByteBuffer;
|
import java.nio.ByteBuffer;
|
||||||
import java.util.*;
|
import java.util.Map;
|
||||||
|
import java.util.Stack;
|
||||||
|
import java.util.TreeMap;
|
||||||
|
|
||||||
|
import static ch.dissem.bitmessage.ports.NetworkHandler.HEADER_SIZE;
|
||||||
|
import static ch.dissem.bitmessage.ports.NetworkHandler.MAX_PAYLOAD_SIZE;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A pool for {@link ByteBuffer}s. As they may use up a lot of memory,
|
* A pool for {@link ByteBuffer}s. As they may use up a lot of memory,
|
||||||
@ -30,78 +34,58 @@ import java.util.*;
|
|||||||
class BufferPool {
|
class BufferPool {
|
||||||
private static final Logger LOG = LoggerFactory.getLogger(BufferPool.class);
|
private static final Logger LOG = LoggerFactory.getLogger(BufferPool.class);
|
||||||
|
|
||||||
public static final BufferPool bufferPool = new BufferPool(256, 2048);
|
public static final BufferPool bufferPool = new BufferPool();
|
||||||
|
|
||||||
private final Map<Size, Integer> capacities = new EnumMap<>(Size.class);
|
private final Map<Integer, Stack<ByteBuffer>> pools = new TreeMap<>();
|
||||||
private final Map<Size, Stack<ByteBuffer>> pools = new EnumMap<>(Size.class);
|
|
||||||
|
|
||||||
private BufferPool(int small, int medium) {
|
private BufferPool() {
|
||||||
capacities.put(Size.HEADER, 24);
|
pools.put(HEADER_SIZE, new Stack<ByteBuffer>());
|
||||||
capacities.put(Size.SMALL, small);
|
pools.put(54, new Stack<ByteBuffer>());
|
||||||
capacities.put(Size.MEDIUM, medium);
|
pools.put(1000, new Stack<ByteBuffer>());
|
||||||
capacities.put(Size.LARGE, NetworkHandler.MAX_PAYLOAD_SIZE);
|
pools.put(60000, new Stack<ByteBuffer>());
|
||||||
pools.put(Size.HEADER, new Stack<ByteBuffer>());
|
pools.put(MAX_PAYLOAD_SIZE, new Stack<ByteBuffer>());
|
||||||
pools.put(Size.SMALL, new Stack<ByteBuffer>());
|
|
||||||
pools.put(Size.MEDIUM, new Stack<ByteBuffer>());
|
|
||||||
pools.put(Size.LARGE, new Stack<ByteBuffer>());
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized ByteBuffer allocate(int capacity) {
|
public synchronized ByteBuffer allocate(int capacity) {
|
||||||
Size targetSize = getTargetSize(capacity);
|
for (Map.Entry<Integer, Stack<ByteBuffer>> e : pools.entrySet()) {
|
||||||
Size s = targetSize;
|
if (e.getKey() >= capacity && !e.getValue().isEmpty()) {
|
||||||
do {
|
return e.getValue().pop();
|
||||||
Stack<ByteBuffer> pool = pools.get(s);
|
|
||||||
if (!pool.isEmpty()) {
|
|
||||||
return pool.pop();
|
|
||||||
}
|
}
|
||||||
s = s.next();
|
}
|
||||||
} while (s != null);
|
Integer targetSize = getTargetSize(capacity);
|
||||||
LOG.debug("Creating new buffer of size " + targetSize);
|
LOG.debug("Creating new buffer of size " + targetSize);
|
||||||
return ByteBuffer.allocate(capacities.get(targetSize));
|
return ByteBuffer.allocate(targetSize);
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized ByteBuffer allocate() {
|
/**
|
||||||
Stack<ByteBuffer> pool = pools.get(Size.HEADER);
|
* Returns a buffer that has the size of the Bitmessage network message header, 24 bytes.
|
||||||
|
*
|
||||||
|
* @return a buffer of size 24
|
||||||
|
*/
|
||||||
|
public synchronized ByteBuffer allocateHeaderBuffer() {
|
||||||
|
Stack<ByteBuffer> pool = pools.get(HEADER_SIZE);
|
||||||
if (!pool.isEmpty()) {
|
if (!pool.isEmpty()) {
|
||||||
return pool.pop();
|
return pool.pop();
|
||||||
} else {
|
} else {
|
||||||
return ByteBuffer.allocate(capacities.get(Size.HEADER));
|
return ByteBuffer.allocate(HEADER_SIZE);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized void deallocate(ByteBuffer buffer) {
|
public synchronized void deallocate(ByteBuffer buffer) {
|
||||||
buffer.clear();
|
buffer.clear();
|
||||||
Size size = getTargetSize(buffer.capacity());
|
|
||||||
if (buffer.capacity() != capacities.get(size)) {
|
if (!pools.keySet().contains(buffer.capacity())) {
|
||||||
throw new IllegalArgumentException("Illegal buffer capacity " + buffer.capacity() +
|
throw new IllegalArgumentException("Illegal buffer capacity " + buffer.capacity() +
|
||||||
" one of " + capacities.values() + " expected.");
|
" one of " + pools.keySet() + " expected.");
|
||||||
}
|
}
|
||||||
pools.get(size).push(buffer);
|
pools.get(buffer.capacity()).push(buffer);
|
||||||
}
|
}
|
||||||
|
|
||||||
private Size getTargetSize(int capacity) {
|
private Integer getTargetSize(int capacity) {
|
||||||
for (Size s : Size.values()) {
|
for (Integer size : pools.keySet()) {
|
||||||
if (capacity <= capacities.get(s)) {
|
if (size >= capacity) return size;
|
||||||
return s;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
throw new IllegalArgumentException("Requested capacity too large: " +
|
throw new IllegalArgumentException("Requested capacity too large: " +
|
||||||
"requested=" + capacity + "; max=" + capacities.get(Size.LARGE));
|
"requested=" + capacity + "; max=" + MAX_PAYLOAD_SIZE);
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
private enum Size {
|
|
||||||
HEADER, SMALL, MEDIUM, LARGE;
|
|
||||||
|
|
||||||
public Size next() {
|
|
||||||
switch (this) {
|
|
||||||
case SMALL:
|
|
||||||
return MEDIUM;
|
|
||||||
case MEDIUM:
|
|
||||||
return LARGE;
|
|
||||||
default:
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -179,7 +179,13 @@ class V3MessageFactory {
|
|||||||
long services = Decode.int64(stream);
|
long services = Decode.int64(stream);
|
||||||
byte[] ipv6 = Decode.bytes(stream, 16);
|
byte[] ipv6 = Decode.bytes(stream, 16);
|
||||||
int port = Decode.uint16(stream);
|
int port = Decode.uint16(stream);
|
||||||
return new NetworkAddress.Builder().time(time).stream(streamNumber).services(services).ipv6(ipv6).port(port).build();
|
return new NetworkAddress.Builder()
|
||||||
|
.time(time)
|
||||||
|
.stream(streamNumber)
|
||||||
|
.services(services)
|
||||||
|
.ipv6(ipv6)
|
||||||
|
.port(port)
|
||||||
|
.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
private static boolean testChecksum(byte[] checksum, byte[] payload) {
|
private static boolean testChecksum(byte[] checksum, byte[] payload) {
|
||||||
|
@ -58,7 +58,7 @@ public class V3MessageReader {
|
|||||||
public ByteBuffer getActiveBuffer() {
|
public ByteBuffer getActiveBuffer() {
|
||||||
if (state != null && state != ReaderState.DATA) {
|
if (state != null && state != ReaderState.DATA) {
|
||||||
if (headerBuffer == null) {
|
if (headerBuffer == null) {
|
||||||
headerBuffer = bufferPool.allocate();
|
headerBuffer = bufferPool.allocateHeaderBuffer();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return state == ReaderState.DATA ? dataBuffer : headerBuffer;
|
return state == ReaderState.DATA ? dataBuffer : headerBuffer;
|
||||||
|
@ -31,8 +31,9 @@ import java.util.concurrent.Future;
|
|||||||
*/
|
*/
|
||||||
public interface NetworkHandler {
|
public interface NetworkHandler {
|
||||||
int NETWORK_MAGIC_NUMBER = 8;
|
int NETWORK_MAGIC_NUMBER = 8;
|
||||||
|
int HEADER_SIZE = 24;
|
||||||
int MAX_PAYLOAD_SIZE = 1600003;
|
int MAX_PAYLOAD_SIZE = 1600003;
|
||||||
int MAX_MESSAGE_SIZE = 24 + MAX_PAYLOAD_SIZE;
|
int MAX_MESSAGE_SIZE = HEADER_SIZE + MAX_PAYLOAD_SIZE;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Connects to the trusted host, fetches and offers new messages and disconnects afterwards.
|
* Connects to the trusted host, fetches and offers new messages and disconnects afterwards.
|
||||||
|
@ -99,6 +99,10 @@ public abstract class AbstractConnection {
|
|||||||
return state;
|
return state;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public long[] getStreams() {
|
||||||
|
return streams;
|
||||||
|
}
|
||||||
|
|
||||||
protected void handleMessage(MessagePayload payload) {
|
protected void handleMessage(MessagePayload payload) {
|
||||||
switch (state) {
|
switch (state) {
|
||||||
case ACTIVE:
|
case ACTIVE:
|
||||||
|
@ -170,7 +170,7 @@ public class DefaultNetworkHandler implements NetworkHandler, ContextHolder {
|
|||||||
|
|
||||||
for (Connection connection : connections) {
|
for (Connection connection : connections) {
|
||||||
if (connection.getState() == ACTIVE) {
|
if (connection.getState() == ACTIVE) {
|
||||||
long stream = connection.getNode().getStream();
|
for (long stream : connection.getStreams()) {
|
||||||
streams.add(stream);
|
streams.add(stream);
|
||||||
if (connection.getMode() == SERVER) {
|
if (connection.getMode() == SERVER) {
|
||||||
inc(incomingConnections, stream);
|
inc(incomingConnections, stream);
|
||||||
@ -179,6 +179,7 @@ public class DefaultNetworkHandler implements NetworkHandler, ContextHolder {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
Property[] streamProperties = new Property[streams.size()];
|
Property[] streamProperties = new Property[streams.size()];
|
||||||
int i = 0;
|
int i = 0;
|
||||||
for (Long stream : streams) {
|
for (Long stream : streams) {
|
||||||
|
@ -44,7 +44,7 @@ public class ConnectionInfo extends AbstractConnection {
|
|||||||
private ByteBuffer payloadOut;
|
private ByteBuffer payloadOut;
|
||||||
private V3MessageReader reader = new V3MessageReader();
|
private V3MessageReader reader = new V3MessageReader();
|
||||||
private boolean syncFinished;
|
private boolean syncFinished;
|
||||||
private long lastUpdate = Long.MAX_VALUE;
|
private long lastUpdate = System.currentTimeMillis();
|
||||||
|
|
||||||
public ConnectionInfo(InternalContext context, Mode mode,
|
public ConnectionInfo(InternalContext context, Mode mode,
|
||||||
NetworkAddress node, NetworkHandler.MessageListener listener,
|
NetworkAddress node, NetworkHandler.MessageListener listener,
|
||||||
|
@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
|
|||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.net.InetAddress;
|
import java.net.InetAddress;
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
|
import java.net.NoRouteToHostException;
|
||||||
import java.nio.ByteBuffer;
|
import java.nio.ByteBuffer;
|
||||||
import java.nio.channels.*;
|
import java.nio.channels.*;
|
||||||
import java.util.*;
|
import java.util.*;
|
||||||
@ -46,7 +47,6 @@ import static ch.dissem.bitmessage.utils.DebugUtils.inc;
|
|||||||
import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool;
|
import static ch.dissem.bitmessage.utils.ThreadFactoryBuilder.pool;
|
||||||
import static java.nio.channels.SelectionKey.OP_READ;
|
import static java.nio.channels.SelectionKey.OP_READ;
|
||||||
import static java.nio.channels.SelectionKey.OP_WRITE;
|
import static java.nio.channels.SelectionKey.OP_WRITE;
|
||||||
import static java.util.Collections.synchronizedMap;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Network handler using java.nio, resulting in less threads.
|
* Network handler using java.nio, resulting in less threads.
|
||||||
@ -209,7 +209,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
|
|||||||
connection,
|
connection,
|
||||||
channel.register(selector, OP_READ | OP_WRITE, connection)
|
channel.register(selector, OP_READ | OP_WRITE, connection)
|
||||||
);
|
);
|
||||||
} catch (AsynchronousCloseException ignore) {
|
} catch (NoRouteToHostException | AsynchronousCloseException ignore) {
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
LOG.error(e.getMessage(), e);
|
LOG.error(e.getMessage(), e);
|
||||||
}
|
}
|
||||||
@ -268,7 +268,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
|
|||||||
} else {
|
} else {
|
||||||
key.interestOps(OP_READ | OP_WRITE);
|
key.interestOps(OP_READ | OP_WRITE);
|
||||||
}
|
}
|
||||||
} catch (NodeException | IOException e) {
|
} catch (CancelledKeyException | NodeException | IOException e) {
|
||||||
connection.disconnect();
|
connection.disconnect();
|
||||||
}
|
}
|
||||||
if (connection.getState() == DISCONNECTED) {
|
if (connection.getState() == DISCONNECTED) {
|
||||||
@ -413,7 +413,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
|
|||||||
|
|
||||||
for (ConnectionInfo connection : connections.keySet()) {
|
for (ConnectionInfo connection : connections.keySet()) {
|
||||||
if (connection.getState() == ACTIVE) {
|
if (connection.getState() == ACTIVE) {
|
||||||
long stream = connection.getNode().getStream();
|
for (long stream : connection.getStreams()) {
|
||||||
streams.add(stream);
|
streams.add(stream);
|
||||||
if (connection.getMode() == SERVER) {
|
if (connection.getMode() == SERVER) {
|
||||||
inc(incomingConnections, stream);
|
inc(incomingConnections, stream);
|
||||||
@ -422,6 +422,7 @@ public class NioNetworkHandler implements NetworkHandler, InternalContext.Contex
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
Property[] streamProperties = new Property[streams.size()];
|
Property[] streamProperties = new Property[streams.size()];
|
||||||
int i = 0;
|
int i = 0;
|
||||||
for (Long stream : streams) {
|
for (Long stream : streams) {
|
||||||
|
@ -169,15 +169,24 @@ public class NetworkHandlerTest {
|
|||||||
} while (networkHandler.isRunning());
|
} while (networkHandler.isRunning());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
private Property waitForNetworkStatus(BitmessageContext ctx) throws InterruptedException {
|
||||||
public void ensureNodesAreConnecting() throws Exception {
|
|
||||||
node.startup();
|
|
||||||
Property status;
|
Property status;
|
||||||
do {
|
do {
|
||||||
Thread.sleep(100);
|
Thread.sleep(100);
|
||||||
status = node.status().getProperty("network", "connections", "stream 0");
|
status = ctx.status().getProperty("network", "connections", "stream 1");
|
||||||
} while (status == null);
|
} while (status == null);
|
||||||
assertEquals(1, status.getProperty("outgoing").getValue());
|
return status;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void ensureNodesAreConnecting() throws Exception {
|
||||||
|
node.startup();
|
||||||
|
|
||||||
|
Property nodeStatus = waitForNetworkStatus(node);
|
||||||
|
Property peerStatus = waitForNetworkStatus(peer);
|
||||||
|
|
||||||
|
assertEquals(1, nodeStatus.getProperty("outgoing").getValue());
|
||||||
|
assertEquals(1, peerStatus.getProperty("incoming").getValue());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
Loading…
Reference in New Issue
Block a user