From e78e070558bcb5a1b0bf37bb4679287fb57c214e Mon Sep 17 00:00:00 2001 From: leventov Date: Thu, 17 Dec 2015 13:46:26 +0300 Subject: [PATCH] Refactor entry writing to replication buffer, #62 --- .../map/AbstractChannelReplicator.java | 5 +- .../chronicle/map/ChannelProvider.java | 24 ++++--- .../net/openhft/chronicle/map/Replica.java | 8 +-- .../chronicle/map/ReplicatedChronicleMap.java | 65 ++++++++--------- .../chronicle/map/Issue62ChronicleClient.java | 58 +++++++++++++++ .../chronicle/map/Issue62ChronicleServer.java | 72 +++++++++++++++++++ 6 files changed, 179 insertions(+), 53 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java create mode 100644 src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index b3d30c506..439706b44 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/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); diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 971549cd4..25eabf847 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/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] - .writeExternalEntry(entry, destination, chronicleChannel, bootstrapTime); + channelEntryExternalizables[chronicleChannel].writeExternalEntry( + 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(); - if (bytes == null) + final Bytes payload = payloads.poll(); + 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, - int na, long bootstrapTime) { - destination.write(entry, entry.readPosition(), entry.readRemaining()); + public void writeExternalEntry( + ReplicableEntry entry, @NotNull Bytes payload, @NotNull Bytes destination, + int na, long bootstrapTime) { + destination.write(payload, payload.readPosition(), payload.readRemaining()); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 7a9f24f38..1a8b0ccd9 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/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() { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index d27c5d9ba..7cb6b1aa8 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/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 extends VanillaChronicleMap extends VanillaChronicleMap. + */ + +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 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 map = + cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_B)) { + + LOGGER.info("Starting"); + Thread.sleep(3000); + + for (Entry entry : map.entrySet()) { + LOGGER.info("{} : {}", entry.getKey(), entry.getValue()); + } + } + } + +} diff --git a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java new file mode 100644 index 000000000..dc0dfa96c --- /dev/null +++ b/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 . + */ + +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 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 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); + + } + +}