From a8479ee02f90c68035415219026c6d94ec31cd73 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 20 Nov 2014 18:48:58 +0000 Subject: [PATCH] HCOLL-134 better handling of toString() and close() of the TCP connections --- .../hash/replication/AbstractReplication.java | 58 ++++++++++++------- .../hash/replication/ReplicationHub.java | 7 +-- .../chronicle/map/IORuntimeException.java | 6 +- .../chronicle/map/StatelessChronicleMap.java | 6 +- .../openhft/chronicle/map/TcpReplicator.java | 22 +++++-- .../openhft/chronicle/map/MapByNameTest.java | 8 +-- 6 files changed, 70 insertions(+), 37 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java index fa39cc671..6c9d23f8c 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -24,11 +24,17 @@ import java.io.Serializable; /** * Common configurations of {@link SingleChronicleHashReplication} and {@link ReplicationHub}. */ -public abstract class AbstractReplication implements Serializable{ +public abstract class AbstractReplication implements Serializable { private final byte localIdentifier; - private final @Nullable TcpTransportAndNetworkConfig tcpConfig; - private final @Nullable UdpTransportConfig udpConfig; - private final @Nullable RemoteNodeValidator remoteNodeValidator; + private final + @Nullable + TcpTransportAndNetworkConfig tcpConfig; + private final + @Nullable + UdpTransportConfig udpConfig; + private final + @Nullable + RemoteNodeValidator remoteNodeValidator; // package-private to forbid subclassing from outside of the package AbstractReplication(byte localIdentifier, Builder builder) { @@ -38,6 +44,16 @@ public abstract class AbstractReplication implements Serializable{ remoteNodeValidator = builder.remoteNodeValidator; } + + @Override + public String toString() { + return ", localIdentifier=" + localIdentifier + + ", tcpConfig=" + tcpConfig + + ", udpConfig=" + udpConfig + + ", remoteNodeValidator=" + remoteNodeValidator; + + } + public byte identifier() { return localIdentifier; } @@ -58,13 +74,13 @@ public abstract class AbstractReplication implements Serializable{ } /** - * Builder of {@link AbstractReplication} configurations. This class and it's subclasses are - * mutable, configuration methods mutate the builder and return it back for convenient chaining. + * Builder of {@link AbstractReplication} configurations. This class and it's subclasses are mutable, + * configuration methods mutate the builder and return it back for convenient chaining. * - * @param the concrete {@link AbstractReplication} subclass: {@link - * SingleChronicleHashReplication} or {@link ReplicationHub} - * @param the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or - * {@link ReplicationHub.Builder} + * @param the concrete {@link AbstractReplication} subclass: {@link SingleChronicleHashReplication} or + * {@link ReplicationHub} + * @param the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or {@link + * ReplicationHub.Builder} */ public static abstract class Builder> { private TcpTransportAndNetworkConfig tcpConfig = null; @@ -72,7 +88,8 @@ public abstract class AbstractReplication implements Serializable{ private RemoteNodeValidator remoteNodeValidator = null; // package-private to forbid subclassing from outside of the package - Builder() {} + Builder() { + } @NotNull public B tcpTransportAndNetwork(@Nullable TcpTransportAndNetworkConfig tcpConfig) { @@ -81,8 +98,8 @@ public abstract class AbstractReplication implements Serializable{ } /** - * Configures UDP transport settings, used by Replications, created by this builder. - * {@code null} means that UDP transport shouldn't being used. + * Configures UDP transport settings, used by Replications, created by this builder. {@code null} + * means that UDP transport shouldn't being used. * * @param udpConfig the new UDP transport config for replications, created by this builder. * @return this builder back, for chaining @@ -106,10 +123,10 @@ public abstract class AbstractReplication implements Serializable{ * @param identifier the node (server) identifier of the returned replication * @return a new Replication instance with the specified node (server) identifier * @throws IllegalArgumentException if the given identifier is non-positive - * @throws IllegalStateException if neither {@link #tcpTransportAndNetwork( - * TcpTransportAndNetworkConfig)} nor {@link #udpTransport(UdpTransportConfig)} are - * configured to non-{@code null}. At least one of the transport-level configs should be - * specified. + * @throws IllegalStateException if neither {@link #tcpTransportAndNetwork(TcpTransportAndNetworkConfig)} + * nor {@link #udpTransport(UdpTransportConfig)} are configured to + * non-{@code null}. At least one of the transport-level configs + * should be specified. */ @NotNull public abstract R createWithId(byte identifier); @@ -120,12 +137,11 @@ public abstract class AbstractReplication implements Serializable{ " given"); } + @Override public String toString() { - return - "tcpConfig=" + tcpConfig + - ", udpConfig=" + udpConfig + - ", remoteNodeValidator=" + remoteNodeValidator ; + return ", udpConfig=" + udpConfig + + ", remoteNodeValidator=" + remoteNodeValidator; } } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java index 866d7fa17..6d8352785 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -72,11 +72,10 @@ public final class ReplicationHub extends AbstractReplication { @Override public String toString() { - return "ReplicationHub{" + + + return "ReplicationHub{" + super.toString() + "maxEntrySize=" + maxEntrySize + - ", channels=" + Arrays.toString(channels) + - super.toString() + - '}'; + ", channels=" + Arrays.toString(channels) + '}'; } diff --git a/src/main/java/net/openhft/chronicle/map/IORuntimeException.java b/src/main/java/net/openhft/chronicle/map/IORuntimeException.java index e58603bd8..f52f21277 100644 --- a/src/main/java/net/openhft/chronicle/map/IORuntimeException.java +++ b/src/main/java/net/openhft/chronicle/map/IORuntimeException.java @@ -23,9 +23,13 @@ import java.io.IOException; /** * @author Rob Austin. */ -class IORuntimeException extends RuntimeException{ +class IORuntimeException extends RuntimeException { public IORuntimeException(IOException e) { super(e); } + + public IORuntimeException(String message) { + super(message); + } } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 09937598c..c1935cdbe 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -996,7 +996,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { while (bytes.buffer().position() < requiredNumberOfBytes) { - clientChannel.read(bytes.buffer()); + int len = clientChannel.read(bytes.buffer()); + + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + checkTimeout(timeoutTime); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 2510ff683..4a262a53a 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -589,11 +589,22 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); - if (attached.entryWriter.isWorkIncomplete()) - return; - try { - if (attached.entryReader.readSocketToBuffer(socketChannel) <= 0) + + int len = attached.entryReader.readSocketToBuffer(socketChannel); + if (len == -1) { + socketChannel.shutdownInput(); + socketChannel.shutdownOutput(); + + //todo we should call close, but if we do it breaks the bootstrap test case, further + // investigation required + return; + } + + if (len == 0) + return; + + if (attached.entryWriter.isWorkIncomplete()) return; } catch (IOException e) { @@ -1108,9 +1119,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // its set to MIN_VALUE when it should be read again if (state == NOT_SET) { - if (out.remaining() < SIZE_OF_SIZE + 1) { + if (out.remaining() < SIZE_OF_SIZE + 1) return; - } // state is used for both heartbeat and stateless state = out.readByte(); diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index cd27d71ec..196d0b158 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -94,17 +94,17 @@ public class MapByNameTest { final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); Thread.sleep(2000); -/* ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence + ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .minSegments(2) .entries(2) .name("myMap5") .removeReturnsNull(true); - ChronicleMap map = mapByName.create(builder); - map.put("hello", "world");*/ + ChronicleMap map = mapByName.create(builder); + map.put("hello", "world"); - final ChronicleMap map = mapByName.from("myMap5"); + // final ChronicleMap map = mapByName.from("myMap5"); // allow time for replication