From 6037bd27e439136b6529d18849f6bce05c85ff20 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 9 Apr 2015 13:09:25 +0100 Subject: [PATCH] CE-20 moved the map lookup by name from engine to the map code ( but its a bit of a hack and requires implementing properly, but for the moment this demonstrates the API ) --- .../chronicle/map/ChannelProvider.java | 44 +++---- ...ientWiredChronicleMapStatelessBuilder.java | 48 ++------ .../map/ClientWiredStatelessChronicleMap.java | 9 +- .../ClientWiredStatelessTcpConnectionHub.java | 21 ++-- .../openhft/chronicle/map/MapWireHandler.java | 113 +++++++++++++++--- .../chronicle/map/MapWireHandlerBuilder.java | 17 ++- .../chronicle/map/StatelessChronicleMap.java | 15 +-- .../map/TCPSocketReplication4WayMapTest.java | 2 + 8 files changed, 161 insertions(+), 108 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 25bbdb1f5..85866a274 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -174,9 +174,9 @@ public final class ChannelProvider implements Closeable { try { // old-style iteration to avoid 1) iterator object creation // 2) ConcurrentModificationException - for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + for (Replica chronicleChannel : chronicleChannelMap.values()) { final ModificationIterator modificationIterator = - chronicleChannelList.get(i).acquireModificationIterator( + chronicleChannel.acquireModificationIterator( remoteIdentifier, notifier); if (modificationIterator.hasNext()) return true; @@ -192,12 +192,14 @@ public final class ChannelProvider implements Closeable { final int na) { channelDataLock.readLock().lock(); try { - for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { - Replica chronicleChannel = chronicleChannelList.get(i); - final ModificationIterator modificationIterator = chronicleChannel + + for (Map.Entry chronicleChannel : chronicleChannelMap.entrySet()) { + + final ModificationIterator modificationIterator = + chronicleChannel.getValue() .acquireModificationIterator(remoteIdentifier, notifier); if (modificationIterator - .nextEntry(callback, chronicleChannelIds.get(i))) + .nextEntry(callback, chronicleChannel.getKey())) return true; } return false; @@ -210,9 +212,9 @@ public final class ChannelProvider implements Closeable { public void dirtyEntries(long fromTimeStamp) { channelDataLock.readLock().lock(); try { - for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { - chronicleChannelList.get(i) - .acquireModificationIterator(remoteIdentifier, notifier) + for (Replica chronicleChannel : chronicleChannelMap.values()) { + + chronicleChannel.acquireModificationIterator(remoteIdentifier, notifier) .dirtyEntries(fromTimeStamp); notifier.onChange(); } @@ -238,8 +240,7 @@ public final class ChannelProvider implements Closeable { try { long t = 0; // not including the SystemQueue at index 0 - for (int i = 1, len = chronicleChannelList.size(); i < len; i++) { - Replica channel = chronicleChannelList.get(i); + for (Replica channel : chronicleChannelMap.values()) { t = (t == 0) ? channel.lastModificationTime(remoteIdentifier) : min(t, channel.lastModificationTime(remoteIdentifier)); } @@ -262,7 +263,10 @@ public final class ChannelProvider implements Closeable { // start of channel data private final Replica[] chronicleChannels; - private final List chronicleChannelList; + + // todo perhaps replace with a koloboke map + private final Map chronicleChannelMap; + private final List chronicleChannelIds; private final EntryExternalizable[] channelEntryExternalizables; private final AtomicReferenceArray systemModificationIterator = @@ -271,8 +275,8 @@ public final class ChannelProvider implements Closeable { newBitSet(systemModificationIterator.length()); private final AtomicReferenceArray modificationIterator = new AtomicReferenceArray(128); - // end of channel data + // end of channel data private final Set replicators = new CopyOnWriteArraySet(); private volatile boolean isClosed = false; @@ -284,8 +288,8 @@ public final class ChannelProvider implements Closeable { chronicleChannels = new Replica[hub.maxNumberOfChannels()]; channelEntryExternalizables = new EntryExternalizable[hub.maxNumberOfChannels()]; - chronicleChannelList = new ArrayList(); - chronicleChannelIds = new ArrayList(); + chronicleChannelMap = new HashMap<>(); + chronicleChannelIds = new ArrayList<>(); MessageHandler systemMessageHandler = new MessageHandler() { @Override public void onMessage(Bytes bytes) { @@ -361,7 +365,7 @@ public final class ChannelProvider implements Closeable { " is already in use."); } chronicleChannels[chronicleChannel] = replica; - chronicleChannelList.add(replica); + chronicleChannelMap.put(chronicleChannel, replica); chronicleChannelIds.add(chronicleChannel); channelEntryExternalizables[chronicleChannel] = entryExternalizable; @@ -411,8 +415,8 @@ public final class ChannelProvider implements Closeable { replicators.add(replicator); } - public List chronicleChannelList() { - return Collections.unmodifiableList(chronicleChannelList); + public Map chronicleChannelMap() { + return Collections.unmodifiableMap(chronicleChannelMap); } private interface MessageHandler { @@ -552,12 +556,12 @@ public final class ChannelProvider implements Closeable { public void close() throws IOException { channelDataLock.writeLock().lock(); try { - chronicleChannelList.remove(chronicleChannels[chronicleChannel]); + chronicleChannelMap.remove(chronicleChannel); chronicleChannelIds.remove((Integer) (int) chronicleChannel); chronicleChannels[chronicleChannel] = null; channelEntryExternalizables[chronicleChannel] = null; - if (chronicleChannelList.size() == 1) // i. e. only SystemQueue is left + if (chronicleChannelMap.size() == 1) // i. e. only SystemQueue is left ChannelProvider.this.close(); } finally { channelDataLock.writeLock().unlock(); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java index cc89cdec1..be59a6ab0 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java @@ -35,31 +35,21 @@ public final class ClientWiredChronicleMapStatelessBuilder implements MapBuilder> { ClientWiredStatelessTcpConnectionHub hub; + private Class keyClass; private Class valueClass; - private byte localIdentifier; - private short channelID; - private boolean doHandShaking; public ClientWiredChronicleMapStatelessBuilder( - InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) { - this.keyClass = keyClass; - this.valueClass = valueClass; - this.remoteAddress = remoteAddress; - this.channelID = channelID; - } - - public ClientWiredChronicleMapStatelessBuilder( - ClientWiredStatelessTcpConnectionHub hub, Class keyClass, Class valueClass, - short channelID) { + ClientWiredStatelessTcpConnectionHub hub, + Class keyClass, + Class valueClass, + String name) { this.keyClass = keyClass; this.valueClass = valueClass; this.hub = hub; - this.channelID = channelID; + this.name = name; } - - private InetSocketAddress remoteAddress; private boolean putReturnsNull = false; private boolean removeReturnsNull = false; private long timeoutMs = TimeUnit.SECONDS.toMillis(10); @@ -68,11 +58,6 @@ public final class ClientWiredChronicleMapStatelessBuilder implements private final AtomicBoolean used = new AtomicBoolean(false); - - InetSocketAddress remoteAddress() { - return remoteAddress; - } - @Override public ClientWiredChronicleMapStatelessBuilder putReturnsNull(boolean putReturnsNull) { this.putReturnsNull = putReturnsNull; @@ -100,10 +85,6 @@ public final class ClientWiredChronicleMapStatelessBuilder implements return this; } - long timeoutMs() { - return timeoutMs; - } - @Override public ClientWiredChronicleMapStatelessBuilder name(String name) { this.name = name; @@ -127,13 +108,9 @@ public final class ClientWiredChronicleMapStatelessBuilder implements @Override public ChronicleMap create() throws IOException { - // todo clean this up - if (hub == null) - hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking, "MAP"); - if (!used.getAndSet(true)) { return new ClientWiredStatelessChronicleMap( - this, keyClass, valueClass, channelID); + this, keyClass, valueClass, name); } else { throw new IllegalStateException( @@ -143,16 +120,5 @@ public final class ClientWiredChronicleMapStatelessBuilder implements } } - public void identifier(byte identifier) { - this.localIdentifier = identifier; - } - public short channelID() { - return channelID; - } - - - public ClientWiredStatelessTcpConnectionHub hub() { - return hub; - } } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index a5eee0f03..fcb976890 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -60,7 +60,6 @@ class ClientWiredStatelessChronicleMap private boolean putReturnsNull; private boolean removeReturnsNull; - private short channelID; // used with toString() private static final int MAX_NUM_ENTRIES = 20; @@ -68,9 +67,11 @@ class ClientWiredStatelessChronicleMap public ClientWiredStatelessChronicleMap( @NotNull final ClientWiredChronicleMapStatelessBuilder config, - @NotNull final Class kClass, @NotNull final Class vClass, short channelID) { - this.channelID = channelID; - this.csp = "//server/path/" + channelID + "#MAP"; + @NotNull final Class kClass, + @NotNull final Class vClass, + @NotNull final String channelName) { + + this.csp = "//" + channelName + "#MAP"; this.hub = config.hub; this.putReturnsNull = config.putReturnsNull(); this.removeReturnsNull = config.removeReturnsNull(); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 15e572156..1e3648377 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -61,7 +61,7 @@ public class ClientWiredStatelessTcpConnectionHub { @NotNull private final AtomicLong transactionID = new AtomicLong(0); - private final ClientWiredChronicleMapStatelessBuilder config; + @Nullable protected CloseablesManager closeables; @@ -87,15 +87,20 @@ public class ClientWiredStatelessTcpConnectionHub { public ClientWiredStatelessTcpConnectionHub( - ClientWiredChronicleMapStatelessBuilder config, byte localIdentifier, - boolean doHandShaking, String cspType) { + byte localIdentifier, + boolean doHandShaking, + String cspType, + InetSocketAddress remoteAddress, + int tcpBufferSize, + long timeout) { + this.localIdentifier = localIdentifier; this.doHandShaking = doHandShaking; - this.remoteAddress = config.remoteAddress(); - this.tcpBufferSize = config.tcpBufferSize(); - this.config = config; - this.name = config.name(); - this.timeoutMs = config.timeoutMs(); + + this.tcpBufferSize = tcpBufferSize; + this.remoteAddress = remoteAddress; + this.name = cspType + " connected to " + remoteAddress.toString(); + this.timeoutMs = timeout; this.type = cspType; attemptConnect(remoteAddress); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index d73b7cf79..10ae4a5b5 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -37,11 +37,15 @@ import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.io.PrintWriter; import java.io.StreamCorruptedException; import java.io.StringWriter; import java.nio.BufferOverflowException; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; import java.util.concurrent.Callable; import java.util.function.Consumer; import java.util.function.Function; @@ -56,14 +60,16 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; */ class MapWireHandler implements WireHandler, Consumer { - public static final int SIZE_OF_SIZE = 2; private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); - final Map incompleteWork = new HashMap(); + public static final int MAP_SERVICE = 3; + private final Map incompleteWork = new HashMap(); private final ArrayList bytesChronicleMaps = new ArrayList<>(); - @NotNull + @NotNull private final Supplier>> mapFactory; + private final Supplier>> channelNameToIdFactory; + private Wire inWire = null; private Wire outWire = null; @@ -93,26 +99,44 @@ class MapWireHandler implements WireHandler, Consumer { bytes1.write(entry.getValue()); } }; + private WireHandlers publishLater; private byte localIdentifier; - private short channelId; - private List channels; + private int channelId; + private Map channels; private ReplicationHub hub; private byte remoteIdentifier; + private ChronicleMap channelNameToId; public MapWireHandler( - @NotNull Supplier>> mapFactory, - @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List channels) { - this(mapFactory, hub); + @NotNull final Supplier>> mapFactory, + @NotNull final Supplier>> + channelNameToIdFactory, + @NotNull final ReplicationHub hub, byte localIdentifier, + @NotNull final Map channels) { + this(mapFactory, channelNameToIdFactory, hub); this.channels = channels; this.localIdentifier = localIdentifier; + } - public MapWireHandler(Supplier>> mapFactory, - ReplicationHub hub) { + public MapWireHandler( + @NotNull final Supplier>> mapFactory, + @NotNull final Supplier>> + channelNameToIdFactory, + @NotNull final ReplicationHub hub) { + this.mapFactory = mapFactory; + this.channelNameToIdFactory = channelNameToIdFactory; + this.hub = hub; + try { + channelNameToId = (ChronicleMap) channelNameToIdFactory.get() + .replicatedViaChannel(hub.createChannel(MAP_SERVICE)).create(); + } catch (IOException e) { + e.printStackTrace(); + } } @Override @@ -196,25 +220,75 @@ class MapWireHandler implements WireHandler, Consumer { } } - StringBuilder cspText = new StringBuilder(); + + /** + * @return the next free channel id + */ + short getNextFreeChannel() { + + // todo this is a horrible hack, it slow and not processor safe, but was added to get + // todo something working for now. + + int max = 3; + for (Integer channel : channelNameToId.values()) { + max = Math.max(max, channel); + } + + return (short) (max + 1); + } + + + /** + * gets the channel id for a name, or creates a new one if this name is not yet assosiated to a + * channel + * + * @param fromName the name of the channel + * @return the id assosiated with this name + */ + int acquireChannelId(@NotNull String fromName) throws IOException { + + // todo this is a horrible hack, it slow and NOT processor safe, but was added to get + // todo something working for now. + + final Integer channelId = channelNameToId.get(fromName); + + if (channelId != null) + return channelId; + + final int nextFreeChannel = getNextFreeChannel(); + try { + + mapFactory.get().replicatedViaChannel(hub.createChannel(nextFreeChannel)).create(); + channelNameToId.put(fromName, nextFreeChannel); + + return nextFreeChannel; + } catch (IOException e) { + // todo send this error back to the user + LOG.error("", e); + throw e; + } + + } + + + @SuppressWarnings("UnusedReturnValue") - void onEvent() throws StreamCorruptedException { + void onEvent() throws IOException { + StringBuilder cspText = Wires.acquireStringBuilder(); // it is assumed by this point that the buffer has all the bytes in it for this message - inWire.read(csp).text(cspText); final int i = cspText.lastIndexOf("/"); - if (i != -1 && i < (cspText.length() - 1)) { final String channelStr = cspText.substring(i + 1, cspText.length() - "#MAP".length()); - channelId = Short.parseShort(channelStr); + channelId = acquireChannelId(channelStr); + } else channelId = 0; - final long tid = inWire.read(Fields.tid).int64(); @@ -644,7 +718,7 @@ class MapWireHandler implements WireHandler, Consumer { * @return the chronicle map with this {@code channelId} */ @NotNull - private ReplicatedChronicleMap map(short channelId) { + private ReplicatedChronicleMap map(int channelId) { // todo this cast is a bit of a hack, improve later final ReplicatedChronicleMap map = @@ -653,6 +727,7 @@ class MapWireHandler implements WireHandler, Consumer { if (map != null) return map; + throw new IllegalStateException(); } @@ -663,7 +738,7 @@ class MapWireHandler implements WireHandler, Consumer { * @return a BytesChronicleMap used to update the memory which holds the chronicle map */ @Nullable - private BytesChronicleMap bytesMap(short channelId) { + private BytesChronicleMap bytesMap(int channelId) { final BytesChronicleMap bytesChronicleMap = (channelId < bytesChronicleMaps.size()) ? bytesChronicleMaps.get(channelId) diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index a431ef26d..d78c88c79 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -25,6 +25,7 @@ import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; import java.util.List; +import java.util.Map; import java.util.function.Supplier; /** @@ -34,12 +35,22 @@ public class MapWireHandlerBuilder { public static WireHandler of( @NotNull Supplier>> mapFactory, - @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List channels) { - return new MapWireHandler<>(mapFactory, hub, localIdentifier, channels); + @NotNull Supplier>> + channelNameToIdFactory, + @NotNull ReplicationHub hub, + byte localIdentifier, + @NotNull Map channels) { + + return new MapWireHandler<>( + mapFactory, + channelNameToIdFactory, + hub, + localIdentifier, + channels); } // note : peter has asked for these to be in camel case - public static enum Fields implements WireKey { + public enum Fields implements WireKey { hasNext, timeStamp, channelId, diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 6d21da108..ee3b75473 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -72,14 +72,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private ByteBuffer inBuffer; private ByteBufferBytes inBytes; - - @NotNull private ReaderWithSize keyReaderWithSize; - @NotNull private WriterWithSize keyWriterWithSize; - @NotNull private ReaderWithSize valueReaderWithSize; - @NotNull private WriterWithSize valueWriterWithSize; private SocketChannel clientChannel; @@ -98,15 +93,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final ReentrantLock inBytesLock = new ReentrantLock(true); private final ReentrantLock outBytesLock = new ReentrantLock(); - private final BufferResizer outBufferResizer = new BufferResizer() { - @Override - public Bytes resizeBuffer(int newCapacity) { - return resizeBufferOutBuffer(newCapacity); + private final BufferResizer outBufferResizer = newCapacity -> resizeBufferOutBuffer(newCapacity); - } - }; - - static enum EventId { + enum EventId { HEARTBEAT, STATEFUL_UPDATE, LONG_SIZE, diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java index 50113a515..07ea35751 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java @@ -20,6 +20,7 @@ package net.openhft.chronicle.map; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import java.io.Closeable; @@ -120,6 +121,7 @@ public class TCPSocketReplication4WayMapTest { assertTrue("map2.empty", !map2.isEmpty()); } + @Test public void testBufferOverflowPutIfAbsent() throws IOException, InterruptedException {