From d76691fcaa4bcba181ef860d1e285149ccdfdba3 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 12 Jan 2015 16:25:45 +0000 Subject: [PATCH] HCOLL-134 added a serialization test --- .../hash/replication/AbstractReplication.java | 29 ++--- .../hash/replication/UdpTransportConfig.java | 4 + .../chronicle/map/MapInstanceBuilder.java | 8 +- .../chronicle/map/UdpChannelReplicator.java | 110 ++++++++++++------ .../openhft/chronicle/map/MapByNameTest.java | 21 ++++ 5 files changed, 121 insertions(+), 51 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 e85fec8ff..4ad1d40d4 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java @@ -34,7 +34,7 @@ public abstract class AbstractReplication implements Serializable { UdpTransportConfig udpConfig; private final @Nullable - RemoteNodeValidator remoteNodeValidator; + transient RemoteNodeValidator remoteNodeValidator; private final boolean bootstrapOnlyLocalEntries; // package-private to forbid subclassing from outside of the package @@ -79,13 +79,14 @@ 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; @@ -104,8 +105,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 @@ -135,8 +136,8 @@ public abstract class AbstractReplication implements Serializable { * currently running. * * @param bootstrapOnlyLocalEntries if a node provided with this replication should - * replicate only local data to the new nodes joining - * the replication network + * replicate only local data to the new nodes joining the + * replication network * @return this builder back */ @NotNull @@ -153,9 +154,9 @@ public abstract class AbstractReplication implements Serializable { * @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. + * 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); diff --git a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java index b253e7bff..eb78687a3 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java @@ -37,6 +37,7 @@ public final class UdpTransportConfig { private final ThrottlingConfig throttlingConfig; private final int udpBufferSize; + private String name; private UdpTransportConfig( InetAddress address, @@ -156,4 +157,7 @@ public final class UdpTransportConfig { return create(address(), port(), networkInterface(), throttlingConfig(), udpBufferSize); } + public String name() { + return this.name; + } } diff --git a/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java b/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java index 6e201f834..d9f0a33a5 100644 --- a/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java @@ -27,12 +27,12 @@ import java.io.Serializable; import java.util.concurrent.atomic.AtomicBoolean; final class MapInstanceBuilder - implements ChronicleHashInstanceBuilder>, Serializable{ + implements ChronicleHashInstanceBuilder>, Serializable { final ChronicleMapBuilder mapBuilder; - SingleChronicleHashReplication singleHashReplication; - ReplicationChannel channel; - File file; + transient SingleChronicleHashReplication singleHashReplication; + transient ReplicationChannel channel; + transient File file; String name; final AtomicBoolean used; diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index bd4c691e9..98501725b 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -25,10 +25,8 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.NetworkInterface; -import java.net.StandardSocketOptions; +import java.net.*; +import java.nio.BufferUnderflowException; import java.nio.channels.*; import java.util.Set; @@ -61,6 +59,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. private SelectableChannel writeChannel; private volatile boolean shouldEnableOpWrite; + private final String name; /** * @param replicationConfig @@ -79,6 +78,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. address = replicationConfig.address(); port = replicationConfig.port(); networkInterface = replicationConfig.networkInterface(); + name = replicationConfig.name(); serverConnector = new ServerConnector(); } @@ -99,13 +99,15 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. /** * binds to the server socket and process data This method will block until interrupted */ + + @Override void processEvent() throws IOException { - - connectClient().register(selector, OP_READ); - serverConnector.connectLater(); - try { + + connectClient().register(selector, OP_READ); + serverConnector.connectLater(); + while (selector.isOpen()) { registerPendingRegistrations(); @@ -122,46 +124,88 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. continue; // nothing to do } - final Set selectionKeys = selector.selectedKeys(); - for (final SelectionKey key : selectionKeys) { + if (useJavaNIOSelectionKeys) { + // use the standard java nio selector + + final Set selectionKeys = selector.selectedKeys(); + for (final SelectionKey key : selectionKeys) { + processKey(key); + } + selectionKeys.clear(); + } else { + // use the netty like selector + + final SelectionKey[] keys = selectedKeys.flip(); + try { + for (int i = 0; i < keys.length && keys[i] != null; i++) { + final SelectionKey key = keys[i]; - if (key.isReadable()) { - final DatagramChannel socketChannel = (DatagramChannel) key.channel(); - reader.readAll(socketChannel); - } - - if (key.isWritable()) { - final DatagramChannel socketChannel = (DatagramChannel) key.channel(); try { - int bytesJustWritten = writer.writeAll(socketChannel); - contemplateThrottleWrites(bytesJustWritten); - } catch (NotYetConnectedException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - serverConnector.connectLater(); - } catch (IOException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - serverConnector.connectLater(); + processKey(key); + } catch (BufferUnderflowException e) { + if (!isClosed) + LOG.error("", e); } } - } catch (Exception e) { - LOG.error("", e); - if (!isClosed) - closeEarlyAndQuietly(key.channel()); + } finally { + for (int i = 0; i < keys.length && keys[i] != null; i++) { + keys[i] = null; + } } } - - selectionKeys.clear(); } + } catch (CancelledKeyException | ConnectException | ClosedChannelException | + ClosedSelectorException e) { + if (LOG.isDebugEnabled()) + LOG.debug("", e); + } catch (Exception e) { + LOG.error("", e); + } catch (Throwable e) { + LOG.error("", e); + throw e; } finally { + + if (LOG.isDebugEnabled()) + LOG.debug("closing name=" + this.name); if (!isClosed) { closeResources(); } } } + private void processKey(SelectionKey key) { + + try { + + if (key.isReadable()) { + final DatagramChannel socketChannel = (DatagramChannel) key.channel(); + reader.readAll(socketChannel); + } + + if (key.isWritable()) { + final DatagramChannel socketChannel = (DatagramChannel) key.channel(); + try { + int bytesJustWritten = writer.writeAll(socketChannel); + contemplateThrottleWrites(bytesJustWritten); + } catch (NotYetConnectedException e) { + if (LOG.isDebugEnabled()) + LOG.debug("", e); + serverConnector.connectLater(); + } catch (IOException e) { + if (LOG.isDebugEnabled()) + LOG.debug("", e); + serverConnector.connectLater(); + } + } + } catch (Exception e) { + LOG.error("", e); + if (!isClosed) + closeEarlyAndQuietly(key.channel()); + } + + } + private DatagramChannel connectClient() throws IOException { final DatagramChannel client = address.isMulticastAddress() ? DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) : diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 609ad0696..6ac2e3b72 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -87,6 +87,27 @@ public class MapByNameTest { ChronicleMap myMap2 = mapByName.from("myMap"); } + @Test + public void testReplicationHubSerialization() throws IOException, InterruptedException, + TimeoutException { + NodeDiscovery nodeDiscovery = new NodeDiscovery(); + + final ReplicationHubFindByName mapByName = nodeDiscovery.mapByName(); + Thread.sleep(2000); + + MapInstanceBuilder config = + (MapInstanceBuilder) + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .minSegments(2) + .entries(2) + .removeReturnsNull(true) + .instance() + .name("myMap6"); + + mapByName.create(config); + + } + public static void main(String... args) throws IOException, InterruptedException, TimeoutException { NodeDiscovery nodeDiscovery = new NodeDiscovery();