Refactor entry writing to replication buffer, #62

src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java

@@ -31,7 +31,6 @@ import java.io.IOException;
 import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
 import java.nio.channels.*;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
@@ -465,7 +464,7 @@ abstract class AbstractChannelReplicator implements Closeable {
 
         @Override
         public boolean onEntry(
-                @NotNull final Bytes entry, final int chronicleId, final long bootstrapTime) {
+                ReplicableEntry entry, Bytes payload, int chronicleId, long bootstrapTime) {
 
             long pos0 = entryIn.writePosition();
             // used to denote that this is not a heartbeat
@@ -478,7 +477,7 @@ abstract class AbstractChannelReplicator implements Closeable {
 
             long start = entryIn.writePosition();
 
-            externalizable.writeExternalEntry(entry, entryIn, chronicleId, bootstrapTime);
+            externalizable.writeExternalEntry(entry, payload, entryIn, chronicleId, bootstrapTime);
 
             if (entryIn.writePosition() == start) {
                 entryIn.writePosition(pos0);

src/main/java/net/openhft/chronicle/map/ChannelProvider.java

@@ -121,14 +121,14 @@ final class ChannelProvider implements Closeable {
          * @param chronicleChannel used in cluster into identify the canonical map or queue
          */
         @Override
-        public void writeExternalEntry(@NotNull Bytes entry,
+        public void writeExternalEntry(ReplicableEntry entry, Bytes payload,
                                        @NotNull Bytes destination,
                                        int chronicleChannel, long bootstrapTime) {
             channelDataReadLock();
             try {
                 destination.writeStopBit(chronicleChannel);
-                channelEntryExternalizables[chronicleChannel]
+                channelEntryExternalizables[chronicleChannel].writeExternalEntry(
-                        .writeExternalEntry(entry, destination, chronicleChannel, bootstrapTime);
+                        entry, payload, destination, chronicleChannel, bootstrapTime);
             } finally {
                 channelDataLock.readLock().unlock();
             }
@@ -481,7 +481,7 @@ final class ChannelProvider implements Closeable {
             @Override
             public ModificationIterator acquireModificationIterator(
                     final byte remoteIdentifier) {
-                final ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
+                ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
 
                 if (result != null)
                     return result;
@@ -498,10 +498,10 @@ final class ChannelProvider implements Closeable {
 
                     @Override
                     public boolean nextEntry(@NotNull EntryCallback callback, int na) {
-                        final Bytes bytes = payloads.poll();
+                        final Bytes payload = payloads.poll();
-                        if (bytes == null)
+                        if (payload == null)
                             return false;
-                        callback.onEntry(bytes, 0, System.currentTimeMillis());
+                        callback.onEntry(null, payload, 0, 0);
                         return true;
                     }
 
@@ -511,7 +511,8 @@ final class ChannelProvider implements Closeable {
                     }
 
                     @Override
-                    public void setModificationNotifier(@NotNull ModificationNotifier modificationNotifier) {
+                    public void setModificationNotifier(
+                            @NotNull ModificationNotifier modificationNotifier) {
                         modificationNotifier0 = modificationNotifier;
                     }
 
@@ -559,9 +560,10 @@ final class ChannelProvider implements Closeable {
             }
 
             @Override
-            public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
+            public void writeExternalEntry(
-                                           int na, long bootstrapTime) {
+                    ReplicableEntry entry, @NotNull Bytes payload, @NotNull Bytes destination,
-                destination.write(entry, entry.readPosition(), entry.readRemaining());
+                    int na, long bootstrapTime) {
+                destination.write(payload, payload.readPosition(), payload.readRemaining());
             }
 
             @Override

src/main/java/net/openhft/chronicle/map/Replica.java

@@ -156,7 +156,7 @@ public interface Replica extends Closeable {
          *                    match the maps local
          * @param chronicleId is the channel id used to identify the canonical map or queue
          */
-        void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
+        void writeExternalEntry(ReplicableEntry entry, Bytes payload, @NotNull Bytes destination,
                                 int chronicleId, long bootstrapTime);
 
         /**
@@ -189,17 +189,17 @@ public interface Replica extends Closeable {
          * in the ModificationIterator the entry may have been updated.
          */
         public abstract boolean onEntry(
-                final Bytes entry, final int chronicleId, long bootstrapTime);
+                ReplicableEntry entry, Bytes payload, int chronicleId, long bootstrapTime);
 
         /**
-         * Called just after {@link  #onEntry(Bytes, int, long)}.
+         * Called just after {@link  #onEntry(ReplicableEntry, Bytes, int, long)}.
          * No-op by default.
          */
         public void onAfterEntry() {
         }
 
         /**
-         * Called just before {@link #onEntry(Bytes, int, long)}.
+         * Called just before {@link #onEntry(ReplicableEntry, Bytes, int, long)}.
          * No-op by default.
          */
         public void onBeforeEntry() {

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -19,6 +19,7 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.algo.bitset.*;
 import net.openhft.chronicle.bytes.Bytes;
 import net.openhft.chronicle.core.Maths;
+import net.openhft.chronicle.hash.Data;
 import net.openhft.chronicle.hash.VanillaGlobalMutableState;
 import net.openhft.chronicle.hash.impl.TierCountersArea;
 import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
@@ -29,6 +30,7 @@ import net.openhft.chronicle.hash.replication.TimeProvider;
 import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext;
 import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext;
 import net.openhft.chronicle.map.replication.MapRemoteOperations;
+import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
 import net.openhft.chronicle.values.Values;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
@@ -401,62 +403,58 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
      */
     @Override
     public void writeExternalEntry(
-            @NotNull Bytes entry, @NotNull Bytes destination, int chronicleId, long bootstrapTime) {
+            ReplicableEntry entry, Bytes payload, @NotNull Bytes destination,
+            int chronicleId, long bootstrapTime) {
 
-        final long keySize = keySizeMarshaller.readSize(entry);
+        Data key;
-
+        Data value;
-        final long keyPosition = entry.readPosition();
+        boolean isDeleted;
-        entry.readSkip(keySize);
+        if (entry instanceof MapEntry) {
-        final long timeStamp = entry.readLong();
+            isDeleted = false;
-
+            MapEntry mapEntry = (MapEntry) entry;
-        final byte identifier = entry.readByte();
+            key = mapEntry.key();
-
+            value = mapEntry.value();
-        final boolean isDeleted = entry.readBoolean();
-        long valueSize;
-        if (!isDeleted) {
-            valueSize = valueSizeMarshaller.readSize(entry);
         } else {
-            valueSize = Math.max(0, valueSizeMarshaller.minStorableSize());
+            isDeleted = true;
+            MapAbsentEntry mapAbsentEntry = (MapAbsentEntry) entry;
+            key = mapAbsentEntry.absentKey();
+            value = ((MapRemoteQueryContext) mapAbsentEntry.context()).dummyZeroValue();
         }
 
-        final long valuePosition = entry.readPosition();
         destination.writeLong(bootstrapTime);
-        keySizeMarshaller.writeSize(destination, keySize);
+        keySizeMarshaller.writeSize(destination, key.size());
-        valueSizeMarshaller.writeSize(destination, valueSize);
+        valueSizeMarshaller.writeSize(destination, value.size());
-        destination.writeStopBit(timeStamp);
+        destination.writeStopBit(entry.originTimestamp());
 
-        if (identifier == 0)
+        if (entry.originIdentifier() == 0)
             throw new IllegalStateException("Identifier can't be 0");
-        destination.writeByte(identifier);
+        destination.writeByte(entry.originIdentifier());
         destination.writeBoolean(isDeleted);
 
-        // write the key
+        key.writeTo(destination, destination.writePosition());
-        destination.write(entry, keyPosition, keySize);
+        destination.writeSkip(key.size());
 
         boolean debugEnabled = LOG.isDebugEnabled();
         String message = null;
         if (debugEnabled) {
             if (isDeleted) {
                 LOG.debug("WRITING ENTRY TO DEST -  into local-id={}, remove(key={})",
-                        identifier(), entry.toString().trim());
+                        identifier(), key);
             } else {
                 message = String.format(
                         "WRITING ENTRY TO DEST  -  into local-id=%d, put(key=%s,",
-                        identifier(), entry.toString().trim());
+                        identifier(), key);
             }
         }
 
         if (isDeleted)
             return;
 
-        // skipping the alignment, as alignment wont work when we send the data over the wire.
+        value.writeTo(destination, destination.writePosition());
-        long valueAddr = entry.address(valuePosition);
+        destination.writeSkip(value.size());
-        long skip = alignAddr(valueAddr, alignment) - valueAddr;
-        // writes the value
-        destination.write(entry, valuePosition + skip, valueSize);
 
         if (debugEnabled) {
-            LOG.debug(message + "value=" + entry.toString().trim() + ")");
+            LOG.debug(message + "value=" + value + ")");
         }
     }
     
@@ -750,14 +748,11 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
                         final long segmentPos = position & posMask;
                         context.readExistingEntry(segmentPos);
 
-
                         // it may not be successful if the buffer can not be re-sized so we will
                         // process it later, by NOT clearing the changes.clear(position)
-                        context.segmentBytes()
-                                .readLimit(context.valueOffset() + context.valueSize());
-                        context.segmentBytes().readPosition(context.keySizeOffset());
                         boolean success = entryCallback.onEntry(
-                                context.segmentBytes(), chronicleId, bootStrapTimeStamp());
+                                (ReplicableEntry) context.entryForIteration(), null,
+                                chronicleId, bootStrapTimeStamp());
                         entryCallback.onAfterEntry();
 
                         if (success)

src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java

@@ -0,0 +1,58 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
+
+import static net.openhft.chronicle.map.Issue62ChronicleServer.MAP_FILE_B;
+import static net.openhft.chronicle.map.Issue62ChronicleServer.prepare;
+
+public class Issue62ChronicleClient {
+
+    private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleClient.class);
+
+    public static void main(String[] args) throws Exception {
+        prepare(MAP_FILE_B);
+
+        ChronicleMapBuilder<String, Long> cityPostalCodesMapBuilder =
+                ChronicleMapBuilder.of(String.class, Long.class)
+                        //.averageKeySize(100)
+                        .averageKey(Issue62ChronicleServer.STR)
+                        .replication((byte) 2, TcpTransportAndNetworkConfig
+                                .of(8077, new InetSocketAddress("localhost", 8076))
+                                .heartBeatInterval(1, TimeUnit.SECONDS))
+                        .entries(50_000);
+
+        try (ChronicleMap<String, Long> map =
+                cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_B)) {
+
+            LOGGER.info("Starting");
+            Thread.sleep(3000);
+
+            for (Entry<String, Long> entry : map.entrySet()) {
+                LOGGER.info("{} : {}", entry.getKey(), entry.getValue());
+            }
+        }
+    }
+
+}

src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java

@@ -0,0 +1,72 @@
+/*
+ *      Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *      This program is free software: you can redistribute it and/or modify
+ *      it under the terms of the GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License.
+ *
+ *      This program is distributed in the hope that it will be useful,
+ *      but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *      GNU Lesser General Public License for more details.
+ *
+ *      You should have received a copy of the GNU Lesser General Public License
+ *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
+
+public class Issue62ChronicleServer {
+
+    private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleServer.class);
+
+    final static String STR =
+            //"This is just a long string, which causes sink to fail for some reason.";
+            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+
+    private final static File MAP_FILE_A =
+            Paths.get(System.getProperty("java.io.tmpdir"), "map.a").toFile();
+    final static File MAP_FILE_B =
+            Paths.get(System.getProperty("java.io.tmpdir"), "map.b").toFile();
+
+    static void prepare(File file) {
+        if (file.exists())
+            file.delete();
+        file.deleteOnExit();
+    }
+
+    public static void main(String[] args) throws Exception {
+
+        prepare(MAP_FILE_A);
+
+        ChronicleMapBuilder<String, Long> cityPostalCodesMapBuilder =
+                ChronicleMapBuilder.of(String.class, Long.class)
+                        //.averageKeySize(100)
+                        .averageKey(STR)
+                        .replication((byte) 1, TcpTransportAndNetworkConfig
+                                .of(8076, new InetSocketAddress("localhost", 8077))
+                                .heartBeatInterval(1, TimeUnit.SECONDS))
+                        .entries(50_000);
+
+        ChronicleMap<String, Long> cityPostalCodes =
+                cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_A);
+
+        for(int i = 0; i < 100; i++ ) {
+            cityPostalCodes.put(STR + i, (long) i);
+        }
+
+        LOGGER.info("Map created");
+        Thread.sleep(15000);
+
+    }
+
+}