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 )

2021-09-13 01:26:12 +00:00 · 2015-04-09 13:09:25 +01:00
parent e0363defe5
commit 6037bd27e4
8 changed files with 161 additions and 108 deletions
@@ -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<Integer, Replica> 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<Replica> chronicleChannelList;
+
+    // todo perhaps replace with a koloboke map
+    private final Map<Integer, Replica> chronicleChannelMap;
+
    private final List<Integer> chronicleChannelIds;
    private final EntryExternalizable[] channelEntryExternalizables;
    private final AtomicReferenceArray<PayloadProvider> systemModificationIterator =
@@ -271,8 +275,8 @@ public final class ChannelProvider implements Closeable {
            newBitSet(systemModificationIterator.length());
    private final AtomicReferenceArray<ModificationIterator> modificationIterator =
            new AtomicReferenceArray<ModificationIterator>(128);
-    // end of channel data

+    // end of channel data
    private final Set<Closeable> replicators = new CopyOnWriteArraySet<Closeable>();

    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<Replica>();
-        chronicleChannelIds = new ArrayList<Integer>();
+        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<Replica> chronicleChannelList() {
-        return Collections.unmodifiableList(chronicleChannelList);
+    public Map<Integer, Replica> 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();
@@ -35,31 +35,21 @@ public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
        MapBuilder<ClientWiredChronicleMapStatelessBuilder<K, V>> {

    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<K, V> implements

    private final AtomicBoolean used = new AtomicBoolean(false);

-
-    InetSocketAddress remoteAddress() {
-        return remoteAddress;
-    }
-
    @Override
    public ClientWiredChronicleMapStatelessBuilder<K, V> putReturnsNull(boolean putReturnsNull) {
        this.putReturnsNull = putReturnsNull;
@@ -100,10 +85,6 @@ public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
        return this;
    }

-    long timeoutMs() {
-        return timeoutMs;
-    }
-
    @Override
    public ClientWiredChronicleMapStatelessBuilder<K, V> name(String name) {
        this.name = name;
@@ -127,13 +108,9 @@ public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
    @Override
    public ChronicleMap<K, V> create() throws IOException {

-        // todo clean this up
-        if (hub == null)
-            hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking, "MAP");
-
        if (!used.getAndSet(true)) {
            return new ClientWiredStatelessChronicleMap<K, V>(
-                    this, keyClass, valueClass, channelID);
+                    this, keyClass, valueClass, name);

        } else {
            throw new IllegalStateException(
@@ -143,16 +120,5 @@ public final class ClientWiredChronicleMapStatelessBuilder<K, V> implements
        }
    }

-    public void identifier(byte identifier) {
-        this.localIdentifier = identifier;
-    }

-    public short channelID() {
-        return channelID;
-    }
-
-
-    public ClientWiredStatelessTcpConnectionHub hub() {
-        return hub;
-    }
 }
@@ -60,7 +60,6 @@ class ClientWiredStatelessChronicleMap<K, V>

    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<K, V>

    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();
@@ -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);

@@ -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<K, V> implements WireHandler, Consumer<WireHandlers> {

-
    public static final int SIZE_OF_SIZE = 2;
    private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class);
-    final Map<Long, Runnable> incompleteWork = new HashMap<Long, Runnable>();
+    public static final int MAP_SERVICE = 3;
+    private final Map<Long, Runnable> incompleteWork = new HashMap<Long, Runnable>();
    private final ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
-    @NotNull

+    @NotNull
    private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory;
+    private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<String, Integer>>> channelNameToIdFactory;
+
    private Wire inWire = null;
    private Wire outWire = null;

@@ -93,26 +99,44 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
            bytes1.write(entry.getValue());
        }
    };
+
    private WireHandlers publishLater;
    private byte localIdentifier;

-    private short channelId;
-    private List<Replica> channels;
+    private int channelId;
+    private Map<Integer, Replica> channels;
    private ReplicationHub hub;
    private byte remoteIdentifier;
+    private ChronicleMap<String, Integer> channelNameToId;

    public MapWireHandler(
-            @NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
-            @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List<Replica> channels) {
-        this(mapFactory, hub);
+            @NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
+            @NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<String, Integer>>>
+                    channelNameToIdFactory,
+            @NotNull final ReplicationHub hub, byte localIdentifier,
+            @NotNull final Map<Integer, Replica> channels) {
+        this(mapFactory, channelNameToIdFactory, hub);
        this.channels = channels;
        this.localIdentifier = localIdentifier;
+
    }

-    public MapWireHandler(Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
-                          ReplicationHub hub) {
+    public MapWireHandler(
+            @NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
+            @NotNull final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<String, Integer>>>
+                    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<K, V> implements WireHandler, Consumer<WireHandlers> {
        }
    }

-    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<K, V> implements WireHandler, Consumer<WireHandlers> {
     * @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<K, V> implements WireHandler, Consumer<WireHandlers> {
        if (map != null)
            return map;

+
        throw new IllegalStateException();
    }

@@ -663,7 +738,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
     * @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)
@@ -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 <K, V> WireHandler of(
            @NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
-            @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List<Replica> channels) {
-        return new MapWireHandler<>(mapFactory, hub, localIdentifier, channels);
+            @NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<String, Integer>>>
+                    channelNameToIdFactory,
+            @NotNull ReplicationHub hub,
+            byte localIdentifier,
+            @NotNull Map<Integer,Replica> 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,
@@ -72,14 +72,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon

    private ByteBuffer inBuffer;
    private ByteBufferBytes inBytes;
-
-    @NotNull
    private ReaderWithSize<K> keyReaderWithSize;
-    @NotNull
    private WriterWithSize<K> keyWriterWithSize;
-    @NotNull
    private ReaderWithSize<V> valueReaderWithSize;
-    @NotNull
    private WriterWithSize<V> valueWriterWithSize;
    private SocketChannel clientChannel;

@@ -98,15 +93,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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,