From a72347eebbb7cf0b5ee83a0cca09f0a14859dfdd Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 9 Mar 2015 20:15:00 +0000 Subject: [PATCH] now supports the creating of channels --- .../hash/ChronicleHashInstanceBuilder.java | 2 + .../chronicle/map/ChannelProvider.java | 24 ++++-- .../chronicle/map/ChronicleMapBuilder.java | 16 +++- .../net/openhft/chronicle/map/Replicator.java | 4 +- .../openhft/chronicle/map/Replicators.java | 33 ++++---- .../map/StatelessWiredConnector.java | 83 ++++++++++++++++--- .../openhft/chronicle/map/TcpReplicator.java | 43 +++++----- .../map/WiredStatelessChronicleMap.java | 18 +++- .../net/openhft/chronicle/wire/TestWire.java | 45 ++++++++++ 9 files changed, 200 insertions(+), 68 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java index 1ec13b337..2eb4c90b6 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java @@ -57,4 +57,6 @@ public interface ChronicleHashInstanceBuilder { ChronicleHashInstanceBuilder name(String name); H create() throws IOException; + + } diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index c5ba83b94..8a8e3fe19 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -38,6 +38,7 @@ import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.atomic.AtomicReferenceArray; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import static java.lang.Math.min; import static java.nio.ByteBuffer.wrap; @@ -53,7 +54,8 @@ final class ChannelProvider implements Closeable { static final Map implMapping = new IdentityHashMap<>(); - static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException { + static synchronized ChannelProvider getProvider(ReplicationHub hub, + Supplier statelessWiredConnectorSupplier) throws IOException { ChannelProvider channelProvider = implMapping.get(hub); if (channelProvider != null) return channelProvider; @@ -70,7 +72,8 @@ final class ChannelProvider implements Closeable { hub.remoteNodeValidator(), null, hub.name(), - channelProvider.chronicleChannelList()); + channelProvider.chronicleChannelList(), + statelessWiredConnectorSupplier); channelProvider.add(tcpReplicator); } @@ -547,13 +550,6 @@ final class ChannelProvider implements Closeable { return localIdentifier; } - @Override - protected Closeable applyTo(ChronicleMapBuilder builder, - Replica map, EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) { - add(chronicleChannel, map, entryExternalizable); - return this; - } @Override public void close() throws IOException { @@ -570,5 +566,15 @@ final class ChannelProvider implements Closeable { channelDataLock.writeLock().unlock(); } } + + @Override + protected Closeable applyTo(ChronicleMapBuilder builder, + Replica map, + EntryExternalizable entryExternalizable, + ChronicleMap chronicleMap, + Supplier statelessWiredConnectorSupplier) throws IOException { + add(chronicleChannel, map, entryExternalizable); + return this; + } } } \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index ab7bbe7ef..03f1e0cea 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -22,12 +22,12 @@ import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.ChronicleHashBuilderImpl; import net.openhft.chronicle.hash.impl.hashlookup.HashLookup; -import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder; import net.openhft.chronicle.hash.replication.*; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; import net.openhft.chronicle.hash.serialization.internal.MetaProvider; +import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder; import net.openhft.chronicle.set.ChronicleSetBuilder; import net.openhft.lang.Maths; import net.openhft.lang.io.Bytes; @@ -54,6 +54,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; import static java.lang.Math.round; import static net.openhft.chronicle.hash.impl.util.Objects.builderEquals; @@ -1458,14 +1459,21 @@ public final class ChronicleMapBuilder implements if (singleHashReplication.udpTransport() != null) replicators.add(Replicators.udp(singleHashReplication.udpTransport())); } else { - ReplicationHub hub = channel.hub(); + final ReplicationHub hub = channel.hub(); - ChannelProvider provider = ChannelProvider.getProvider(hub); + Supplier statelessWiredConnectorSupplier = + () -> new StatelessWiredConnector(instance(), hub); + + ChannelProvider provider = ChannelProvider.getProvider(hub, statelessWiredConnectorSupplier); ChannelProvider.ChronicleChannel ch = provider.createChannel(channel.channelId()); replicators.add(ch); } for (Replicator replicator : replicators) { - Closeable token = replicator.applyTo(this, result, result, map); + + Supplier statelessWiredConnectorSupplier = + () -> new StatelessWiredConnector(instance(), channel.hub()); + + Closeable token = replicator.applyTo(this, result, result, map, statelessWiredConnectorSupplier); if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) { LOG.warn(Replicators.ONLY_UDP_WARN_MESSAGE); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java index a6307c3f2..48112095c 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ b/src/main/java/net/openhft/chronicle/map/Replicator.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; import java.io.Closeable; import java.io.IOException; +import java.util.function.Supplier; abstract class Replicator { @@ -49,5 +50,6 @@ abstract class Replicator { */ protected abstract Closeable applyTo(ChronicleMapBuilder builder, Replica map, Replica.EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) throws IOException; + final ChronicleMap chronicleMap, + Supplier statelessWiredConnectorSupplier) throws IOException; } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index 3b235658d..2b1587ac5 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -27,6 +27,7 @@ import java.io.Closeable; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import java.util.function.Supplier; final class Replicators { @@ -111,25 +112,28 @@ final class Replicators { @Override - protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder, - @NotNull final Replica replica, - @NotNull final Replica.EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) throws IOException { + protected Closeable applyTo(ChronicleMapBuilder builder, + Replica map, + Replica.EntryExternalizable entryExternalizable, + ChronicleMap chronicleMap, + Supplier statelessWiredConnectorSupplier) throws IOException { TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork(); TcpReplicator.StatelessClientParameters statelessClientParameters = new TcpReplicator.StatelessClientParameters( - (VanillaChronicleMap) chronicleMap, - builder.keyBuilder, - builder.valueBuilder); + (VanillaChronicleMap) chronicleMap, + builder.keyBuilder, + builder.valueBuilder); - return new TcpReplicator(replica, entryExternalizable, + return new TcpReplicator(map, entryExternalizable, tcpConfig, replication.remoteNodeValidator(), statelessClientParameters, replication.name(), - null); + null, statelessWiredConnectorSupplier); } + + }; } @@ -137,14 +141,15 @@ final class Replicators { final UdpTransportConfig replicationConfig) { return new Replicator() { @Override - protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder, - @NotNull final Replica map, - @NotNull final Replica.EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) - throws IOException { + protected Closeable applyTo(ChronicleMapBuilder builder, + Replica map, + Replica.EntryExternalizable entryExternalizable, + ChronicleMap chronicleMap, + Supplier statelessWiredConnectorSupplier) throws IOException { return new UdpReplicator(map, entryExternalizable, replicationConfig ); } + }; } } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 98c8a16f6..d27045ff9 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -24,7 +24,9 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.bytes.BytesMarshallable; +import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.util.BuildVersion; +import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.ValueIn; import net.openhft.chronicle.wire.Wire; @@ -45,6 +47,7 @@ import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; import java.util.*; +import java.util.concurrent.Callable; import java.util.function.Consumer; import java.util.function.Function; @@ -70,13 +73,16 @@ class StatelessWiredConnector> chronicleHashInstanceBuilder; + + private byte localIdentifier; + private long timestamp; private short channelId; - @NotNull - private final List channelList; + private List channelList; + private ReplicationHub hub; final Consumer writeElement = new Consumer>() { @@ -108,19 +114,27 @@ class StatelessWiredConnector channelList, byte localIdentifier) { - - if (channelList == null) { - throw new IllegalStateException("The StatelessWiredConnector currently only support maps " + - "that are set up via the replication channel. localIdentifier=" + localIdentifier); - } - this.channelList = channelList; - this.localIdentifier = localIdentifier; + public StatelessWiredConnector(ChronicleHashInstanceBuilder> chronicleHashInstanceBuilder, ReplicationHub hub) { + this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder; + this.hub = hub; } private SelectionKey key; + public void setChannelList(@NotNull List channelList) { + if (channelList == null) + throw new IllegalStateException("The StatelessWiredConnector currently only support maps " + + "that are set up via the replication channel. localIdentifier=" + localIdentifier); + + this.channelList = channelList; + } + + public void setLocalIdentifier(byte localIdentifier) { + this.localIdentifier = localIdentifier; + } + + public void onWrite(@NotNull SocketChannel socketChannel, @NotNull SelectionKey key) throws IOException { if (out != null) { @@ -312,7 +326,6 @@ class StatelessWiredConnector map.keySet().iterator(), writeElement); return; @@ -337,6 +350,17 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(transactionId); + if ("CREATE_CHANNEL".contentEquals(methodName)) { + + writeVoid(() -> { + short channelId1 = inWire.read(() -> "ARG_1").int16(); + chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create(); + return null; + }); + return; + } + + if ("REMOTE_IDENTIFIER".contentEquals(methodName)) { this.remoteIdentifier = inWire.read(() -> "RESULT").int8(); return; @@ -719,6 +743,37 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + + try { + r.call(); + } catch (Exception e) { + outWire.bytes().reset(); + // the idea of wire is that is platform independent, + // so we wil have to send the exception as a String + outWire.write(() -> "IS_EXCEPTION").bool(true); + outWire.write(() -> "EXCEPTION").text(toString(e)); + LOG.error("", e); + return; + } + + nofityDataWritten(); + + } + + /** * only used for debugging */ @@ -759,6 +814,10 @@ class StatelessWiredConnector extends AbstractChannelReplicator implements Clo private final BitSet activeKeys = new BitSet(selectionKeysStore.length); private final long heartBeatIntervalMillis; - @Nullable - final List channelList; private long largestEntrySoFar = 128; @@ -110,6 +109,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo StatelessClientParameters statelessClientParameters; + private Supplier statelessWiredConnectorSupplier; + private List channelList; + static class StatelessClientParameters { public StatelessClientParameters(VanillaChronicleMap map, SerializationBuilder keySerializationBuilder, @@ -133,7 +135,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo @Nullable final RemoteNodeValidator remoteNodeValidator, @Nullable final StatelessClientParameters statelessClientParameters, String name, - @Nullable final List channelList) + @Nullable final List channelList, + @NotNull Supplier statelessWiredConnectorSupplier) throws IOException { super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig()); @@ -141,6 +144,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo this.statelessClientParameters = statelessClientParameters; this.channelList = channelList; + this.statelessWiredConnectorSupplier = statelessWiredConnectorSupplier; final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS); @@ -156,6 +160,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo this.remoteNodeValidator = remoteNodeValidator; this.name = name; + + start(); } @@ -810,7 +816,16 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (b == WIRED_CONNECTION) attached.useWire = true; forceRead = true; - attached.statelessWiredConnector = new StatelessWiredConnector(channelList, localIdentifier); + + attached.statelessWiredConnector = statelessWiredConnectorSupplier.get(); + + if (attached.statelessWiredConnector != null) { + attached.statelessWiredConnector.localIdentifier(localIdentifier); + attached.statelessWiredConnector.setChannelList(channelList); + } else { + LOG.error("", new RuntimeException("statelessWiredConnector not set up")); + } + attached.handShakingComplete = true; } @@ -1411,26 +1426,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo * @throws InterruptedException */ void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) { -/* - if (attached.useWire) { - - if (out.remaining() >= 4) { - int size = out.readInt(out.position()); - if (out.remaining() >= size) { - - long end = out.position() + size; - - try { - out.skip(4); // for the size - wiredStatelessServer.processStatelessEvent(out, attached.entryWriter); - } finally { - out.position(end); - } - - } - } - return; - }*/ int entriesRead = 0; try { diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 0392ee17a..ca26ef322 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -33,6 +33,7 @@ import java.util.function.Consumer; import java.util.function.Predicate; import static java.util.Collections.emptyList; +import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.TO_STRING; import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*; @@ -269,6 +270,15 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); } + @NotNull + public void createChannel(short channelID) { + proxyReturnVoid(() -> { + writeField(METHOD_NAME, "CREATE_CHANNEL"); + hub.outWire().write(() -> "ARG_1").int16(channelID); + }); + } + + public void putAll(@NotNull Map map) { final long startTime = System.currentTimeMillis(); @@ -341,7 +351,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } public void clear() { - proxyReturnVoid(CLEAR.toString()); + proxyReturnVoid(() -> writeField(METHOD_NAME, CLEAR.toString())); } @@ -674,16 +684,16 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } @SuppressWarnings("SameParameterValue") - private void proxyReturnVoid(@NotNull final String methodName) { + private void proxyReturnVoid(Runnable r) { + final long startTime = System.currentTimeMillis(); - long transactionId; hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); + r.run(); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index 64db99633..b16c79e03 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -47,6 +47,51 @@ public class TestWire { private ReplicationHub hubA; + @Test + public void testCreateChannel() throws IOException, InterruptedException { + + { + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(++SERVER_PORT) + .heartBeatInterval(1, SECONDS); + + hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) + .createWithId((byte) 1); + + // this is how you add maps after the custer is created + map1a = of(byte[].class, byte[].class) + .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + + + byte identifier = (byte) 2; + + final WiredChronicleMapStatelessClientBuilder builder = + new WiredChronicleMapStatelessClientBuilder<>( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + String.class, + (short) 1); + + builder.identifier(identifier); + final ChronicleMap statelessMap = builder.create(); + ((WiredStatelessChronicleMap) statelessMap).createChannel((short) 2); + + + ChronicleMap map2 = new WiredChronicleMapStatelessClientBuilder( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + String.class, + (short) 2).create(); + + map2.put("hello", "world"); + + assertEquals(1, map2.size()); + + map1a.close(); + } + + } + @Test public void testSimplePutStringWithChannels() throws IOException, InterruptedException {