From ef3e0bbe6902bcbe78019cd09bbec6e74db031f9 Mon Sep 17 00:00:00 2001 From: epickrram Date: Mon, 10 Jul 2017 10:12:26 +0100 Subject: [PATCH] allow Byteable instances that have maxSize() determined at runtime --- pom.xml | 2 +- .../impl/SerializationBuilder.java | 4 +- .../chronicle/map/ChronicleMapBuilder.java | 10 + .../chronicle/map/BasicReplicationTest.java | 216 ++++++++++++++++++ .../net/openhft/lang/values/DoubleArray.java | 11 +- .../openhft/lang/values/DoubleArrayTest.java | 2 +- 6 files changed, 236 insertions(+), 9 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java diff --git a/pom.xml b/pom.xml index 22bd5a585..9f64de04a 100755 --- a/pom.xml +++ b/pom.xml @@ -47,7 +47,7 @@ net.openhft chronicle-bom pom - 1.14.29-SNAPSHOT + 1.14.31-SNAPSHOT import diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java index 6bfd6c418..f379c507c 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/impl/SerializationBuilder.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.hash.serialization.impl; import net.openhft.chronicle.bytes.Byteable; import net.openhft.chronicle.bytes.BytesMarshallable; +import net.openhft.chronicle.bytes.DynamicallySized; import net.openhft.chronicle.core.OS; import net.openhft.chronicle.core.annotation.NotNull; import net.openhft.chronicle.core.util.ReadResolvable; @@ -190,8 +191,9 @@ public final class SerializationBuilder implements Cloneable { } public long constantSizeBySample(T sampleObject) { + // TODO mark.price this should probably be maxSize() long constantSize = serializationSize(sampleObject); - if (constantSizeMarshaller()) { + if (constantSizeMarshaller() && !DynamicallySized.class.isAssignableFrom(sampleObject.getClass())) { long expectedConstantSize = constantSize(); if (constantSize != expectedConstantSize) { throw new IllegalStateException("Although configuring constant size by sample " + diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index b847805f7..e476b19e2 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -820,6 +820,11 @@ public final class ChronicleMapBuilder implements if (maxChunksPerEntry < 1) throw new IllegalArgumentException("maxChunksPerEntry should be >= 1, " + maxChunksPerEntry + " given"); + if (constantlySizedEntries()) { + throw new IllegalStateException("Sizes of key type: " + keyBuilder.tClass + " and " + + "value type: " + valueBuilder.tClass + " are both constant, " + + "so maxChunksPerEntry shouldn't be specified manually"); + } this.maxChunksPerEntry = maxChunksPerEntry; return this; } @@ -1821,6 +1826,11 @@ public final class ChronicleMapBuilder implements keyBuilder, averageKeySize, averageKey, sampleKey, "Key"); averageValueSize = preMapConstruction( valueBuilder, averageValueSize, averageValue, sampleValue, "Value"); + + // TODO mark.price +// System.out.printf(Thread.currentThread().getName() + "|++!! averageValueSize calculated as %.2f%n", +// averageValueSize); + stateChecks(); } diff --git a/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java b/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java new file mode 100644 index 000000000..7e5b4b2c5 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java @@ -0,0 +1,216 @@ +package net.openhft.chronicle.map; + +import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext; +import org.hamcrest.CoreMatchers; +import org.junit.Ignore; +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; + +import static org.junit.Assert.assertThat; + +public class BasicReplicationTest { + + @Ignore + @Test + public void shouldReplicate() throws Exception { + final ChronicleMapBuilder builder = ChronicleMap.of(String.class, String.class) + .entries(1000).averageKeySize(7).averageValueSize(7); + + + try ( + ReplicatedChronicleMap mapOne = createReplicatedMap(builder, asByte(1)); + ReplicatedChronicleMap mapTwo = createReplicatedMap(builder, asByte(2)); + ReplicatedChronicleMap mapThree = createReplicatedMap(builder, asByte(3)); + ) { + + final ReplicationEventProcessor processorOne = + new ReplicationEventProcessor<>(); + processorOne.addDestinationMap(mapOne.acquireModificationIterator(asByte(2)), mapOne, mapTwo); + processorOne.addDestinationMap(mapOne.acquireModificationIterator(asByte(3)), mapOne, mapThree); + + final ReplicationEventProcessor processorTwo = + new ReplicationEventProcessor<>(); + processorTwo.addDestinationMap(mapTwo.acquireModificationIterator(asByte(1)), mapTwo, mapOne); + processorTwo.addDestinationMap(mapTwo.acquireModificationIterator(asByte(3)), mapTwo, mapThree); + + final ReplicationEventProcessor processorThree = + new ReplicationEventProcessor<>(); + processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(1)), mapThree, mapOne); + processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(2)), mapThree, mapTwo); + + final ExecutorService executorService = Executors.newFixedThreadPool(3); + executorService.submit(processorOne::processPendingChangesLoop); + executorService.submit(processorTwo::processPendingChangesLoop); + executorService.submit(processorTwo::processPendingChangesLoop); + + final Map[] maps = new Map[] {mapOne, mapTwo, mapThree}; + + final Random random = new Random(0xBAD5EED); + for (int i = 0; i < 50; i++) { + final int mapIndex = random.nextInt(maps.length); + final Map map = maps[mapIndex]; + final String key = "key" + random.nextInt(100); + final String value = "val" + random.nextInt(500); + map.put(key, value); + System.out.printf("map %d, put(%s, %s)%n", mapIndex, key, value); + } + + LockSupport.parkNanos(TimeUnit.DAYS.toNanos(5L)); + executorService.shutdownNow(); + + for(String key : mapOne.keySet()) { + final String mapOneValue = mapOne.get(key); + final String mapTwoValue = mapTwo.get(key); + final String mapThreeValue = mapThree.get(key); + + assertThat(mapOneValue, CoreMatchers.equalTo(mapTwoValue)); + assertThat(mapOneValue, CoreMatchers.equalTo(mapThreeValue)); + } + } + } + + private static byte asByte(final int i) { + return (byte) i; + } + + private static final class ReplicationEventProcessor { + private final List> destinationMaps = new ArrayList<>(); + private final AtomicBoolean queueEmpty = new AtomicBoolean(true); + private ReplicatedChronicleMap sourceMap; + + void addDestinationMap(final ReplicatedChronicleMap.ModificationIterator modificationIterator, + final ReplicatedChronicleMap sourceMap, + final ReplicatedChronicleMap destinationMap) { + destinationMaps.add(new IteratorAndDestinationMap(modificationIterator, sourceMap, destinationMap)); + if (this.sourceMap != null && this.sourceMap != sourceMap) { + throw new IllegalArgumentException("All iterators must belong to the same source map"); + } + this.sourceMap = sourceMap; + modificationIterator.setModificationNotifier(this::wakeup); + } + + void wakeup() { + queueEmpty.set(false); + } + + void processPendingChangesLoop() { + + while(!Thread.currentThread().isInterrupted()) { + while (queueEmpty.get()) { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(7)); + } + + Collections.shuffle(destinationMaps); + + for (IteratorAndDestinationMap iteratorAndDestinationMap : destinationMaps) { + while (iteratorAndDestinationMap.modificationIterator.hasNext()) { + iteratorAndDestinationMap.modificationIterator.nextEntry(iteratorAndDestinationMap, 17); + } + } + + queueEmpty.compareAndSet(false, true); + } + } + } + + private static final class IteratorAndDestinationMap implements Replica.ModificationIterator.Callback { + private final ReplicatedChronicleMap.ModificationIterator modificationIterator; + private final ReplicatedChronicleMap sourceMap; + private final ReplicatedChronicleMap destinationMap; + private final Bytes buffer = Bytes.elasticByteBuffer(); + + IteratorAndDestinationMap(final ReplicatedChronicleMap.ModificationIterator modificationIterator, + final ReplicatedChronicleMap sourceMap, + final ReplicatedChronicleMap destinationMap) { + this.modificationIterator = modificationIterator; + this.sourceMap = sourceMap; + this.destinationMap = destinationMap; + } + + @Override + public void onEntry(final ReplicableEntry entry, final int chronicleId) { + + try (CompiledReplicatedMapQueryContext iCtx = sourceMap.mapContext()) { + iCtx.updateLock().lock(); + try { + sourceMap.writeExternalEntry(entry, null, buffer, chronicleId); + } finally { + iCtx.updateLock().unlock(); + } + } catch(Throwable e) { + e.printStackTrace(); + } + final long writeLimit = buffer.writeLimit(); + + buffer.readPosition(0); + buffer.readLimit(writeLimit); + + destinationMap.readExternalEntry(buffer, sourceMap.identifier()); + } + + @Override + public void onBootstrapTime(final long bootstrapTime, final int chronicleId) { + + } + } + + + private ReplicatedChronicleMap + createReplicatedMap(final ChronicleMapBuilder builder, final byte replicaId) { + final ChronicleMapBuilderPrivateAPI privateBuilder = + new ChronicleMapBuilderPrivateAPI<>(builder); + privateBuilder.replication(replicaId); + final ChronicleMap map = builder.create(); + return (ReplicatedChronicleMap) map; + } + + private static final class ReplicationCallback implements Replica.ModificationIterator.Callback { + + private final ReplicatedChronicleMap sourceMap; + private final ReplicatedChronicleMap destinationMap; + + ReplicationCallback(final ReplicatedChronicleMap sourceMap, + final ReplicatedChronicleMap destinationMap) { + this.sourceMap = sourceMap; + this.destinationMap = destinationMap; + } + + @Override + public void onEntry(final ReplicableEntry entry, final int chronicleId) { + final boolean changed = entry.isChanged(); + final byte originIdentifier = entry.originIdentifier(); + final long originTimestamp = entry.originTimestamp(); + + System.out.printf("onEntry(%s, %d, %d)%n", changed, originIdentifier, originTimestamp); + + final boolean identifierCheck = sourceMap.identifierCheck(entry, chronicleId); + final Bytes destination = Bytes.elasticByteBuffer(); + sourceMap.writeExternalEntry(entry, null, destination, chronicleId); + System.out.println(identifierCheck); + System.out.println("destination: " + destination); + final long writeLimit = destination.writeLimit(); + destination.readPosition(0); + destination.readLimit(writeLimit); + + destinationMap.readExternalEntry(destination, sourceMap.identifier()); + } + + @Override + public void onBootstrapTime(final long bootstrapTime, final int chronicleId) { + System.out.printf("onBootstrapTime(%d, %d)%n", bootstrapTime, chronicleId); + } + } +} diff --git a/src/test/java/net/openhft/lang/values/DoubleArray.java b/src/test/java/net/openhft/lang/values/DoubleArray.java index bc665082e..07a8dd12e 100755 --- a/src/test/java/net/openhft/lang/values/DoubleArray.java +++ b/src/test/java/net/openhft/lang/values/DoubleArray.java @@ -19,29 +19,28 @@ package net.openhft.lang.values; import net.openhft.chronicle.bytes.Byteable; import net.openhft.chronicle.bytes.BytesStore; +import net.openhft.chronicle.bytes.DynamicallySized; import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.values.Copyable; /** * Created by peter.lawrey on 23/04/2015. */ -public class DoubleArray implements Byteable, Copyable { +public class DoubleArray implements Byteable, Copyable, DynamicallySized { static boolean HACK = true; private static int CAPACITY = 0; // assume a 32-bit size. private static int LENGTH = CAPACITY + 4; // assume a 32-bit size. private static int BASE = LENGTH + 4; + private final int capacity; private BytesStore bs; private long offset; - // creates an empty DoubleArray - public DoubleArray() { - } - public DoubleArray(int capacity) { bs = NativeBytesStore.nativeStoreWithFixedCapacity(BASE + capacity * 8L); bs.writeInt(CAPACITY, capacity); offset = 0; + this.capacity = capacity; } @Override @@ -62,7 +61,7 @@ public class DoubleArray implements Byteable, Copyable { @Override public long maxSize() { - return BASE + length() * 8; + return BASE + capacity * 8; } public int length() { diff --git a/src/test/java/net/openhft/lang/values/DoubleArrayTest.java b/src/test/java/net/openhft/lang/values/DoubleArrayTest.java index 5ae707eb9..4df000ddf 100755 --- a/src/test/java/net/openhft/lang/values/DoubleArrayTest.java +++ b/src/test/java/net/openhft/lang/values/DoubleArrayTest.java @@ -67,7 +67,6 @@ public class DoubleArrayTest { } @Test - @Ignore("TODO What is HACK???") public void addToAMap() { DoubleArray a = new DoubleArray(10); a.setData(new double[]{1, 2, 3, 4, 5}); @@ -77,6 +76,7 @@ public class DoubleArrayTest { ChronicleMap proxyMap = ChronicleMap .of(Integer.class, DoubleArray.class) + .constantValueSizeBySample(a) .entries(2) .create(); proxyMap.put(1, a);